From 636db089ae3b5e547a4bdfd22649c96f3e74aba6 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 29 Jul 2026 12:32:52 +0300 Subject: [PATCH] feat(telemetry): add bounded native pipeline lifecycle --- .../telemetry-plane/normalizer/normalizer.py | 31 +- .../Install-NdcMissionCoreTelegraf.ps1 | 24 +- .../Update-NdcMissionCorePerceptionRunner.ps1 | 58 +++- ...pdate-NdcMissionCorePersistentLauncher.ps1 | 67 +++++ .../Update-NdcMissionCoreTelegraf.ps1 | 26 +- .../telegraf/mission-core-windows.conf.tmpl | 15 +- .../worker/Invoke-E15PersistentShadowRun.ps1 | 13 +- .../worker/run_e15_shadow_inference.py | 277 +++++++++++++++++- src/k1link/compute/pipeline_telemetry.py | 146 ++++++++- src/k1link/web/system_telemetry_api.py | 7 + tests/test_e15_shadow_inference.py | 81 +++++ tests/test_pipeline_telemetry.py | 80 +++++ tests/test_system_telemetry_api.py | 18 ++ 13 files changed, 830 insertions(+), 13 deletions(-) create mode 100644 deploy/telemetry-plane/telegraf/Update-NdcMissionCorePersistentLauncher.ps1 diff --git a/deploy/telemetry-plane/normalizer/normalizer.py b/deploy/telemetry-plane/normalizer/normalizer.py index ab2c818..357aa3b 100644 --- a/deploy/telemetry-plane/normalizer/normalizer.py +++ b/deploy/telemetry-plane/normalizer/normalizer.py @@ -24,6 +24,8 @@ SOURCE_SCHEMAS: Final = { "heartbeat": "missioncore.agent-heartbeat/v1", } TELEGRAF_SCHEMA: Final = "telegraf.metric-json/v1" +PIPELINE_RECORD_SCHEMA: Final = "missioncore.pipeline-telemetry-record/v1" +PIPELINE_EVENT_MEASUREMENT: Final = "missioncore_pipeline_event" QUERY_SCHEMA: Final = "missioncore.telemetry-query/v1" SAFE_IDENTIFIER: Final = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$") MAX_PAYLOAD_BYTES: Final = 1024 * 1024 @@ -43,6 +45,7 @@ ALLOWED_TELEMETRY_TAGS: Final = frozenset( "cpu", "device", "engine_host", + "event_type", "gpu_name", "host", "interface", @@ -53,6 +56,7 @@ ALLOWED_TELEMETRY_TAGS: Final = frozenset( "path", "request_id", "run_id", + "run_state", "server_version", "source_id", "source_package_id", @@ -135,6 +139,30 @@ def _sanitize_tags(document: dict[str, Any]) -> dict[str, Any]: return {**document, "tags": sanitized} +def _unwrap_pipeline_event( + topic: str, + kind: str, + document: dict[str, Any], +) -> dict[str, Any]: + if kind != "pipeline" or document.get("name") != PIPELINE_EVENT_MEASUREMENT: + return document + fields = document.get("fields") + if not isinstance(fields, dict) or not isinstance(fields.get("value"), str): + raise ValueError("Telegraf pipeline event value is missing") + record = json.loads(fields["value"]) + if ( + not isinstance(record, dict) + or record.get("schema_version") != PIPELINE_RECORD_SCHEMA + or record.get("topic") != topic + or not isinstance(record.get("payload"), dict) + ): + raise ValueError("pipeline event record is invalid") + payload = record["payload"] + if payload.get("schema_version") != SOURCE_SCHEMAS["pipeline"]: + raise ValueError("pipeline event payload schema is invalid") + return payload + + def _normalize(topic: str, payload: bytes) -> tuple[object, ...]: if len(payload) > MAX_PAYLOAD_BYTES: raise ValueError("telemetry payload exceeds the 1 MiB contract") @@ -144,8 +172,9 @@ def _normalize(topic: str, payload: bytes) -> tuple[object, ...]: document = json.loads(payload.decode("utf-8")) if not isinstance(document, dict): raise ValueError("payload must be an object") - document = _sanitize_tags(document) kind = match.group("kind") + document = _unwrap_pipeline_event(topic, kind, document) + document = _sanitize_tags(document) source_schema = document.get("schema_version") if source_schema == SOURCE_SCHEMAS[kind]: observed_value = document.get("observed_at_utc") diff --git a/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 index 747c268..febd23b 100644 --- a/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 +++ b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 @@ -3,7 +3,8 @@ param( [string]$Version = "1.38.4", [string]$ExpectedSha256 = "6c7878ec319471ac85b82443baec2f3fa5dbcf1b6e2da5d5cd2cbb60fff2bb45", [string]$ConfigurationTemplate = "$PSScriptRoot\mission-core-windows.conf.tmpl", - [string]$PipelineCollector = "$PSScriptRoot\Get-NdcMissionCorePipelineTelemetry.ps1" + [string]$PipelineCollector = "$PSScriptRoot\Get-NdcMissionCorePipelineTelemetry.ps1", + [string]$PipelineJournal = "D:\NDC_MISSIONCORE\runtime\derived\.perception-persistent-publish\pipeline-telemetry.jsonl" ) $ErrorActionPreference = "Stop" @@ -41,6 +42,26 @@ if (-not (Test-Path -LiteralPath $ConfigurationTemplate -PathType Leaf)) { if (-not (Test-Path -LiteralPath $PipelineCollector -PathType Leaf)) { throw "Pipeline collector not found: $PipelineCollector" } +$journalParent = Get-Item -LiteralPath ( + Resolve-Path -LiteralPath (Split-Path $PipelineJournal -Parent) +).Path -Force +if ( + -not $journalParent.PSIsContainer -or + ($journalParent.Attributes -band [IO.FileAttributes]::ReparsePoint) -or + [IO.Path]::GetPathRoot($journalParent.FullName).TrimEnd("\") -ine "D:" +) { + throw "Pipeline journal parent must be a real D: directory" +} +if (-not (Test-Path -LiteralPath $PipelineJournal)) { + New-Item -ItemType File -Path $PipelineJournal | Out-Null +} +$journal = Get-Item -LiteralPath $PipelineJournal -Force +if ( + $journal.PSIsContainer -or + ($journal.Attributes -band [IO.FileAttributes]::ReparsePoint) +) { + throw "Pipeline journal must be a regular file" +} $temporaryRoot = Join-Path $env:TEMP "ndc-mission-core-telegraf-$([Guid]::NewGuid().ToString('N'))" $archivePath = Join-Path $temporaryRoot "telegraf.zip" @@ -121,6 +142,7 @@ try { StartType = $service.StartType.ToString() Configuration = $configurationPath PipelineCollector = $collectorPath + PipelineJournal = $PipelineJournal Sha256 = $actualSha256 } | ConvertTo-Json -Compress } diff --git a/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePerceptionRunner.ps1 b/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePerceptionRunner.ps1 index 3edf6bf..6800a72 100644 --- a/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePerceptionRunner.ps1 +++ b/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePerceptionRunner.ps1 @@ -6,14 +6,22 @@ param( [string]$ExpectedPredecessorSha256, [Parameter(Mandatory = $true)] [string]$ExpectedCandidateSha256, + [Parameter(Mandatory = $true)] + [string]$PipelineTelemetryCandidate, + [Parameter(Mandatory = $true)] + [string]$ExpectedPipelineTelemetrySha256, + [string]$ExpectedPipelineTelemetryPredecessorSha256 = "absent", [string]$ContainerName = "ndc-mission-core-perception-worker", [string]$RunnerRoot = "D:\NDC_MISSIONCORE\runtime\derived\e23-runner-20260724-002" ) $ErrorActionPreference = "Stop" $target = Join-Path $RunnerRoot "run_e15_shadow_inference.py" +$telemetryTarget = Join-Path $RunnerRoot "pipeline_telemetry.py" $expectedPredecessor = $ExpectedPredecessorSha256.ToLowerInvariant() $expectedCandidate = $ExpectedCandidateSha256.ToLowerInvariant() +$expectedTelemetry = $ExpectedPipelineTelemetrySha256.ToLowerInvariant() +$expectedTelemetryPredecessor = $ExpectedPipelineTelemetryPredecessorSha256.ToLowerInvariant() $predecessor = ( Get-FileHash -Algorithm SHA256 -LiteralPath $target ).Hash.ToLowerInvariant() @@ -26,12 +34,34 @@ if ( ) { throw "Runner candidate digest changed" } +$telemetryCandidateDigest = ( + Get-FileHash -Algorithm SHA256 -LiteralPath $PipelineTelemetryCandidate +).Hash.ToLowerInvariant() +if ($telemetryCandidateDigest -ne $expectedTelemetry) { + throw "Pipeline telemetry candidate digest changed" +} +$hadTelemetryModule = Test-Path -LiteralPath $telemetryTarget -PathType Leaf +$telemetryPredecessor = if ($hadTelemetryModule) { + (Get-FileHash -Algorithm SHA256 -LiteralPath $telemetryTarget).Hash.ToLowerInvariant() +} +else { + "absent" +} +if ($telemetryPredecessor -ne $expectedTelemetryPredecessor) { + throw "Pipeline telemetry predecessor digest changed" +} $backup = Join-Path $RunnerRoot ( "run_e15_shadow_inference.py.rollback-" + [DateTime]::UtcNow.ToString("yyyyMMddTHHmmssZ") + "-" + $predecessor.Substring(0, 12) ) +$telemetryBackup = Join-Path $RunnerRoot ( + "pipeline_telemetry.py.rollback-" + + [DateTime]::UtcNow.ToString("yyyyMMddTHHmmssZ") + + "-" + + $telemetryPredecessor.Substring(0, [Math]::Min(12, $telemetryPredecessor.Length)) +) $healthCode = "import urllib.request;print(urllib.request.urlopen('http://127.0.0.1:18020/health',timeout=2).read().decode())" function Wait-PerceptionHealth { @@ -52,7 +82,11 @@ function Wait-PerceptionHealth { if ( $health.ok -and $health.state -eq "ready" -and - (-not $RequireStageMetrics -or $health.stage_metrics) + (-not $RequireStageMetrics -or ( + $health.stage_metrics -and + $health.native_pipeline_telemetry -and + $health.native_pipeline_telemetry.ready + )) ) { return $health } @@ -63,7 +97,12 @@ function Wait-PerceptionHealth { } Copy-Item -LiteralPath $target -Destination $backup +if ($hadTelemetryModule) { + Copy-Item -LiteralPath $telemetryTarget -Destination $telemetryBackup +} try { + Copy-Item -LiteralPath $PipelineTelemetryCandidate ` + -Destination $telemetryTarget -Force Copy-Item -LiteralPath $Candidate -Destination $target -Force if ( (Get-FileHash -Algorithm SHA256 -LiteralPath $target).Hash.ToLowerInvariant() ` @@ -71,6 +110,12 @@ try { ) { throw "Runner replacement digest changed" } + if ( + (Get-FileHash -Algorithm SHA256 -LiteralPath $telemetryTarget).Hash.ToLowerInvariant() ` + -ne $expectedTelemetry + ) { + throw "Pipeline telemetry replacement digest changed" + } docker restart --time 20 $ContainerName | Out-Null if ($LASTEXITCODE -ne 0) { throw "Perception container restart failed" @@ -82,6 +127,12 @@ try { } catch { Copy-Item -LiteralPath $backup -Destination $target -Force + if ($hadTelemetryModule) { + Copy-Item -LiteralPath $telemetryBackup -Destination $telemetryTarget -Force + } + else { + Remove-Item -LiteralPath $telemetryTarget -Force -ErrorAction SilentlyContinue + } docker restart --time 20 $ContainerName | Out-Null $rollbackHealth = Wait-PerceptionHealth -RequireStageMetrics $false if (-not $rollbackHealth) { @@ -98,6 +149,11 @@ catch { CandidateSha256 = ( Get-FileHash -Algorithm SHA256 -LiteralPath $target ).Hash.ToLowerInvariant() + PipelineTelemetryPredecessorSha256 = $telemetryPredecessor + PipelineTelemetryCandidateSha256 = ( + Get-FileHash -Algorithm SHA256 -LiteralPath $telemetryTarget + ).Hash.ToLowerInvariant() + PipelineTelemetryBackup = if ($hadTelemetryModule) { $telemetryBackup } else { $null } Backup = $backup Health = $health } | ConvertTo-Json -Depth 12 -Compress diff --git a/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePersistentLauncher.ps1 b/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePersistentLauncher.ps1 new file mode 100644 index 0000000..94b8a03 --- /dev/null +++ b/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePersistentLauncher.ps1 @@ -0,0 +1,67 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$Candidate, + [Parameter(Mandatory = $true)] + [string]$ExpectedPredecessorSha256, + [Parameter(Mandatory = $true)] + [string]$ExpectedCandidateSha256, + [string]$RunnerRoot = "D:\NDC_MISSIONCORE\runtime\derived\e23-runner-20260724-002" +) + +$ErrorActionPreference = "Stop" +$target = Join-Path $RunnerRoot "Invoke-E15PersistentShadowRun.ps1" +$expectedPredecessor = $ExpectedPredecessorSha256.ToLowerInvariant() +$expectedCandidate = $ExpectedCandidateSha256.ToLowerInvariant() +$predecessor = ( + Get-FileHash -Algorithm SHA256 -LiteralPath $target +).Hash.ToLowerInvariant() +if ($predecessor -ne $expectedPredecessor) { + throw "Persistent launcher predecessor digest changed" +} +if ( + (Get-FileHash -Algorithm SHA256 -LiteralPath $Candidate).Hash.ToLowerInvariant() ` + -ne $expectedCandidate +) { + throw "Persistent launcher candidate digest changed" +} +$backup = Join-Path $RunnerRoot ( + "Invoke-E15PersistentShadowRun.ps1.rollback-" + + [DateTime]::UtcNow.ToString("yyyyMMddTHHmmssZ") + + "-" + + $predecessor.Substring(0, 12) +) + +Copy-Item -LiteralPath $target -Destination $backup +try { + Copy-Item -LiteralPath $Candidate -Destination $target -Force + $installed = ( + Get-FileHash -Algorithm SHA256 -LiteralPath $target + ).Hash.ToLowerInvariant() + if ($installed -ne $expectedCandidate) { + throw "Persistent launcher replacement digest changed" + } + $tokens = $null + $parseErrors = $null + [Management.Automation.Language.Parser]::ParseFile( + $target, + [ref]$tokens, + [ref]$parseErrors + ) | Out-Null + if (@($parseErrors).Count -ne 0) { + throw "Persistent launcher candidate has PowerShell parse errors" + } +} +catch { + Copy-Item -LiteralPath $backup -Destination $target -Force + throw +} + +[ordered]@{ + SchemaVersion = "missioncore.worker-persistent-launcher-deploy-result/v1" + PredecessorSha256 = $predecessor + CandidateSha256 = ( + Get-FileHash -Algorithm SHA256 -LiteralPath $target + ).Hash.ToLowerInvariant() + Backup = $backup +} | ConvertTo-Json -Compress diff --git a/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 b/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 index 34e4ab2..3788892 100644 --- a/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 +++ b/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 @@ -2,7 +2,8 @@ param( [Parameter(Mandatory = $true)] [string]$ConfigurationTemplate, - [string]$PipelineCollector = "$PSScriptRoot\Get-NdcMissionCorePipelineTelemetry.ps1" + [string]$PipelineCollector = "$PSScriptRoot\Get-NdcMissionCorePipelineTelemetry.ps1", + [string]$PipelineJournal = "D:\NDC_MISSIONCORE\runtime\derived\.perception-persistent-publish\pipeline-telemetry.jsonl" ) $ErrorActionPreference = "Stop" @@ -24,6 +25,28 @@ if (-not (Test-Path -LiteralPath $ConfigurationTemplate -PathType Leaf)) { if (-not (Test-Path -LiteralPath $PipelineCollector -PathType Leaf)) { throw "Pipeline collector not found: $PipelineCollector" } +$journalParent = Get-Item -LiteralPath ( + Resolve-Path -LiteralPath (Split-Path $PipelineJournal -Parent) +).Path -Force +if ( + -not $journalParent.PSIsContainer -or + ($journalParent.Attributes -band [IO.FileAttributes]::ReparsePoint) -or + [IO.Path]::GetPathRoot($journalParent.FullName).TrimEnd("\") -ine "D:" +) { + throw "Pipeline journal parent must be a real D: directory" +} +if (Test-Path -LiteralPath $PipelineJournal) { + $journal = Get-Item -LiteralPath $PipelineJournal -Force + if ( + $journal.PSIsContainer -or + ($journal.Attributes -band [IO.FileAttributes]::ReparsePoint) + ) { + throw "Pipeline journal must be a regular file" + } +} +else { + New-Item -ItemType File -Path $PipelineJournal | Out-Null +} $serviceEnvironmentBefore = @((Get-ItemProperty -Path $serviceRegistryPath).Environment) foreach ($entry in $serviceEnvironmentBefore) { @@ -116,6 +139,7 @@ try { Backup = $backupPath PipelineCollector = $collectorPath PipelineCollectorBackup = if ($hadCollector) { $backupCollectorPath } else { $null } + PipelineJournal = $PipelineJournal } | ConvertTo-Json -Compress } catch { diff --git a/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl b/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl index 1a76e3c..fab2e09 100644 --- a/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl +++ b/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl @@ -61,6 +61,17 @@ "collector_state", ] +[[inputs.tail]] + files = ["D:\\NDC_MISSIONCORE\\runtime\\derived\\.perception-persistent-publish\\pipeline-telemetry.jsonl"] + initial_read_offset = "saved-or-beginning" + watch_method = "poll" + max_undelivered_lines = 1000 + character_encoding = "utf-8" + path_tag = "" + name_override = "missioncore_pipeline_event" + data_format = "value" + data_type = "string" + [[outputs.mqtt]] servers = ["tcp://${MISSIONCORE_MQTT_HOST}:${MISSIONCORE_MQTT_PORT}"] topic = "mission-core/v1/contours/${MISSIONCORE_CONTOUR_ID}/agents/${MISSIONCORE_AGENT_ID}/host" @@ -69,7 +80,7 @@ qos = 1 data_format = "json" json_timestamp_units = "1s" - namedrop = ["missioncore_pipeline"] + namedrop = ["missioncore_pipeline", "missioncore_pipeline_event"] [[outputs.mqtt]] servers = ["tcp://${MISSIONCORE_MQTT_HOST}:${MISSIONCORE_MQTT_PORT}"] @@ -79,4 +90,4 @@ qos = 1 data_format = "json" json_timestamp_units = "1s" - namepass = ["missioncore_pipeline"] + namepass = ["missioncore_pipeline", "missioncore_pipeline_event"] diff --git a/experiments/perception/worker/Invoke-E15PersistentShadowRun.ps1 b/experiments/perception/worker/Invoke-E15PersistentShadowRun.ps1 index 45dd40c..a63be7a 100644 --- a/experiments/perception/worker/Invoke-E15PersistentShadowRun.ps1 +++ b/experiments/perception/worker/Invoke-E15PersistentShadowRun.ps1 @@ -7,7 +7,13 @@ param( [string]$PersistentOutputRoot = "D:\NDC_MISSIONCORE\runtime\derived\.perception-persistent-publish", [ValidateRange(1024, 65535)] [int]$PersistentPort = 18020, [ValidateRange(5, 3600)] [int]$MaximumDurationSeconds = 20, - [ValidateRange(1, 1000)] [int]$FreeGiBFloor = 360 + [ValidateRange(1, 1000)] [int]$FreeGiBFloor = 360, + [ValidatePattern("^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$")] + [string]$ContourId = "worker-006", + [ValidatePattern("^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$")] + [string]$AgentId = "worker-006", + [ValidatePattern("^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")] + [string]$NodeId = "DESKTOP-OPJ8J04" ) $ErrorActionPreference = "Stop" @@ -55,6 +61,11 @@ $request = @{ output_name = $outputName token = $token max_duration_seconds = $MaximumDurationSeconds + telemetry = @{ + contour_id = $ContourId + agent_id = $AgentId + node_id = $NodeId + } } | ConvertTo-Json -Compress $token = $null $client = ( diff --git a/experiments/perception/worker/run_e15_shadow_inference.py b/experiments/perception/worker/run_e15_shadow_inference.py index dc34afe..a1d41e9 100644 --- a/experiments/perception/worker/run_e15_shadow_inference.py +++ b/experiments/perception/worker/run_e15_shadow_inference.py @@ -22,7 +22,7 @@ import time from collections import Counter, deque from collections.abc import Iterator from contextlib import contextmanager, suppress -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -106,6 +106,21 @@ from k1link.compute.live_perception import ( from k1link.data_plane import DecodedPointCloudView, DecodedPoseView from k1link.device_plugins.xgrids_k1.protocol.normalizer import normalize_k1_message +try: + from k1link.compute.pipeline_telemetry import ( + JsonlPipelineTelemetrySink, + PipelineTelemetryEmitter, + PipelineTelemetryIdentity, + ) +except ModuleNotFoundError as exc: + if exc.name != "k1link.compute.pipeline_telemetry": + raise + from pipeline_telemetry import ( # type: ignore[no-redef] + JsonlPipelineTelemetrySink, + PipelineTelemetryEmitter, + PipelineTelemetryIdentity, + ) + PROFILE_SCHEMA = "missioncore.e15-shadow-inference-profile/v1" PROJECTION_SCHEMA = "missioncore.e15-live-projection-pack/v1" WORKER_PACKAGE_SCHEMA = "missioncore.e15-worker-package/v1" @@ -790,19 +805,71 @@ class _RuntimeTelemetry: return summary +class _FailureIsolatingPipelineTelemetrySink: + """Keep telemetry failures visible without changing compute control flow.""" + + def __init__(self, sink: Any) -> None: + self._sink = sink + self._lock = threading.Lock() + self._publish_failures = 0 + self._last_error_type: str | None = None + + def publish(self, topic: str, payload: bytes) -> None: + try: + self._sink.publish(topic, payload) + except Exception as exc: + with self._lock: + self._publish_failures += 1 + self._last_error_type = type(exc).__name__ + print( + json.dumps( + { + "event": "pipeline-telemetry-publish-failed", + "error_type": type(exc).__name__, + }, + sort_keys=True, + ), + flush=True, + ) + + def status(self) -> dict[str, Any]: + with self._lock: + return { + "transport": "jsonl-tail", + "ready": self._publish_failures == 0, + "publish_failures": self._publish_failures, + "last_error_type": self._last_error_type, + } + + class _StageExecutionTelemetry: """Measure named pipeline spans without pretending they are OS processes.""" - def __init__(self, stage_ids: tuple[str, ...] = PIPELINE_STAGE_IDS) -> None: + def __init__( + self, + stage_ids: tuple[str, ...] = PIPELINE_STAGE_IDS, + *, + identity: PipelineTelemetryIdentity | None = None, + sink: _FailureIsolatingPipelineTelemetrySink | None = None, + ) -> None: if not stage_ids or len(stage_ids) != len(set(stage_ids)): raise RuntimeError("pipeline stage identities are invalid") + if (identity is None) != (sink is None): + raise RuntimeError("pipeline telemetry identity and sink must be paired") self._stage_ids = stage_ids + self._identity = identity + self._sink = sink self._lock = threading.Lock() self._next_token = 0 self._active: dict[int, tuple[str, float, int | None]] = {} self._elapsed_seconds = dict.fromkeys(stage_ids, 0.0) self._activations = dict.fromkeys(stage_ids, 0) self._last_frame_index: int | None = None + self._first_frame_by_stage: dict[str, int | None] = {} + self._last_frame_by_stage: dict[str, int | None] = {} + self._native_started: set[str] = set() + self._native_failed: set[str] = set() + self._native_finalized = False @contextmanager def measure( @@ -813,21 +880,116 @@ class _StageExecutionTelemetry: if stage_id not in self._elapsed_seconds: raise RuntimeError(f"unknown pipeline stage: {stage_id}") started = time.perf_counter() + emit_started = False with self._lock: self._next_token += 1 token = self._next_token self._active[token] = (stage_id, started, frame_index) self._activations[stage_id] += 1 + self._last_frame_by_stage[stage_id] = frame_index + if stage_id not in self._first_frame_by_stage: + self._first_frame_by_stage[stage_id] = frame_index + if self.native_events_enabled and stage_id not in self._native_started: + self._native_started.add(stage_id) + emit_started = True if frame_index is not None: self._last_frame_index = frame_index + if emit_started: + self._emit_native( + stage_id=stage_id, + state="started", + frame_index=frame_index, + activation_count=1, + ) + failure: BaseException | None = None try: yield + except BaseException as exc: + failure = exc + raise finally: finished = time.perf_counter() + failure_event: tuple[float, int, int | None] | None = None with self._lock: active = self._active.pop(token, None) if active is not None: self._elapsed_seconds[stage_id] += max(0.0, finished - active[1]) + if ( + failure is not None + and self.native_events_enabled + and stage_id not in self._native_failed + ): + self._native_failed.add(stage_id) + failure_event = ( + self._elapsed_seconds[stage_id], + self._activations[stage_id], + self._last_frame_by_stage.get(stage_id), + ) + if failure_event is not None: + elapsed_seconds, activations, last_frame_index = failure_event + self._emit_native( + stage_id=stage_id, + state="failed", + frame_index=last_frame_index, + duration_ms=elapsed_seconds * 1000, + activation_count=activations, + error_type=type(failure).__name__, + ) + + @property + def native_events_enabled(self) -> bool: + return self._identity is not None and self._sink is not None + + def finalize_native(self) -> None: + """Emit one aggregate terminal event for each stage used by the run.""" + + rows: list[tuple[str, float, int, int | None]] = [] + with self._lock: + if not self.native_events_enabled or self._native_finalized: + return + self._native_finalized = True + rows = [ + ( + stage_id, + self._elapsed_seconds[stage_id], + self._activations[stage_id], + self._last_frame_by_stage.get(stage_id), + ) + for stage_id in self._stage_ids + if stage_id in self._native_started + and stage_id not in self._native_failed + ] + for stage_id, elapsed_seconds, activations, frame_index in rows: + self._emit_native( + stage_id=stage_id, + state="completed", + frame_index=frame_index, + duration_ms=elapsed_seconds * 1000, + activation_count=activations, + ) + + def _emit_native( + self, + *, + stage_id: str, + state: str, + frame_index: int | None, + activation_count: int, + duration_ms: float | None = None, + error_type: str | None = None, + ) -> None: + if self._identity is None or self._sink is None: + return + PipelineTelemetryEmitter( + identity=replace(self._identity, frame_index=frame_index), + sink=self._sink, + ).stage_event( + stage_id, + state, + duration_ms=duration_ms, + activation_count=activation_count, + error_type=error_type, + ) def snapshot(self) -> dict[str, Any]: now = time.perf_counter() @@ -2172,6 +2334,44 @@ def _persistent_run_arguments( return argparse.Namespace(**values) +def _persistent_run_telemetry_identity( + common: dict[str, Any], + request: dict[str, Any], +) -> PipelineTelemetryIdentity | None: + telemetry = request.get("telemetry") + if telemetry is None: + return None + if not isinstance(telemetry, dict) or set(telemetry) != { + "contour_id", + "agent_id", + "node_id", + }: + raise RuntimeError("persistent worker telemetry identity is invalid") + request_id = request.get("request_id") + if not isinstance(request_id, str): + raise RuntimeError("persistent worker telemetry request identity is invalid") + stability = common.get("stability") + lab_id = ( + stability["profile_id"] + if isinstance(stability, dict) and isinstance(stability.get("profile_id"), str) + else "lab-e15-shadow-inference-v1" + ) + method_id = ( + INLINE_TEMPORAL_PIPELINE_ID if isinstance(stability, dict) else PIPELINE_ID + ) + return PipelineTelemetryIdentity( + contour_id=telemetry.get("contour_id"), + agent_id=telemetry.get("agent_id"), + node_id=telemetry.get("node_id"), + lab_id=lab_id, + run_id=request_id, + request_id=request_id, + source_id=common["live"]["source"]["source_id"], + source_package_id=common["worker_package"]["package_id"], + method_id=method_id, + ) + + def serve(args: argparse.Namespace) -> int: from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -2183,6 +2383,9 @@ def serve(args: argparse.Namespace) -> int: raise RuntimeError("LAB E15 PyAV version changed") output_root = args.output_root.resolve(strict=True) _assert_disk(output_root, args.free_bytes_floor) + telemetry_sink = _FailureIsolatingPipelineTelemetrySink( + JsonlPipelineTelemetrySink(output_root / "pipeline-telemetry.jsonl") + ) load_started = time.perf_counter() loaded = _load_models(args, common) model_load_seconds = time.perf_counter() - load_started @@ -2193,6 +2396,7 @@ def serve(args: argparse.Namespace) -> int: "failed_runs": 0, "active_request_id": None, "active_frame_index": None, + "last_run_outcome": None, "_stage_telemetry": _StageExecutionTelemetry(), } @@ -2237,6 +2441,13 @@ def serve(args: argparse.Namespace) -> int: "current_stage": stage_snapshot["current_stage"], "active_stages": stage_snapshot["active_stages"], "stage_metrics": stage_snapshot["stages"], + "last_run_outcome": state["last_run_outcome"], + "native_pipeline_telemetry": { + **telemetry_sink.status(), + "active_run_events_enabled": ( + state["_stage_telemetry"].native_events_enabled + ), + }, "authority": common["live"]["authority"], "gpu": torch.cuda.get_device_name(), }, @@ -2259,16 +2470,51 @@ def serve(args: argparse.Namespace) -> int: return state["busy"] = True request_id = "invalid" + run_identity: PipelineTelemetryIdentity | None = None + run_emitter: PipelineTelemetryEmitter | None = None + run_started: float | None = None + terminal_event_emitted = False try: document = json.loads(self.rfile.read(length)) if not isinstance(document, dict): raise RuntimeError("persistent worker request is not an object") request_id = str(document.get("request_id", "invalid")) run_args = _persistent_run_arguments(args, document) + run_identity = _persistent_run_telemetry_identity(common, document) document["token"] = None state["active_request_id"] = request_id - state["_stage_telemetry"] = _StageExecutionTelemetry() + state["_stage_telemetry"] = _StageExecutionTelemetry( + identity=run_identity, + sink=telemetry_sink if run_identity is not None else None, + ) + run_emitter = ( + PipelineTelemetryEmitter( + identity=run_identity, + sink=telemetry_sink, + ) + if run_identity is not None + else None + ) + run_started = time.perf_counter() + if run_emitter is not None: + run_emitter.run("started") exit_code = run(run_args, loaded, state) + state["_stage_telemetry"].finalize_native() + duration_ms = max(0.0, (time.perf_counter() - run_started) * 1000) + if run_emitter is not None: + run_emitter.run( + "completed", + duration_ms=duration_ms, + exit_code=exit_code, + ) + terminal_event_emitted = True + state["last_run_outcome"] = { + "request_id": request_id, + "state": "completed", + "duration_ms": round(duration_ms, 6), + "exit_code": exit_code, + "error_type": None, + } state["completed_runs"] += 1 self._send( 200, @@ -2280,6 +2526,31 @@ def serve(args: argparse.Namespace) -> int: }, ) except Exception as exc: + state["_stage_telemetry"].finalize_native() + duration_ms = ( + max(0.0, (time.perf_counter() - run_started) * 1000) + if run_started is not None + else None + ) + if ( + run_emitter is not None + and duration_ms is not None + and not terminal_event_emitted + ): + run_emitter.run( + "failed", + duration_ms=duration_ms, + error_type=type(exc).__name__, + ) + state["last_run_outcome"] = { + "request_id": request_id, + "state": "failed", + "duration_ms": ( + round(duration_ms, 6) if duration_ms is not None else None + ), + "exit_code": None, + "error_type": type(exc).__name__, + } state["failed_runs"] += 1 print( json.dumps( diff --git a/src/k1link/compute/pipeline_telemetry.py b/src/k1link/compute/pipeline_telemetry.py index da8dd41..ed1d848 100644 --- a/src/k1link/compute/pipeline_telemetry.py +++ b/src/k1link/compute/pipeline_telemetry.py @@ -31,6 +31,7 @@ SAFE_TOPIC_IDENTIFIER: Final = re.compile( MAX_TEXT_LENGTH: Final = 256 MAX_PAYLOAD_BYTES: Final = 1024 * 1024 STAGE_STATES: Final = frozenset({"started", "completed", "failed"}) +RUN_STATES: Final = STAGE_STATES _AUTHORITY: Final = { "commands_enabled": False, "navigation_or_safety_accepted": False, @@ -162,6 +163,52 @@ class PipelineTelemetryEmitter: duration_ms=duration_ms, ) + def run( + self, + state: str, + *, + duration_ms: float | None = None, + exit_code: int | None = None, + error_type: str | None = None, + ) -> None: + """Publish one run-level lifecycle event with an explicit outcome.""" + + document = build_pipeline_run_telemetry_document( + identity=self.identity, + state=state, + duration_ms=duration_ms, + exit_code=exit_code, + error_type=error_type, + ) + self._publish(document) + + def stage_event( + self, + stage_id: str, + state: str, + *, + duration_ms: float | None = None, + activation_count: int = 1, + input_count: int | None = None, + output_count: int | None = None, + queue_wait_ms: float | None = None, + error_type: str | None = None, + ) -> None: + """Publish one explicit stage event, including a per-run aggregate.""" + + document = build_pipeline_telemetry_document( + identity=self.identity, + stage_id=stage_id, + state=state, + duration_ms=duration_ms, + activation_count=activation_count, + input_count=input_count, + output_count=output_count, + queue_wait_ms=queue_wait_ms, + error_type=error_type, + ) + self._publish(document) + def _emit( self, *, @@ -177,16 +224,18 @@ class PipelineTelemetryEmitter: outcome.queue_wait_ms, "queue_wait_ms", ) - document = build_pipeline_telemetry_document( - identity=self.identity, + self.stage_event( stage_id=stage_id, state=state, duration_ms=duration_ms, + activation_count=1, input_count=outcome.input_count, output_count=outcome.output_count, queue_wait_ms=outcome.queue_wait_ms, error_type=error_type, ) + + def _publish(self, document: dict[str, Any]) -> None: payload = _canonical_json(document) if len(payload) > MAX_PAYLOAD_BYTES: raise PipelineTelemetryError("pipeline telemetry exceeds the 1 MiB contract") @@ -245,6 +294,7 @@ def build_pipeline_telemetry_document( stage_id: str, state: str, duration_ms: float | None = None, + activation_count: int = 1, input_count: int | None = None, output_count: int | None = None, queue_wait_ms: float | None = None, @@ -257,6 +307,9 @@ def build_pipeline_telemetry_document( if state not in STAGE_STATES: raise PipelineTelemetryError("stage telemetry state is invalid") duration_ms = _optional_duration(duration_ms, "duration_ms") + activation_count = _optional_count(activation_count, "activation_count") + if activation_count in {None, 0}: + raise PipelineTelemetryError("activation_count must be positive") input_count = _optional_count(input_count, "input_count") output_count = _optional_count(output_count, "output_count") queue_wait_ms = _optional_duration(queue_wait_ms, "queue_wait_ms") @@ -289,7 +342,7 @@ def build_pipeline_telemetry_document( "elapsed_seconds": ( round(duration_ms / 1000.0, 9) if duration_ms is not None else None ), - "activations": 1, + "activations": activation_count, "input_count": input_count, "output_count": output_count, "queue_wait_ms": queue_wait_ms, @@ -298,6 +351,7 @@ def build_pipeline_telemetry_document( "stage_id": stage_id, "state": state, "duration_ms": duration_ms, + "activation_count": activation_count, "input_count": input_count, "output_count": output_count, "queue_wait_ms": queue_wait_ms, @@ -336,6 +390,92 @@ def build_pipeline_telemetry_document( return document +def build_pipeline_run_telemetry_document( + *, + identity: PipelineTelemetryIdentity, + state: str, + duration_ms: float | None = None, + exit_code: int | None = None, + error_type: str | None = None, + observed_at_utc: str | None = None, +) -> dict[str, Any]: + """Build a run-level start or terminal outcome document.""" + + if state not in RUN_STATES: + raise PipelineTelemetryError("run telemetry state is invalid") + duration_ms = _optional_duration(duration_ms, "duration_ms") + if state == "started": + if duration_ms is not None or exit_code is not None or error_type is not None: + raise PipelineTelemetryError("a started run cannot have an outcome") + elif duration_ms is None: + raise PipelineTelemetryError("a terminal run requires duration_ms") + if exit_code is not None and ( + isinstance(exit_code, bool) or not isinstance(exit_code, int) or exit_code < 0 + ): + raise PipelineTelemetryError("exit_code must be a non-negative integer") + if state == "completed" and exit_code is None: + raise PipelineTelemetryError("a completed run requires exit_code") + if error_type is not None: + _validate_text(error_type, "error_type") + if state == "failed" and error_type is None: + raise PipelineTelemetryError("a failed run requires error_type") + if state != "failed" and error_type is not None: + raise PipelineTelemetryError("only a failed run can have error_type") + + tags = { + "agent_id": identity.agent_id, + "contour_id": identity.contour_id, + "event_type": "run", + "lab_id": identity.lab_id, + "method_id": identity.method_id, + "node_id": identity.node_id, + "run_id": identity.run_id, + "run_state": state, + "source_id": identity.source_id, + "source_package_id": identity.source_package_id, + } + if identity.request_id is not None: + tags["request_id"] = identity.request_id + document: dict[str, Any] = { + "schema_version": PIPELINE_TELEMETRY_SCHEMA, + "observed_at_utc": observed_at_utc or _utc_now(), + "node_id": identity.node_id, + "lab_id": identity.lab_id, + "run_id": identity.run_id, + "source_id": identity.source_id, + "source_package_id": identity.source_package_id, + "method_id": identity.method_id, + "event_type": "run", + "run_state": state, + "tags": tags, + "payload": { + "state": ( + "busy" + if state == "started" + else ("failed" if state == "failed" else "ready") + ), + "current_stage": None, + "active_request_id": ( + identity.request_id or identity.run_id + if state == "started" + else None + ), + "active_stages": [], + "event": { + "event_type": "run", + "state": state, + "duration_ms": duration_ms, + "exit_code": exit_code, + "error_type": error_type, + }, + }, + "authority": _AUTHORITY, + } + if identity.request_id is not None: + document["request_id"] = identity.request_id + return document + + def _validate_text(value: object, name: str) -> str: if ( not isinstance(value, str) diff --git a/src/k1link/web/system_telemetry_api.py b/src/k1link/web/system_telemetry_api.py index e0db2d7..fd3ba40 100644 --- a/src/k1link/web/system_telemetry_api.py +++ b/src/k1link/web/system_telemetry_api.py @@ -75,6 +75,7 @@ PROMETHEUS_SAMPLE = re.compile( r"(?:\{[^}]*\})?\s+(?P[0-9.eE+-]+)$" ) TELEMETRY_QUERY_SCHEMA: Final = "missioncore.telemetry-query/v1" +PIPELINE_TELEMETRY_SCHEMA: Final = "missioncore.agent-pipeline-telemetry/v1" DEFAULT_TELEMETRY_QUERY_URL: Final = "http://127.0.0.1:18030" RootProvider = Callable[[], Path] @@ -510,6 +511,12 @@ def _agent_raw_document(document: dict[str, Any]) -> dict[str, Any]: if model_load_seconds is not None: perception["model_load_seconds"] = model_load_seconds perception["collector_state"] = fields.get("collector_state") + elif sample.get("source_schema") == PIPELINE_TELEMETRY_SCHEMA: + # Native lifecycle rows are immutable run evidence. The current + # service snapshot remains owned by the periodic health sample, + # so a terminal event cannot make a healthy worker look idle, + # failed, or busy after the fact. + continue else: pipeline_payload = _mapping(payload.get("payload")) or payload for name, value in pipeline_payload.items(): diff --git a/tests/test_e15_shadow_inference.py b/tests/test_e15_shadow_inference.py index 506ff0b..a66290e 100644 --- a/tests/test_e15_shadow_inference.py +++ b/tests/test_e15_shadow_inference.py @@ -371,6 +371,46 @@ def test_persistent_worker_request_is_single_run_d_backed_and_token_bounded( ) +def test_persistent_worker_derives_native_telemetry_from_accepted_runtime() -> None: + module = _module() + common = { + "live": {"source": {"source_id": "sensor.camera.right"}}, + "stability": {"profile_id": "lab-e23-warm-worker-inline-temporal-v1"}, + "worker_package": {"package_id": "e15-worker-package-example"}, + } + request = { + "request_id": "physical-k1-shadow-001", + "telemetry": { + "contour_id": "worker-006", + "agent_id": "worker-006", + "node_id": "DESKTOP-OPJ8J04", + }, + } + + identity = module._persistent_run_telemetry_identity(common, request) + + assert identity is not None + assert identity.lab_id == "lab-e23-warm-worker-inline-temporal-v1" + assert identity.run_id == "physical-k1-shadow-001" + assert identity.request_id == "physical-k1-shadow-001" + assert identity.source_id == "sensor.camera.right" + assert identity.source_package_id == "e15-worker-package-example" + assert identity.method_id == "warm-worker-inline-bounded-temporal-2d-3d-semantic/v1" + assert module._persistent_run_telemetry_identity( + common, + {"request_id": "legacy-request"}, + ) is None + + with pytest.raises(RuntimeError, match="telemetry identity"): + module._persistent_run_telemetry_identity( + common, + { + "request_id": "invalid-telemetry", + "telemetry": {"contour_id": "worker-006"}, + }, + ) + + def test_runtime_telemetry_captures_bounded_queue_snapshots() -> None: module = _module() stream = io.StringIO() @@ -422,3 +462,44 @@ def test_stage_execution_telemetry_measures_named_spans_without_process_claims() telemetry.measure("unregistered"), ): pass + + +def test_stage_execution_telemetry_writes_native_frame_lifecycle() -> None: + module = _module() + published: list[dict[str, object]] = [] + + class Sink: + def publish(self, _topic: str, payload: bytes) -> None: + published.append(json.loads(payload)) + + identity = module.PipelineTelemetryIdentity( + contour_id="worker-006", + agent_id="worker-006", + node_id="DESKTOP-OPJ8J04", + lab_id="lab-e23-warm-worker-inline-temporal-v1", + run_id="run-001", + request_id="run-001", + source_id="sensor.camera.right", + source_package_id="e15-worker-package-example", + method_id="warm-worker-inline-bounded-temporal-2d-3d-semantic/v1", + ) + telemetry = module._StageExecutionTelemetry( + ("detector",), + identity=identity, + sink=module._FailureIsolatingPipelineTelemetrySink(Sink()), + ) + + with telemetry.measure("detector", frame_index=42): + pass + with telemetry.measure("detector", frame_index=43): + pass + + assert telemetry.native_events_enabled is True + assert [row["stage_state"] for row in published] == ["started"] + telemetry.finalize_native() + assert [row["stage_state"] for row in published] == ["started", "completed"] + assert [row["frame_index"] for row in published] == [42, 43] + assert published[-1]["payload"]["event"]["activation_count"] == 2 + + telemetry.finalize_native() + assert len(published) == 2 diff --git a/tests/test_pipeline_telemetry.py b/tests/test_pipeline_telemetry.py index 1f4287c..1963dfc 100644 --- a/tests/test_pipeline_telemetry.py +++ b/tests/test_pipeline_telemetry.py @@ -13,6 +13,7 @@ from k1link.compute.pipeline_telemetry import ( PipelineTelemetryEmitter, PipelineTelemetryError, PipelineTelemetryIdentity, + build_pipeline_run_telemetry_document, build_pipeline_telemetry_document, ) @@ -59,6 +60,7 @@ def test_pipeline_document_is_accepted_without_losing_stage_identity() -> None: stage_id="predict", state="completed", duration_ms=125.5, + activation_count=7, input_count=89, output_count=89, queue_wait_ms=2.25, @@ -89,9 +91,51 @@ def test_pipeline_document_is_accepted_without_losing_stage_identity() -> None: } stored = json.loads(row[13]) assert stored["payload"]["event"]["duration_ms"] == 125.5 + assert stored["payload"]["event"]["activation_count"] == 7 + assert stored["payload"]["stage_metrics"]["predict"]["activations"] == 7 assert stored["authority"]["commands_enabled"] is False +def test_telegraf_tail_wrapper_restores_the_native_pipeline_document() -> None: + identity = _identity() + native = build_pipeline_telemetry_document( + identity=identity, + stage_id="tracking", + state="completed", + duration_ms=4.25, + observed_at_utc="2026-07-28T12:00:00Z", + ) + record = { + "schema_version": "missioncore.pipeline-telemetry-record/v1", + "topic": identity.topic, + "payload": native, + } + telegraf = { + "name": "missioncore_pipeline_event", + "timestamp": 1785240000, + "tags": { + "agent_id": identity.agent_id, + "contour_id": identity.contour_id, + "node_id": identity.node_id, + }, + "fields": { + "value": json.dumps(record, separators=(",", ":")), + }, + } + + row = _normalizer()._normalize(identity.topic, json.dumps(telegraf).encode()) + + assert row[5] == "pipeline" + assert row[7] == "missioncore.agent-pipeline-telemetry/v1" + assert row[9:13] == ("E41", "run-001", "request-001", 17) + assert json.loads(row[13]) == native + + record["topic"] = "mission-core/v1/contours/other/agents/other/pipeline" + telegraf["fields"]["value"] = json.dumps(record) + with pytest.raises(ValueError, match="pipeline event record"): + _normalizer()._normalize(identity.topic, json.dumps(telegraf).encode()) + + def test_stage_context_emits_terminal_event_and_preserves_failure() -> None: published: list[tuple[str, dict[str, object]]] = [] @@ -131,6 +175,42 @@ def test_stage_context_emits_terminal_event_and_preserves_failure() -> None: assert "source failure" not in json.dumps(published[-1][1]) +def test_run_events_record_explicit_terminal_outcome() -> None: + identity = _identity() + started = build_pipeline_run_telemetry_document( + identity=identity, + state="started", + observed_at_utc="2026-07-28T12:00:00Z", + ) + completed = build_pipeline_run_telemetry_document( + identity=identity, + state="completed", + duration_ms=12_345.5, + exit_code=2, + observed_at_utc="2026-07-28T12:00:12Z", + ) + + assert started["payload"]["state"] == "busy" + assert started["payload"]["active_request_id"] == "request-001" + assert completed["event_type"] == "run" + assert completed["run_state"] == "completed" + assert completed["payload"]["active_request_id"] is None + assert completed["payload"]["event"] == { + "event_type": "run", + "state": "completed", + "duration_ms": 12_345.5, + "exit_code": 2, + "error_type": None, + } + + with pytest.raises(PipelineTelemetryError, match="completed run"): + build_pipeline_run_telemetry_document( + identity=identity, + state="completed", + duration_ms=1.0, + ) + + def test_jsonl_sink_records_topic_bound_documents(tmp_path: Path) -> None: path = tmp_path / "telemetry" / "e41.jsonl" identity = _identity() diff --git a/tests/test_system_telemetry_api.py b/tests/test_system_telemetry_api.py index a8b24a2..82ac9a5 100644 --- a/tests/test_system_telemetry_api.py +++ b/tests/test_system_telemetry_api.py @@ -306,6 +306,24 @@ def test_agent_pipeline_snapshot_merges_all_stage_series() -> None: }, } ) + samples.append( + { + "node_id": EXPECTED_NODE_ID, + "kind": "pipeline", + "measurement": "pipeline", + "source_schema": "missioncore.agent-pipeline-telemetry/v1", + "observed_at_utc": "2026-07-28T12:00:01Z", + "payload": { + "schema_version": "missioncore.agent-pipeline-telemetry/v1", + "payload": { + "state": "failed", + "current_stage": None, + "active_request_id": None, + "active_stages": [], + }, + }, + } + ) raw = _agent_raw_document({"samples": samples})