feat(telemetry): expose compute pipeline stage metrics

2026-07-29 01:53:31 +03:00
parent 783aa444ac
commit 79eb4b46f7
19 changed files with 1000 additions and 24 deletions
@@ -57,6 +57,7 @@ function emptyDraft(): ComputeContourDraft {
    mqtt_host: "127.0.0.1",
    mqtt_port: 1883,
    telemetry_poll_interval_seconds: DEFAULT_TELEMETRY_POLL_INTERVAL_SECONDS,
+    mqtt_publish_interval_seconds: 2,
  };
 }

@@ -72,6 +73,7 @@ function draftFromContour(contour: ComputeContour | null): ComputeContourDraft {
    mqtt_host: contour.mqtt_host,
    mqtt_port: contour.mqtt_port,
    telemetry_poll_interval_seconds: contour.telemetry_poll_interval_seconds,
+    mqtt_publish_interval_seconds: contour.mqtt_publish_interval_seconds,
  };
 }

@@ -87,6 +89,9 @@ export function ComputeContourSettingsWindow({
  const [telemetryPollIntervalDraft, setTelemetryPollIntervalDraft] = useState(
    () => String(draftFromContour(contour).telemetry_poll_interval_seconds),
  );
+  const [mqttPublishIntervalDraft, setMqttPublishIntervalDraft] = useState(
+    () => String(draftFromContour(contour).mqtt_publish_interval_seconds),
+  );
  const [activeSection, setActiveSection] = useState<"connection" | "agent">("connection");
  const [install, setInstall] = useState<ComputeContourAgentInstall | null>(null);
  const [busy, setBusy] = useState(false);
@@ -97,6 +102,7 @@ export function ComputeContourSettingsWindow({
    const nextDraft = draftFromContour(mode === "edit" ? contour : null);
    setDraft(nextDraft);
    setTelemetryPollIntervalDraft(String(nextDraft.telemetry_poll_interval_seconds));
+    setMqttPublishIntervalDraft(String(nextDraft.mqtt_publish_interval_seconds));
    setActiveSection("connection");
    setInstall(null);
    setError(null);
@@ -121,6 +127,10 @@ export function ComputeContourSettingsWindow({
    () => parseTelemetryPollIntervalDraft(telemetryPollIntervalDraft),
    [telemetryPollIntervalDraft],
  );
+  const mqttPublishIntervalSeconds = useMemo(
+    () => parseTelemetryPollIntervalDraft(mqttPublishIntervalDraft),
+    [mqttPublishIntervalDraft],
+  );

  const valid = useMemo(() => (
    Boolean(draft.display_name.trim())
@@ -130,7 +140,8 @@ export function ComputeContourSettingsWindow({
    && Number.isInteger(draft.mqtt_port)
    && draft.mqtt_port > 0
    && telemetryPollIntervalSeconds !== null
-  ), [draft, telemetryPollIntervalSeconds]);
+    && mqttPublishIntervalSeconds !== null
+  ), [draft, mqttPublishIntervalSeconds, telemetryPollIntervalSeconds]);

  const commitTelemetryPollInterval = () => {
    const resolution = resolveTelemetryPollIntervalDraft(
@@ -149,11 +160,34 @@ export function ComputeContourSettingsWindow({
    ));
  };

+  const commitMqttPublishInterval = () => {
+    const resolution = resolveTelemetryPollIntervalDraft(
+      mqttPublishIntervalDraft,
+      draft.mqtt_publish_interval_seconds,
+    );
+    setMqttPublishIntervalDraft(resolution.draft);
+    if (!resolution.accepted) return;
+    setDraft((current) => (
+      current.mqtt_publish_interval_seconds === resolution.seconds
+        ? current
+        : {
+          ...current,
+          mqtt_publish_interval_seconds: resolution.seconds,
+        }
+    ));
+  };
+
  const save = async () => {
-    if (!valid || busy || telemetryPollIntervalSeconds === null) return;
+    if (
+      !valid
+      || busy
+      || telemetryPollIntervalSeconds === null
+      || mqttPublishIntervalSeconds === null
+    ) return;
    const committedDraft = {
      ...draft,
      telemetry_poll_interval_seconds: telemetryPollIntervalSeconds,
+      mqtt_publish_interval_seconds: mqttPublishIntervalSeconds,
    };
    setBusy(true);
    setError(null);
@@ -273,7 +307,7 @@ export function ComputeContourSettingsWindow({
              }))}
            />
            <TextField
-              label="Интервал MQTT"
+              label="Обновление интерфейса"
              hint="1–60 с"
              type="text"
              inputMode="numeric"
@@ -296,6 +330,30 @@ export function ComputeContourSettingsWindow({
                }
              }}
            />
+            <TextField
+              label="Интервал MQTT агента"
+              hint="1–60 с · применяется конфигурацией агента"
+              type="text"
+              inputMode="numeric"
+              autoComplete="off"
+              aria-invalid={mqttPublishIntervalSeconds === null}
+              value={mqttPublishIntervalDraft}
+              onChange={(event) => setMqttPublishIntervalDraft(event.currentTarget.value)}
+              onBlur={commitMqttPublishInterval}
+              onKeyDown={(event) => {
+                if (event.key === "Enter") {
+                  event.preventDefault();
+                  event.stopPropagation();
+                  commitMqttPublishInterval();
+                } else if (event.key === "Escape") {
+                  event.preventDefault();
+                  event.stopPropagation();
+                  setMqttPublishIntervalDraft(
+                    String(draft.mqtt_publish_interval_seconds),
+                  );
+                }
+              }}
+            />
            {draft.telemetry_mode === "legacy-ssh" ? (
              <TextField
                label="SSH port"
@@ -14,6 +14,7 @@ export interface ComputeContour {
  mqtt_host: string;
  mqtt_port: number;
  telemetry_poll_interval_seconds: number;
+  mqtt_publish_interval_seconds: number;
  revision: number;
  updated_at_utc: string | null;
 }
@@ -33,6 +34,7 @@ export interface ComputeContourDraft {
  mqtt_host: string;
  mqtt_port: number;
  telemetry_poll_interval_seconds: number;
+  mqtt_publish_interval_seconds: number;
 }

 export interface ComputeContourAgentInstall {
@@ -99,6 +101,7 @@ export async function createComputeContour(
      mqtt_host: draft.mqtt_host,
      mqtt_port: draft.mqtt_port,
      telemetry_poll_interval_seconds: draft.telemetry_poll_interval_seconds,
+      mqtt_publish_interval_seconds: draft.mqtt_publish_interval_seconds,
    }),
    signal,
  });
@@ -16,6 +16,7 @@ export interface WorkerTelemetryState {
 }

 export function useWorkerTelemetry(
+  contourId: string | null,
  pollMilliseconds = DEFAULT_WORKER_TELEMETRY_POLL_MILLISECONDS,
  enabled = true,
  externalRefreshGeneration = 0,
@@ -30,7 +31,7 @@ export function useWorkerTelemetry(
  const refresh = useCallback(() => setGeneration((value) => value + 1), []);

  useEffect(() => {
-    if (!enabled) {
+    if (!enabled || !contourId) {
      setTelemetry(null);
      setLoading(false);
      setError(null);
@@ -38,7 +39,7 @@ export function useWorkerTelemetry(
    }
    const controller = new AbortController();
    setLoading(true);
-    void fetchWorkerTelemetry(controller.signal)
+    void fetchWorkerTelemetry(contourId, controller.signal)
      .then((document) => {
        if (controller.signal.aborted) return;
        setTelemetry(document);
@@ -52,7 +53,7 @@ export function useWorkerTelemetry(
        if (!controller.signal.aborted) setLoading(false);
      });
    return () => controller.abort();
-  }, [enabled, externalRefreshGeneration, generation]);
+  }, [contourId, enabled, externalRefreshGeneration, generation]);

  useEffect(() => {
    if (!enabled || loading) return;
@@ -181,13 +181,17 @@ async function requestJson(
 }

 export async function fetchWorkerTelemetry(
+  contourId: string,
  signal?: AbortSignal,
 ): Promise<WorkerTelemetry> {
  const document = requireJsonRecord(
-    await requestJson("/api/v1/system/worker-telemetry?history=90", {
-      method: "GET",
-      signal,
-    }),
+    await requestJson(
+      `/api/v1/system/contours/${encodeURIComponent(contourId)}/telemetry?history=90`,
+      {
+        method: "GET",
+        signal,
+      },
+    ),
    "missioncore.worker-telemetry/v1",
  );
  if (
@@ -198,7 +202,7 @@ export async function fetchWorkerTelemetry(
    || !isRecord(document.network)
    || !Array.isArray(document.history)
  ) {
-    throw new Error("Срез Worker 006 неполон.");
+    throw new Error("Срез вычислительного контура неполон.");
  }
  return document as unknown as WorkerTelemetry;
 }
@@ -380,7 +380,10 @@
 .worker-stage-list {
  display: grid;
  min-width: 0;
-  grid-template-columns: repeat(3, minmax(0, 1fr));
+  grid-template-columns: repeat(
+    auto-fit,
+    minmax(min(100%, 24rem), 1fr)
+  );
  margin: 0;
  padding: 0;
  gap: 0.48rem;
@@ -64,10 +64,10 @@ export function ComputeModulesWorkspace() {
  } = useComputeContours();
  const legacyDiagnostic = selectedContour?.contour_id === "worker-006"
    && selectedContour.telemetry_mode === "legacy-ssh";
-  const supportsLiveTelemetry = selectedContour?.contour_id === "worker-006";
  const { telemetry, error } = useWorkerTelemetry(
+    selectedContour?.contour_id ?? null,
    (selectedContour?.telemetry_poll_interval_seconds ?? 3) * 1_000,
-    supportsLiveTelemetry,
+    selectedContour !== null,
    telemetryRefreshGeneration,
  );
  const node = telemetry?.node ?? null;
@@ -22,10 +22,10 @@ export function NetworkWorkspace() {
  } = useComputeContours();
  const legacyDiagnostic = selectedContour?.contour_id === "worker-006"
    && selectedContour.telemetry_mode === "legacy-ssh";
-  const supportsLiveTelemetry = selectedContour?.contour_id === "worker-006";
  const { telemetry, error } = useWorkerTelemetry(
+    selectedContour?.contour_id ?? null,
    (selectedContour?.telemetry_poll_interval_seconds ?? 3) * 1_000,
-    supportsLiveTelemetry,
+    selectedContour !== null,
    telemetryRefreshGeneration,
  );
  const aggregate = telemetry?.network.aggregate ?? null;
@@ -49,7 +49,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
  assert.match(workspaceHub, /<ComputeModulesWorkspace \/>/);
  assert.match(workspaceHub, /<NetworkWorkspace \/>/);
  assert.doesNotMatch(workspaceHub, /worker-telemetry|worker-profile|DESKTOP-OPJ8J04/);
-  assert.match(core, /\/api\/v1\/system\/worker-telemetry/);
+  assert.match(core, /\/api\/v1\/system\/contours\/.*\/telemetry/);
  assert.match(core, /\/api\/v1\/system\/worker-profile/);
  assert.match(core, /share_percent/);
  assert.doesNotMatch(core, /@nodedc\/ui-react/);
@@ -68,7 +68,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
  );
  assert.match(
    telemetryStyles,
-    /\.worker-stage-list\s*\{[^}]*grid-template-columns:\s*repeat\(3,/s,
+    /\.worker-stage-list\s*\{[^}]*grid-template-columns:\s*repeat\(\s*auto-fit,\s*minmax\(min\(100%,\s*24rem\),\s*1fr\)/s,
  );
  assert.doesNotMatch(
    telemetryStyles,
@@ -87,7 +87,8 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
  assert.match(contourSettingsHook, /icon: "refresh"/);
  assert.match(contourSettingsHook, /icon: "settings"/);
  assert.match(contourSettings, /FieldFrame label="Операционная система"/);
-  assert.match(contourSettings, /label="Интервал MQTT"/);
+  assert.match(contourSettings, /label="Обновление интерфейса"/);
+  assert.match(contourSettings, /label="Интервал MQTT агента"/);
  assert.match(contourSettings, /value=\{telemetryPollIntervalDraft\}/);
  assert.match(contourSettings, /onBlur=\{commitTelemetryPollInterval\}/);
  assert.match(contourSettings, /event\.key === "Escape"/);
@@ -96,6 +97,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () =>
    /telemetry_poll_interval_seconds:\s*Number\(event\.currentTarget\.value\)/,
  );
  assert.match(contourContract, /telemetry_poll_interval_seconds: number/);
+  assert.match(contourContract, /mqtt_publish_interval_seconds: number/);
  assert.match(pollIntervalContract, /MIN_TELEMETRY_POLL_INTERVAL_SECONDS\s*=\s*1/);
  assert.match(telemetryPolling, /normalizeWorkerTelemetryPollMilliseconds/);
  assert.match(styles, /system-telemetry\.css/);
@@ -97,5 +97,17 @@ configuration and rolls back if the service does not return to `Running`. MQTT
 credentials are scoped to the service environment and must not be passed on a command
 line or stored in the repository.

+The same host service reads the existing perception worker's loopback `/health`
+contract with `Get-NdcMissionCorePipelineTelemetry.ps1` and publishes nine
+stage-keyed snapshots to the contour's `pipeline` topic. The perception container
+does not receive broker credentials and no second agent container is introduced.
+These snapshots expose current durable-worker state and cumulative stage timing;
+native per-run lifecycle events remain a separate compute contract.
+
+When the mounted perception runner itself changes, use
+`Update-NdcMissionCorePerceptionRunner.ps1` with exact predecessor and candidate
+digests. It backs up the mounted runner, restarts the same container, accepts only a
+ready health document with stage metrics, and restores the predecessor on failure.
+
 The stack and agent are intentionally not started by repository tests. Provisioning a
 machine is a separate, explicit operation.
@@ -151,7 +151,7 @@ def _normalize(topic: str, payload: bytes) -> tuple[object, ...]:
        observed_value = document.get("observed_at_utc")
        measurement = kind
    elif (
-        kind == "host"
+        kind in {"host", "pipeline"}
        and isinstance(document.get("name"), str)
        and isinstance(document.get("fields"), dict)
    ):
@@ -160,6 +160,8 @@ def _normalize(topic: str, payload: bytes) -> tuple[object, ...]:
        measurement = document["name"].strip()
        if not measurement or len(measurement) > 128:
            raise ValueError("Telegraf measurement name is required")
+        if kind == "pipeline" and measurement != "missioncore_pipeline":
+            raise ValueError("Telegraf pipeline measurement is not recognized")
    else:
        raise ValueError("payload schema does not match topic kind")
    node_id = document.get("node_id") or _tag_text(document, "node_id")
@@ -0,0 +1,130 @@
+[CmdletBinding()]
+param(
+    [string]$ContainerName = "ndc-mission-core-perception-worker",
+    [string]$HealthUrl = "http://127.0.0.1:18020/health"
+)
+
+$ErrorActionPreference = "Stop"
+$stageIds = @(
+    "source-ingress",
+    "camera-decode",
+    "preprocessing",
+    "detector",
+    "semantic-model",
+    "sensor-fusion",
+    "tracking",
+    "temporal-state",
+    "result-publication"
+)
+$health = $null
+$collectorState = "unavailable"
+try {
+    $healthJson = docker exec $ContainerName python3 -c `
+        "import urllib.request;print(urllib.request.urlopen('$HealthUrl',timeout=2).read().decode())" `
+        2>$null
+    if ($LASTEXITCODE -eq 0 -and $healthJson) {
+        $health = $healthJson | ConvertFrom-Json
+        $collectorState = "live"
+    }
+}
+catch {
+    $health = $null
+}
+
+$serviceState = if ($health -and $health.state) {
+    [string]$health.state
+}
+else {
+    "unavailable"
+}
+$currentStage = if ($health -and $health.current_stage) {
+    [string]$health.current_stage
+}
+else {
+    ""
+}
+$activeStages = if ($health -and $health.active_stages) {
+    @($health.active_stages | ForEach-Object { [string]$_ })
+}
+else {
+    @()
+}
+$activeRequestId = if ($health -and $health.active_request_id) {
+    [string]$health.active_request_id
+}
+else {
+    ""
+}
+$activeFrameIndex = if (
+    $health -and
+    $null -ne $health.active_frame_index
+) {
+    [int64]$health.active_frame_index
+}
+else {
+    [int64]-1
+}
+$completedRuns = if ($health -and $null -ne $health.completed_runs) {
+    [int64]$health.completed_runs
+}
+else {
+    [int64]0
+}
+$failedRuns = if ($health -and $null -ne $health.failed_runs) {
+    [int64]$health.failed_runs
+}
+else {
+    [int64]0
+}
+$modelLoadSeconds = if ($health -and $null -ne $health.model_load_seconds) {
+    [double]$health.model_load_seconds
+}
+else {
+    [double]0
+}
+
+$samples = foreach ($stageId in $stageIds) {
+    $stageState = if ($collectorState -ne "live") {
+        "unavailable"
+    }
+    elseif ($currentStage -eq $stageId -or $activeStages -contains $stageId) {
+        "active"
+    }
+    elseif ($serviceState -eq "busy") {
+        "waiting"
+    }
+    else {
+        "ready"
+    }
+    $sample = [ordered]@{
+        stage_id = $stageId
+        stage_state = $stageState
+        service_state = $serviceState
+        current_stage = $currentStage
+        active_request_id = $activeRequestId
+        active_frame_index = $activeFrameIndex
+        completed_runs = $completedRuns
+        failed_runs = $failedRuns
+        model_load_seconds = $modelLoadSeconds
+        collector_state = $collectorState
+        health_ok = [bool]($health -and $health.ok)
+    }
+    if ($health -and $health.stage_metrics) {
+        $property = $health.stage_metrics.PSObject.Properties[$stageId]
+        if ($property) {
+            $metric = $property.Value
+            if ($null -ne $metric.elapsed_seconds) {
+                $sample.elapsed_seconds = [double]$metric.elapsed_seconds
+            }
+            if ($null -ne $metric.activations) {
+                $sample.activations = [int64]$metric.activations
+            }
+            if ($null -ne $metric.share_percent) {
+                $sample.share_percent = [double]$metric.share_percent
+            }
+        }
+    }
+    [pscustomobject]$sample
+}
+
+@($samples) | ConvertTo-Json -Compress -Depth 8
@@ -2,7 +2,8 @@
 param(
    [string]$Version = "1.38.4",
    [string]$ExpectedSha256 = "6c7878ec319471ac85b82443baec2f3fa5dbcf1b6e2da5d5cd2cbb60fff2bb45",
-    [string]$ConfigurationTemplate = "$PSScriptRoot\mission-core-windows.conf.tmpl"
+    [string]$ConfigurationTemplate = "$PSScriptRoot\mission-core-windows.conf.tmpl",
+    [string]$PipelineCollector = "$PSScriptRoot\Get-NdcMissionCorePipelineTelemetry.ps1"
 )

 $ErrorActionPreference = "Stop"
@@ -10,6 +11,7 @@ $serviceName = "telegraf"
 $installRoot = "C:\Program Files\NDC\Mission Core\Telegraf"
 $configurationRoot = "C:\ProgramData\NDC\MissionCore\telemetry-agent"
 $configurationPath = Join-Path $configurationRoot "telegraf.conf"
+$collectorPath = Join-Path $configurationRoot "Get-NdcMissionCorePipelineTelemetry.ps1"
 $archiveUrl = "https://dl.influxdata.com/telegraf/releases/telegraf-$($Version)_windows_amd64.zip"
 $payload = [Console]::In.ReadToEnd() | ConvertFrom-Json

@@ -36,6 +38,9 @@ if (Get-Service -Name $serviceName -ErrorAction SilentlyContinue) {
 if (-not (Test-Path -LiteralPath $ConfigurationTemplate -PathType Leaf)) {
    throw "Configuration template not found: $ConfigurationTemplate"
 }
+if (-not (Test-Path -LiteralPath $PipelineCollector -PathType Leaf)) {
+    throw "Pipeline collector not found: $PipelineCollector"
+}

 $temporaryRoot = Join-Path $env:TEMP "ndc-mission-core-telegraf-$([Guid]::NewGuid().ToString('N'))"
 $archivePath = Join-Path $temporaryRoot "telegraf.zip"
@@ -57,6 +62,7 @@ try {
    New-Item -ItemType Directory -Path $installRoot, $configurationRoot -Force | Out-Null
    Copy-Item -LiteralPath $sourceExecutable.FullName -Destination (Join-Path $installRoot "telegraf.exe")
    Copy-Item -LiteralPath $ConfigurationTemplate -Destination $configurationPath
+    Copy-Item -LiteralPath $PipelineCollector -Destination $collectorPath
    & icacls.exe $configurationRoot /inheritance:r /grant:r `
        "*S-1-5-18:(OI)(CI)F" "*S-1-5-32-544:(OI)(CI)F" | Out-Null
    if ($LASTEXITCODE -ne 0) {
@@ -114,6 +120,7 @@ try {
        Status = $service.Status.ToString()
        StartType = $service.StartType.ToString()
        Configuration = $configurationPath
+        PipelineCollector = $collectorPath
        Sha256 = $actualSha256
    } | ConvertTo-Json -Compress
 }
@@ -0,0 +1,103 @@
+[CmdletBinding()]
+param(
+    [Parameter(Mandatory = $true)]
+    [string]$Candidate,
+    [Parameter(Mandatory = $true)]
+    [string]$ExpectedPredecessorSha256,
+    [Parameter(Mandatory = $true)]
+    [string]$ExpectedCandidateSha256,
+    [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"
+$expectedPredecessor = $ExpectedPredecessorSha256.ToLowerInvariant()
+$expectedCandidate = $ExpectedCandidateSha256.ToLowerInvariant()
+$predecessor = (
+    Get-FileHash -Algorithm SHA256 -LiteralPath $target
+).Hash.ToLowerInvariant()
+if ($predecessor -ne $expectedPredecessor) {
+    throw "Runner predecessor digest changed"
+}
+if (
+    (Get-FileHash -Algorithm SHA256 -LiteralPath $Candidate).Hash.ToLowerInvariant() `
+        -ne $expectedCandidate
+) {
+    throw "Runner candidate digest changed"
+}
+$backup = Join-Path $RunnerRoot (
+    "run_e15_shadow_inference.py.rollback-" +
+    [DateTime]::UtcNow.ToString("yyyyMMddTHHmmssZ") +
+    "-" +
+    $predecessor.Substring(0, 12)
+)
+$healthCode = "import urllib.request;print(urllib.request.urlopen('http://127.0.0.1:18020/health',timeout=2).read().decode())"
+
+function Wait-PerceptionHealth {
+    param(
+        [bool]$RequireStageMetrics,
+        [int]$TimeoutSeconds = 150
+    )
+
+    $deadline = [DateTime]::UtcNow.AddSeconds($TimeoutSeconds)
+    while ([DateTime]::UtcNow -lt $deadline) {
+        Start-Sleep -Seconds 3
+        try {
+            $healthJson = docker exec $ContainerName python3 -c $healthCode 2>$null
+            if ($LASTEXITCODE -ne 0 -or -not $healthJson) {
+                continue
+            }
+            $health = $healthJson | ConvertFrom-Json
+            if (
+                $health.ok -and
+                $health.state -eq "ready" -and
+                (-not $RequireStageMetrics -or $health.stage_metrics)
+            ) {
+                return $health
+            }
+        }
+        catch {}
+    }
+    return $null
+}
+
+Copy-Item -LiteralPath $target -Destination $backup
+try {
+    Copy-Item -LiteralPath $Candidate -Destination $target -Force
+    if (
+        (Get-FileHash -Algorithm SHA256 -LiteralPath $target).Hash.ToLowerInvariant() `
+            -ne $expectedCandidate
+    ) {
+        throw "Runner replacement digest changed"
+    }
+    docker restart --time 20 $ContainerName | Out-Null
+    if ($LASTEXITCODE -ne 0) {
+        throw "Perception container restart failed"
+    }
+    $health = Wait-PerceptionHealth -RequireStageMetrics $true
+    if (-not $health) {
+        throw "Stage-instrumented runner did not become ready"
+    }
+}
+catch {
+    Copy-Item -LiteralPath $backup -Destination $target -Force
+    docker restart --time 20 $ContainerName | Out-Null
+    $rollbackHealth = Wait-PerceptionHealth -RequireStageMetrics $false
+    if (-not $rollbackHealth) {
+        throw "Candidate failed and predecessor rollback did not become ready"
+    }
+    throw
+}
+
+[ordered]@{
+    SchemaVersion = "missioncore.worker-runner-deploy-result/v1"
+    Container = $ContainerName
+    State = (docker inspect --format "{{.State.Status}}" $ContainerName)
+    PredecessorSha256 = $predecessor
+    CandidateSha256 = (
+        Get-FileHash -Algorithm SHA256 -LiteralPath $target
+    ).Hash.ToLowerInvariant()
+    Backup = $backup
+    Health = $health
+} | ConvertTo-Json -Depth 12 -Compress
@@ -1,7 +1,8 @@
 [CmdletBinding()]
 param(
    [Parameter(Mandatory = $true)]
-    [string]$ConfigurationTemplate
+    [string]$ConfigurationTemplate,
+    [string]$PipelineCollector = "$PSScriptRoot\Get-NdcMissionCorePipelineTelemetry.ps1"
 )

 $ErrorActionPreference = "Stop"
@@ -9,6 +10,7 @@ $serviceName = "telegraf"
 $installRoot = "C:\Program Files\NDC\Mission Core\Telegraf"
 $configurationRoot = "C:\ProgramData\NDC\MissionCore\telemetry-agent"
 $configurationPath = Join-Path $configurationRoot "telegraf.conf"
+$collectorPath = Join-Path $configurationRoot "Get-NdcMissionCorePipelineTelemetry.ps1"
 $executable = Join-Path $installRoot "telegraf.exe"
 $serviceRegistryPath = "HKLM:\SYSTEM\CurrentControlSet\Services\$serviceName"

@@ -19,8 +21,12 @@ if (-not (Test-Path -LiteralPath $executable -PathType Leaf)) {
 if (-not (Test-Path -LiteralPath $ConfigurationTemplate -PathType Leaf)) {
    throw "Configuration template not found: $ConfigurationTemplate"
 }
+if (-not (Test-Path -LiteralPath $PipelineCollector -PathType Leaf)) {
+    throw "Pipeline collector not found: $PipelineCollector"
+}

-foreach ($entry in @((Get-ItemProperty -Path $serviceRegistryPath).Environment)) {
+$serviceEnvironmentBefore = @((Get-ItemProperty -Path $serviceRegistryPath).Environment)
+foreach ($entry in $serviceEnvironmentBefore) {
    $name, $value = $entry -split "=", 2
    if ($name -and $value) {
        Set-Item -Path "Env:$name" -Value $value
@@ -29,14 +35,38 @@ foreach ($entry in @((Get-ItemProperty -Path $serviceRegistryPath).Environment))
 if (-not $env:MISSIONCORE_TELEMETRY_INTERVAL) {
    $env:MISSIONCORE_TELEMETRY_INTERVAL = "2s"
 }
+$serviceEnvironmentCandidate = @($serviceEnvironmentBefore)
+if (-not ($serviceEnvironmentCandidate | Where-Object {
+    $_ -like "MISSIONCORE_TELEMETRY_INTERVAL=*"
+})) {
+    $serviceEnvironmentCandidate += (
+        "MISSIONCORE_TELEMETRY_INTERVAL=$($env:MISSIONCORE_TELEMETRY_INTERVAL)"
+    )
+}

 $temporaryRoot = Join-Path $env:TEMP "ndc-mission-core-telegraf-update-$([Guid]::NewGuid().ToString('N'))"
 $validationOutput = Join-Path $temporaryRoot "validation.out.log"
 $validationError = Join-Path $temporaryRoot "validation.error.log"
 $backupRoot = Join-Path $configurationRoot "backups"
 $backupPath = Join-Path $backupRoot "telegraf-$([DateTime]::UtcNow.ToString('yyyyMMddTHHmmssZ')).conf"
+$backupCollectorPath = Join-Path $backupRoot `
+    "pipeline-collector-$([DateTime]::UtcNow.ToString('yyyyMMddTHHmmssZ')).ps1"
+$hadCollector = Test-Path -LiteralPath $collectorPath -PathType Leaf
 try {
    New-Item -ItemType Directory -Path $temporaryRoot, $backupRoot -Force | Out-Null
+    $collectorProbe = & powershell.exe -NoLogo -NoProfile -NonInteractive `
+        -ExecutionPolicy Bypass -File $PipelineCollector
+    if ($LASTEXITCODE -ne 0) {
+        throw "Pipeline collector validation failed"
+    }
+    $collectorDocument = $collectorProbe | ConvertFrom-Json
+    if (@($collectorDocument).Count -ne 9) {
+        throw "Pipeline collector must publish exactly nine stage samples"
+    }
+    if ($hadCollector) {
+        Copy-Item -LiteralPath $collectorPath -Destination $backupCollectorPath
+    }
+    Copy-Item -LiteralPath $PipelineCollector -Destination $collectorPath
    $validation = Start-Process -FilePath $executable `
        -ArgumentList @("--config", $ConfigurationTemplate, "--test") `
        -NoNewWindow -Wait -PassThru `
@@ -49,7 +79,14 @@ try {
    }

    Copy-Item -LiteralPath $configurationPath -Destination $backupPath
+    Set-ItemProperty -Path $serviceRegistryPath -Name Environment `
+        -Type MultiString -Value $serviceEnvironmentCandidate
    Stop-Service -Name $serviceName
+    $service = Get-Service -Name $serviceName
+    $service.WaitForStatus(
+        [ServiceProcess.ServiceControllerStatus]::Stopped,
+        [TimeSpan]::FromSeconds(20)
+    )
    try {
        Copy-Item -LiteralPath $ConfigurationTemplate -Destination $configurationPath
        Start-Service -Name $serviceName
@@ -61,6 +98,12 @@ try {
    }
    catch {
        Copy-Item -LiteralPath $backupPath -Destination $configurationPath
+        if ($hadCollector) {
+            Copy-Item -LiteralPath $backupCollectorPath -Destination $collectorPath
+        }
+        else {
+            Remove-Item -LiteralPath $collectorPath -Force -ErrorAction SilentlyContinue
+        }
        Start-Service -Name $serviceName
        throw
    }
@@ -71,8 +114,21 @@ try {
        Status = (Get-Service -Name $serviceName).Status.ToString()
        Configuration = $configurationPath
        Backup = $backupPath
+        PipelineCollector = $collectorPath
+        PipelineCollectorBackup = if ($hadCollector) { $backupCollectorPath } else { $null }
    } | ConvertTo-Json -Compress
 }
+catch {
+    Set-ItemProperty -Path $serviceRegistryPath -Name Environment `
+        -Type MultiString -Value $serviceEnvironmentBefore
+    if ($hadCollector -and (Test-Path -LiteralPath $backupCollectorPath -PathType Leaf)) {
+        Copy-Item -LiteralPath $backupCollectorPath -Destination $collectorPath
+    }
+    elseif (-not $hadCollector) {
+        Remove-Item -LiteralPath $collectorPath -Force -ErrorAction SilentlyContinue
+    }
+    throw
+}
 finally {
    Remove-Item -LiteralPath $temporaryRoot -Recurse -Force -ErrorAction SilentlyContinue
 }
@@ -47,6 +47,20 @@
    "nv_inference_count",
  ]

+[[inputs.exec]]
+  commands = ['powershell.exe -NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass -File "C:\ProgramData\NDC\MissionCore\telemetry-agent\Get-NdcMissionCorePipelineTelemetry.ps1"']
+  timeout = "4s"
+  name_override = "missioncore_pipeline"
+  data_format = "json"
+  tag_keys = ["stage_id"]
+  json_string_fields = [
+    "stage_state",
+    "service_state",
+    "current_stage",
+    "active_request_id",
+    "collector_state",
+  ]
+
 [[outputs.mqtt]]
  servers = ["tcp://${MISSIONCORE_MQTT_HOST}:${MISSIONCORE_MQTT_PORT}"]
  topic = "mission-core/v1/contours/${MISSIONCORE_CONTOUR_ID}/agents/${MISSIONCORE_AGENT_ID}/host"
@@ -55,3 +69,14 @@
  qos = 1
  data_format = "json"
  json_timestamp_units = "1s"
+  namedrop = ["missioncore_pipeline"]
+
+[[outputs.mqtt]]
+  servers = ["tcp://${MISSIONCORE_MQTT_HOST}:${MISSIONCORE_MQTT_PORT}"]
+  topic = "mission-core/v1/contours/${MISSIONCORE_CONTOUR_ID}/agents/${MISSIONCORE_AGENT_ID}/pipeline"
+  username = "${MISSIONCORE_MQTT_USERNAME}"
+  password = "${MISSIONCORE_MQTT_PASSWORD}"
+  qos = 1
+  data_format = "json"
+  json_timestamp_units = "1s"
+  namepass = ["missioncore_pipeline"]
@@ -0,0 +1,434 @@
+--- run_e15_shadow_inference.py
+++ run_e15_shadow_inference.py
+@@ -20,7 +20,8 @@
+ import threading
+ import time
+ from collections import Counter, deque
+-from contextlib import suppress
+from collections.abc import Iterator
+from contextlib import contextmanager, suppress
+ from dataclasses import dataclass
+ from datetime import UTC, datetime
+ from pathlib import Path
+@@ -108,6 +109,17 @@
+ WORLD_SCHEMA = "missioncore.live-perception-world-state/v1"
+ SEMANTIC_SCHEMA = "missioncore.e15-shadow-semantic-frame/v1"
+ PIPELINE_ID = "shadow-fmp4-yolox-eomt-kb4-amodal-world-state/v1"
+PIPELINE_STAGE_IDS = (
+    "source-ingress",
+    "camera-decode",
+    "preprocessing",
+    "detector",
+    "semantic-model",
+    "sensor-fusion",
+    "tracking",
+    "temporal-state",
+    "result-publication",
+)
+
+
+ def arguments() -> argparse.Namespace:
+@@ -539,6 +551,76 @@
+             summary["rss_growth_mib"] = 0.0
+             summary["final_queues"] = {}
+         return summary
+
+
+class _StageExecutionTelemetry:
+    """Measure named pipeline spans without pretending they are OS processes."""
+
+    def __init__(self, stage_ids: tuple[str, ...] = PIPELINE_STAGE_IDS) -> None:
+        if not stage_ids or len(stage_ids) != len(set(stage_ids)):
+            raise RuntimeError("pipeline stage identities are invalid")
+        self._stage_ids = stage_ids
+        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
+
+    @contextmanager
+    def measure(
+        self,
+        stage_id: str,
+        frame_index: int | None = None,
+    ) -> Iterator[None]:
+        if stage_id not in self._elapsed_seconds:
+            raise RuntimeError(f"unknown pipeline stage: {stage_id}")
+        started = time.perf_counter()
+        with self._lock:
+            self._next_token += 1
+            token = self._next_token
+            self._active[token] = (stage_id, started, frame_index)
+            self._activations[stage_id] += 1
+            if frame_index is not None:
+                self._last_frame_index = frame_index
+        try:
+            yield
+        finally:
+            finished = time.perf_counter()
+            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])
+
+    def snapshot(self) -> dict[str, Any]:
+        now = time.perf_counter()
+        with self._lock:
+            elapsed = dict(self._elapsed_seconds)
+            active_rows = list(self._active.values())
+            for stage_id, started, _frame_index in active_rows:
+                elapsed[stage_id] += max(0.0, now - started)
+            total = sum(elapsed.values())
+            active_stages = list(
+                dict.fromkeys(stage_id for stage_id, _started, _frame in active_rows)
+            )
+            current_stage = active_rows[-1][0] if active_rows else None
+            return {
+                "current_stage": current_stage,
+                "active_stages": active_stages,
+                "active_frame_index": self._last_frame_index,
+                "stages": {
+                    stage_id: {
+                        "elapsed_seconds": round(elapsed[stage_id], 6),
+                        "activations": self._activations[stage_id],
+                        "share_percent": (
+                            round(elapsed[stage_id] / total * 100, 6)
+                            if total > 0
+                            else None
+                        ),
+                    }
+                    for stage_id in self._stage_ids
+                },
+            }
+
+
+ def _send_client_binary_frame(stream: Any, payload: bytes) -> None:
+@@ -571,6 +653,7 @@
+     sensor_decode_ms: dict[str, list[float]],
+     result_queue: queue.Queue[bytes],
+     result_complete: threading.Event,
+    stage_telemetry: _StageExecutionTelemetry,
+ ) -> None:
+     import select
+
+@@ -632,7 +715,8 @@
+                 continue
+             if opcode != 0x2:
+                 raise ShadowRuntimeError(f"unexpected websocket opcode: {opcode}")
+-            header, payload = _decode_event(frame)
+            with stage_telemetry.measure("source-ingress"):
+                header, payload = _decode_event(frame)
+             sequence = int(header["ingress_sequence"])
+             if state.last_ingress_sequence is not None:
+                 if sequence <= state.last_ingress_sequence:
+@@ -883,7 +967,11 @@
+     )
+
+
+-def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int:
+def run(
+    args: argparse.Namespace,
+    loaded: _LoadedModels | None = None,
+    runtime_state: dict[str, Any] | None = None,
+) -> int:
+     import av
+     import torch
+     import transformers
+@@ -894,6 +982,15 @@
+     token = sys.stdin.readline().strip() if args.token_stdin else args.token
+     if not token or len(token) < 40:
+         raise RuntimeError("LAB E15 shadow token is missing")
+    stage_telemetry = (
+        runtime_state.get("_stage_telemetry")
+        if runtime_state is not None
+        else None
+    )
+    if not isinstance(stage_telemetry, _StageExecutionTelemetry):
+        stage_telemetry = _StageExecutionTelemetry()
+        if runtime_state is not None:
+            runtime_state["_stage_telemetry"] = stage_telemetry
+     common = _common(args) if loaded is None else loaded.common
+     live = common["live"]
+     e14 = common["e14"]
+@@ -982,6 +1079,10 @@
+         height=600,
+         maximum_buffer_bytes=int(live["transport"]["maximum_media_buffer_bytes"]),
+         metadata_capacity=int(live["transport"]["camera_metadata_capacity"]),
+        measure_decode=lambda frame_index: stage_telemetry.measure(
+            "camera-decode",
+            frame_index,
+        ),
+     )
+     transport = _TransportState(Counter(), Counter())
+     result_queue: queue.Queue[bytes] = queue.Queue(maxsize=2)
+@@ -1086,6 +1187,11 @@
+                 self.count += 1
+
+         completed_semantics = SemanticResultStream()
+
+        def monitored_semantic_inference(image: Any) -> Any:
+            with stage_telemetry.measure("semantic-model"):
+                return infer_semantic(image)
+
+         semantic_thread = threading.Thread(
+             target=semantic_worker,
+             kwargs={
+@@ -1094,7 +1200,7 @@
+                 "valid_mask": valid_mask,
+                 "target_lut": target_lut,
+                 "target_names": target_names,
+-                "infer": infer_semantic,
+                "infer": monitored_semantic_inference,
+                 "latency": semantic_latency,
+                 "completed": completed_semantics,
+                 "failures": semantic_errors,
+@@ -1125,6 +1231,7 @@
+                 "sensor_decode_ms": sensor_decode_ms,
+                 "result_queue": result_queue,
+                 "result_complete": result_complete,
+                "stage_telemetry": stage_telemetry,
+             },
+             name="lab-e15-shadow-receiver",
+             daemon=True,
+@@ -1157,10 +1264,21 @@
+             )
+             try:
+                 detector_started = time.perf_counter()
+-                tensor = _preprocess(envelope.image, valid_mask, detector)
+-                output_tensor, _request_ms = _infer(args.triton_url, detector["model"], tensor)
+-                detections, _rejected = _detections(output_tensor, detector, valid_mask)
+-                tracks = tracker.update(detections, envelope.frame_index)
+                with stage_telemetry.measure("preprocessing", envelope.frame_index):
+                    tensor = _preprocess(envelope.image, valid_mask, detector)
+                with stage_telemetry.measure("detector", envelope.frame_index):
+                    output_tensor, _request_ms = _infer(
+                        args.triton_url,
+                        detector["model"],
+                        tensor,
+                    )
+                    detections, _rejected = _detections(
+                        output_tensor,
+                        detector,
+                        valid_mask,
+                    )
+                with stage_telemetry.measure("tracking", envelope.frame_index):
+                    tracks = tracker.update(detections, envelope.frame_index)
+                 latency["detector_ms"].append(
+                     (time.perf_counter() - detector_started) * 1000
+                 )
+@@ -1194,34 +1312,35 @@
+                     ).reshape((-1, 3))
+                     position = binding.pose.position_xyz
+                     quaternion = binding.pose.orientation_xyzw
+-                    projection_started = time.perf_counter()
+-                    pixels, depths, source_indices, points_lidar = project_points(
+-                        points_map,
+-                        position,
+-                        quaternion,
+-                        projection,
+-                    )
+-                    latency["projection_ms"].append(
+-                        (time.perf_counter() - projection_started) * 1000
+-                    )
+-                    association_started = time.perf_counter()
+-                    fusions = fuse_tracks(
+-                        tracks=[_track_document(track) for track in tracks],
+-                        semantic_map=current_semantic.mask,
+-                        pixels=pixels,
+-                        depths=depths,
+-                        source_indices=source_indices,
+-                        points_map=points_map,
+-                        points_lidar=points_lidar,
+-                        association=e14["association"],
+-                        distance_history=history,
+-                        completion_tracker=completion_tracker,
+-                        sensor_position_map=position,
+-                        session_seconds=frame_seconds,
+-                    )
+-                    latency["association_ms"].append(
+-                        (time.perf_counter() - association_started) * 1000
+-                    )
+                    with stage_telemetry.measure("sensor-fusion", envelope.frame_index):
+                        projection_started = time.perf_counter()
+                        pixels, depths, source_indices, points_lidar = project_points(
+                            points_map,
+                            position,
+                            quaternion,
+                            projection,
+                        )
+                        latency["projection_ms"].append(
+                            (time.perf_counter() - projection_started) * 1000
+                        )
+                        association_started = time.perf_counter()
+                        fusions = fuse_tracks(
+                            tracks=[_track_document(track) for track in tracks],
+                            semantic_map=current_semantic.mask,
+                            pixels=pixels,
+                            depths=depths,
+                            source_indices=source_indices,
+                            points_map=points_map,
+                            points_lidar=points_lidar,
+                            association=e14["association"],
+                            distance_history=history,
+                            completion_tracker=completion_tracker,
+                            sensor_position_map=position,
+                            session_seconds=frame_seconds,
+                        )
+                        latency["association_ms"].append(
+                            (time.perf_counter() - association_started) * 1000
+                        )
+                     fusion_state = "fused"
+                     fused_frames += 1
+                 fusion_state_counts[fusion_state] += 1
+@@ -1270,20 +1389,21 @@
+                 if temporal_stabilizer is None:
+                     fusion_objects = raw_fusion_objects
+                 else:
+-                    temporal_started = time.perf_counter()
+-                    fusion_objects = temporal_stabilizer.update(
+-                        frame_index=envelope.frame_index,
+-                        session_seconds=frame_seconds,
+-                        objects=raw_fusion_objects,
+-                    )
+-                    world = stabilize_world_state(
+-                        world,
+-                        fusion_objects,
+-                        temporal_world_memory,
+-                    )
+-                    latency["temporal_2d_3d_ms"].append(
+-                        (time.perf_counter() - temporal_started) * 1000
+-                    )
+                    with stage_telemetry.measure("temporal-state", envelope.frame_index):
+                        temporal_started = time.perf_counter()
+                        fusion_objects = temporal_stabilizer.update(
+                            frame_index=envelope.frame_index,
+                            session_seconds=frame_seconds,
+                            objects=raw_fusion_objects,
+                        )
+                        world = stabilize_world_state(
+                            world,
+                            fusion_objects,
+                            temporal_world_memory,
+                        )
+                        latency["temporal_2d_3d_ms"].append(
+                            (time.perf_counter() - temporal_started) * 1000
+                        )
+                 stabilized_cuboids += sum(
+                     str(item.get("cuboid_status", "")).startswith("accepted-")
+                     for item in fusion_objects
+@@ -1351,28 +1471,29 @@
+                     quality=80,
+                     optimize=False,
+                 )
+-                live_result = encode_live_perception_result(
+-                    frame_index=envelope.frame_index,
+-                    source_frame_index=int(envelope.timeline["source_frame_index"]),
+-                    session_seconds=frame_seconds,
+-                    captured_at_epoch_ns=int(envelope.timeline["captured_at_epoch_ns"]),
+-                    image_jpeg=encoded_image.getvalue(),
+-                    segmentation_mask=(
+-                        current_semantic.mask
+-                        if current_semantic is not None and semantic_status == "fresh"
+-                        else None
+-                    ),
+-                    objects=fusion_objects,
+-                    delivery=world["delivery"],
+-                )
+-                try:
+-                    result_queue.put_nowait(live_result)
+-                except queue.Full:
+-                    with suppress(queue.Empty):
+-                        result_queue.get_nowait()
+-                        result_queue.task_done()
+-                        transport.results_dropped += 1
+-                    result_queue.put_nowait(live_result)
+                with stage_telemetry.measure("result-publication", envelope.frame_index):
+                    live_result = encode_live_perception_result(
+                        frame_index=envelope.frame_index,
+                        source_frame_index=int(envelope.timeline["source_frame_index"]),
+                        session_seconds=frame_seconds,
+                        captured_at_epoch_ns=int(envelope.timeline["captured_at_epoch_ns"]),
+                        image_jpeg=encoded_image.getvalue(),
+                        segmentation_mask=(
+                            current_semantic.mask
+                            if current_semantic is not None and semantic_status == "fresh"
+                            else None
+                        ),
+                        objects=fusion_objects,
+                        delivery=world["delivery"],
+                    )
+                    try:
+                        result_queue.put_nowait(live_result)
+                    except queue.Full:
+                        with suppress(queue.Empty):
+                            result_queue.get_nowait()
+                            result_queue.task_done()
+                            transport.results_dropped += 1
+                        result_queue.put_nowait(live_result)
+             except Exception:
+                 detector_failures += 1
+                 raise
+@@ -1645,6 +1766,7 @@
+         },
+         "gpu_telemetry": gpu.summary(),
+         "runtime_telemetry": runtime_summary,
+        "stage_telemetry": stage_telemetry.snapshot(),
+         "process_cpu_seconds": time.process_time() - process_cpu_started,
+         "process_peak_rss_mib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024,
+         "cuda_peak_memory_allocated_mib": torch.cuda.max_memory_allocated() / 2**20,
+@@ -1816,7 +1938,13 @@
+     loaded = _load_models(args, common)
+     model_load_seconds = time.perf_counter() - load_started
+     run_lock = threading.Lock()
+-    state = {"busy": False, "completed_runs": 0, "failed_runs": 0}
+    state = {
+        "busy": False,
+        "completed_runs": 0,
+        "failed_runs": 0,
+        "active_request_id": None,
+        "_stage_telemetry": _StageExecutionTelemetry(),
+    }
+
+     class Handler(BaseHTTPRequestHandler):
+         server_version = "MissionCorePersistentPerception/1"
+@@ -1844,6 +1972,7 @@
+             if self.path != "/health":
+                 self._send(404, {"ok": False, "error": "not-found"})
+                 return
+            stage_snapshot = state["_stage_telemetry"].snapshot()
+             self._send(
+                 200,
+                 {
+@@ -1853,6 +1982,11 @@
+                     "model_load_seconds": model_load_seconds,
+                     "completed_runs": state["completed_runs"],
+                     "failed_runs": state["failed_runs"],
+                    "active_request_id": state["active_request_id"],
+                    "active_frame_index": stage_snapshot["active_frame_index"],
+                    "current_stage": stage_snapshot["current_stage"],
+                    "active_stages": stage_snapshot["active_stages"],
+                    "stage_metrics": stage_snapshot["stages"],
+                     "authority": common["live"]["authority"],
+                     "gpu": torch.cuda.get_device_name(),
+                 },
+@@ -1882,7 +2016,9 @@
+                 request_id = str(document.get("request_id", "invalid"))
+                 run_args = _persistent_run_arguments(args, document)
+                 document["token"] = None
+-                exit_code = run(run_args, loaded)
+                state["active_request_id"] = request_id
+                state["_stage_telemetry"] = _StageExecutionTelemetry()
+                exit_code = run(run_args, loaded, state)
+                 state["completed_runs"] += 1
+                 self._send(
+                     200,
+@@ -1917,6 +2053,7 @@
+                 )
+             finally:
+                 state["busy"] = False
+                state["active_request_id"] = None
+                 run_lock.release()
+
+     server = ThreadingHTTPServer((args.listen_host, args.listen_port), Handler)
@@ -474,11 +474,57 @@ def _agent_raw_document(document: dict[str, Any]) -> dict[str, Any]:
    }
    docker_stats: dict[str, dict[str, str]] = raw["docker_stats"]
    container_states: dict[str, dict[str, object]] = raw["container_states"]
+    perception: dict[str, Any] = raw["perception"]
+    pipeline_stage_metrics: dict[str, dict[str, Any]] = {}
+    pipeline_active_stages: set[str] = set()
    for sample in samples:
        measurement = sample.get("measurement")
        payload, fields, tags = _metric_payload(sample)
        if sample.get("kind") == "pipeline":
-            raw["perception"] = _mapping(payload.get("payload")) or payload
+            if measurement == "missioncore_pipeline":
+                stage_id = tags.get("stage_id")
+                if isinstance(stage_id, str) and stage_id:
+                    stage_state = fields.get("stage_state")
+                    if stage_state == "active":
+                        pipeline_active_stages.add(stage_id)
+                    pipeline_stage_metrics[stage_id] = {
+                        "elapsed_seconds": _number(fields.get("elapsed_seconds")),
+                        "activations": fields.get("activations"),
+                        "share_percent": _number(fields.get("share_percent")),
+                    }
+                for name in (
+                    "service_state",
+                    "current_stage",
+                    "active_request_id",
+                ):
+                    value = fields.get(name)
+                    if isinstance(value, str):
+                        perception[name if name != "service_state" else "state"] = (
+                            value or None
+                        )
+                for name in ("active_frame_index", "completed_runs", "failed_runs"):
+                    value = fields.get(name)
+                    if isinstance(value, int) and not isinstance(value, bool):
+                        perception[name] = value if value >= 0 else None
+                model_load_seconds = _number(fields.get("model_load_seconds"))
+                if model_load_seconds is not None:
+                    perception["model_load_seconds"] = model_load_seconds
+                perception["collector_state"] = fields.get("collector_state")
+            else:
+                pipeline_payload = _mapping(payload.get("payload")) or payload
+                for name, value in pipeline_payload.items():
+                    if name == "stage_metrics":
+                        for stage_id, metrics in _mapping(value).items():
+                            if isinstance(metrics, dict):
+                                pipeline_stage_metrics[stage_id] = metrics
+                    elif name == "active_stages":
+                        pipeline_active_stages.update(
+                            stage_id
+                            for stage_id in _items(value)
+                            if isinstance(stage_id, str)
+                        )
+                    else:
+                        perception[name] = value
            continue
        if sample.get("kind") == "runtime":
            runtime_payload = _mapping(payload.get("payload")) or payload
@@ -610,6 +656,10 @@ def _agent_raw_document(document: dict[str, Any]) -> dict[str, Any]:
                    "version": None,
                    "uptime_seconds": uptime,
                }
+    if pipeline_stage_metrics:
+        perception["stage_metrics"] = pipeline_stage_metrics
+    if pipeline_active_stages:
+        perception["active_stages"] = sorted(pipeline_active_stages)
    return raw


@@ -274,6 +274,57 @@ def test_worker_telemetry_prefers_ndc_container_names_during_migration(
    assert triton["canonical_name"] == "ndc-mission-core-triton"


+def test_agent_pipeline_snapshot_merges_all_stage_series() -> None:
+    samples = []
+    for stage_id, stage_state, elapsed, activations, share in (
+        ("detector", "active", 2.5, 42, 62.5),
+        ("semantic-model", "waiting", 1.5, 11, 37.5),
+    ):
+        samples.append(
+            {
+                "node_id": EXPECTED_NODE_ID,
+                "kind": "pipeline",
+                "measurement": "missioncore_pipeline",
+                "observed_at_utc": "2026-07-28T12:00:00Z",
+                "payload": {
+                    "name": "missioncore_pipeline",
+                    "tags": {"stage_id": stage_id},
+                    "fields": {
+                        "service_state": "busy",
+                        "current_stage": "detector",
+                        "active_request_id": "request-006",
+                        "active_frame_index": 17,
+                        "completed_runs": 3,
+                        "failed_runs": 0,
+                        "model_load_seconds": 10.5,
+                        "collector_state": "live",
+                        "stage_state": stage_state,
+                        "elapsed_seconds": elapsed,
+                        "activations": activations,
+                        "share_percent": share,
+                    },
+                },
+            }
+        )
+
+    raw = _agent_raw_document({"samples": samples})
+
+    assert raw["perception"]["state"] == "busy"
+    assert raw["perception"]["active_stages"] == ["detector"]
+    assert raw["perception"]["stage_metrics"] == {
+        "detector": {
+            "elapsed_seconds": 2.5,
+            "activations": 42,
+            "share_percent": 62.5,
+        },
+        "semantic-model": {
+            "elapsed_seconds": 1.5,
+            "activations": 11,
+            "share_percent": 37.5,
+        },
+    }
+
+
 def test_worker_telemetry_history_keeps_one_row_per_agent_observation(
    tmp_path: Path,
 ) -> None:
@@ -181,6 +181,41 @@ def test_normalizer_preserves_native_pipeline_stage_tags() -> None:
    }


+def test_normalizer_accepts_telegraf_pipeline_snapshot() -> None:
+    normalizer = _normalizer()
+    row = normalizer._normalize(
+        "mission-core/v1/contours/worker-006/agents/worker-006/pipeline",
+        json.dumps(
+            {
+                "name": "missioncore_pipeline",
+                "tags": {
+                    "node_id": "DESKTOP-OPJ8J04",
+                    "contour_id": "worker-006",
+                    "agent_id": "worker-006",
+                    "stage_id": "detector",
+                },
+                "fields": {
+                    "service_state": "busy",
+                    "stage_state": "active",
+                    "elapsed_seconds": 2.5,
+                    "activations": 42,
+                },
+                "timestamp": 1_785_179_600,
+            }
+        ).encode(),
+    )
+
+    assert row[1:8] == (
+        "worker-006",
+        "worker-006",
+        "DESKTOP-OPJ8J04",
+        "pipeline",
+        "missioncore_pipeline",
+        '{"stage_id":"detector"}',
+        "telegraf.metric-json/v1",
+    )
+
+
 def test_normalizer_rejects_oversized_payload_and_series_identity() -> None:
    normalizer = _normalizer()
    with pytest.raises(ValueError, match="1 MiB"):