fix(telemetry): recover worker agent and Tailscale transport automatically

2026-09-25 23:24:27 +03:00
parent bd5873ef34
commit f3640026cf
11 changed files with 755 additions and 4 deletions
@@ -149,9 +149,11 @@ globally unique because Mosquitto ACL ownership is username-based.

 ## Worker agent

-MQTT outputs use `startup_error_behavior="retry"`, with a 2000-metric buffer and
-the configured flush interval. A missing broker at agent startup must not end
-the service. See [Telegraf's startup policy](https://docs.influxdata.com/telegraf/v1/configuration/plugin-options/).
+MQTT outputs request `startup_error_behavior="retry"`, with a 2000-metric buffer
+and the configured flush interval. This setting alone does not guarantee
+recovery: Telegraf 1.38.4 can exit cleanly when MQTT is unavailable at startup.
+The Tailscale profile below therefore also installs a system recovery task.
+See [Telegraf's startup policy](https://docs.influxdata.com/telegraf/v1/configuration/plugin-options/).
 If Docker lost a published listener while the saved LAN address is unchanged,
 the explicit broker Apply action reconciles that listener; a telemetry GET never
 restarts infrastructure. Node connectivity does not prove a profile is ready.
@@ -228,3 +230,26 @@ changing the K1 command sequence.

 The stack and agent are intentionally not started by repository tests. Provisioning a
 machine is a separate, explicit operation.
+
+## Tailscale operator profile and startup recovery
+
+The accepted 2026-09-25 operator profile no longer depends on a shared LAN or
+DHCP address. A prepared Windows agent uses `127.0.0.1:1883`; the existing strict
+Tailscale SSH identity carries a loopback reverse forward to the operator
+broker. Operator deployment is owned by `scripts/manage_telemetry_startup.py`
+(plan, hash-bound apply, rollback). This is a macOS **login** profile, not a
+pre-login Docker daemon. Preserve the explicit prepared stack root and its
+private credentials when moving the Core source checkout.
+
+For this profile the Windows install/update bundle includes and invokes
+`Install-NdcMissionCoreTelemetryRecovery.ps1`: boot + once-per-minute SYSTEM
+reconciliation for a stopped Telegraf service, independent of login. Its
+maintenance/rollback boundary is described in the audit. Do not rely solely on
+SCM failure actions or the MQTT `startup_error_behavior` setting: the pinned
+1.38.4 release was observed to terminate with a clean service exit when the
+broker was unavailable during boot.
+
+See [the measured recovery audit](../../docs/audits/2026-09-25-worker-telemetry-recovery.md)
+for exact acceptance, remaining cold-boot/clean-host gates and bounded test
+entrypoints. The earlier LAN workflow remains a legacy explicit configuration;
+it is not the current Worker 006 transport.
@@ -29,6 +29,7 @@ SAFE_IDENTIFIER: Final = re.compile(
 )
 SAFE_INTERVAL: Final = re.compile(r"^[1-9][0-9]{0,2}s$")
 WINDOWS_BUNDLE_FILES: Final = (
+    "Install-NdcMissionCoreTelemetryRecovery.ps1",
    "Get-NdcMissionCorePipelineTelemetry.ps1",
    "Install-NdcMissionCoreTelegraf.ps1",
    "Update-NdcMissionCoreTelegraf.ps1",
@@ -33,6 +33,10 @@ foreach ($name in @(
    Set-Item -Path "Env:$name" -Value ([string]$value)
 }

+$recoveryInstaller = Join-Path $PSScriptRoot 'Install-NdcMissionCoreTelemetryRecovery.ps1'
+if ($payload.MISSIONCORE_MQTT_HOST -eq '127.0.0.1' -and -not (Test-Path $recoveryInstaller -PathType Leaf)) {
+    throw 'Telemetry recovery installer missing from bundle'
+}
 if (Get-Service -Name $serviceName -ErrorAction SilentlyContinue) {
    throw "Service '$serviceName' already exists; refusing an implicit replacement"
 }
@@ -137,6 +141,10 @@ try {
    if ($LASTEXITCODE -ne 0) {
        throw "Failed to enable Telegraf recovery for non-crash failures"
    }
+    $recoveryTask = $null
+    if ($payload.MISSIONCORE_MQTT_HOST -eq '127.0.0.1') {
+        $recoveryTask = & $recoveryInstaller -ExpectedNodeId $env:COMPUTERNAME -Action Apply | ConvertFrom-Json
+    }
    Start-Service -Name $serviceName
    $service = Get-Service -Name $serviceName
    $service.WaitForStatus([ServiceProcess.ServiceControllerStatus]::Running, [TimeSpan]::FromSeconds(20))
@@ -150,6 +158,7 @@ try {
        Status = $service.Status.ToString()
        StartType = $service.StartType.ToString()
        RecoveryConfigured = $true
+        RecoveryTask = $recoveryTask
        Configuration = $configurationPath
        PipelineCollector = $collectorPath
        PipelineJournal = $PipelineJournal
@@ -0,0 +1,87 @@
+[CmdletBinding()]
+param(
+    [Parameter(Mandatory = $true)][string]$ExpectedNodeId,
+    [ValidateSet('Plan','Apply','Rollback')][string]$Action = 'Plan',
+    [string]$BackupDirectory
+)
+# Telegraf 1.38.4 can report MQTT startup failure as a clean Windows-service exit.
+# SCM failure actions alone cannot recover that stopped service.
+$ErrorActionPreference = 'Stop'
+$ProgressPreference = 'SilentlyContinue'
+if ($env:COMPUTERNAME -cne $ExpectedNodeId) { throw 'Worker identity mismatch' }
+$taskName = 'ndc-mission-core-telemetry-recovery'
+$root = Join-Path $env:ProgramFiles 'NDC\Mission Core\TelemetryRecovery'
+$guardPath = Join-Path $root 'Resume-Telemetry.ps1'
+$service = Get-CimInstance Win32_Service -Filter "Name='telegraf'"
+if (-not $service -or $service.PathName -notlike '*NDC\Mission Core\Telegraf\telegraf.exe*') { throw 'Managed Telegraf service required' }
+$guard = @'
+$ErrorActionPreference = 'Stop'
+$service = Get-Service -Name 'telegraf'
+if ($service.Status -eq 'Running') { exit 0 }
+if ($service.Status -ne 'Stopped') { exit 0 }
+$values = @{}
+foreach ($entry in (Get-ItemProperty 'HKLM:\SYSTEM\CurrentControlSet\Services\telegraf').Environment) {
+    $name, $value = $entry -split '=', 2
+    if ($name -in @('MISSIONCORE_MQTT_HOST','MISSIONCORE_MQTT_PORT')) { $values[$name] = $value }
+}
+if ($values['MISSIONCORE_MQTT_HOST'] -ne '127.0.0.1' -or $values['MISSIONCORE_MQTT_PORT'] -ne '1883') { exit 1 }
+$client = [Net.Sockets.TcpClient]::new()
+try {
+    $connect = $client.ConnectAsync('127.0.0.1', 1883)
+    if (-not $connect.Wait(3000) -or -not $client.Connected) { exit 0 }
+} catch { exit 0 } finally { $client.Dispose() }
+Start-Service -Name 'telegraf'
+(Get-Service -Name 'telegraf').WaitForStatus([ServiceProcess.ServiceControllerStatus]::Running, [TimeSpan]::FromSeconds(15))
+'@
+if ($Action -eq 'Plan') {
+    [ordered]@{node=$env:COMPUTERNAME;task=$taskName;path=$guardPath;startMode=$service.StartMode;triggers=@('boot','every-minute');scope='Start stopped telemetry service only';requiresInteractiveLogin=$false}|ConvertTo-Json -Compress
+    exit 0
+}
+if ($Action -eq 'Rollback') {
+    if (-not $BackupDirectory) { throw 'BackupDirectory required' }
+    $receipt = Get-Content (Join-Path $BackupDirectory 'receipt.json') -Raw | ConvertFrom-Json
+    if ($receipt.node -cne $ExpectedNodeId -or $receipt.task -ne $taskName) { throw 'Backup identity mismatch' }
+    if ((Get-FileHash $guardPath -Algorithm SHA256).Hash -ne $receipt.installedHash) { throw 'Installed recovery script changed' }
+    Unregister-ScheduledTask -TaskName $taskName -Confirm:$false -ErrorAction SilentlyContinue
+    if (Test-Path (Join-Path $BackupDirectory 'task.xml')) {
+        Register-ScheduledTask -TaskName $taskName -Xml (Get-Content (Join-Path $BackupDirectory 'task.xml') -Raw) -Force | Out-Null
+    }
+    if (Test-Path (Join-Path $BackupDirectory 'guard.ps1')) {
+        Copy-Item (Join-Path $BackupDirectory 'guard.ps1') $guardPath -Force
+    } else { Remove-Item $guardPath }
+    [ordered]@{restored=$true;task=$taskName}|ConvertTo-Json -Compress
+    exit 0
+}
+$principal = [Security.Principal.WindowsPrincipal][Security.Principal.WindowsIdentity]::GetCurrent()
+if (-not $principal.IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator)) { throw 'Administrative installer required' }
+if (Test-Path $root) {
+    if ((Get-Item $root -Force).Attributes -band [IO.FileAttributes]::ReparsePoint) { throw 'Recovery directory must not be a reparse point' }
+}
+New-Item -ItemType Directory -Path $root -Force | Out-Null
+# SYSTEM executes this file: only SYSTEM and Administrators may modify it.
+& icacls.exe $root /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) { throw 'Recovery directory ACL failed' }
+$backup = Join-Path $root ('backups\' + [Guid]::NewGuid().ToString('N'))
+New-Item -ItemType Directory -Path $backup -Force | Out-Null
+$previousTask = Get-ScheduledTask -TaskName $taskName -ErrorAction SilentlyContinue
+if ($previousTask) { Export-ScheduledTask -TaskName $taskName | Set-Content (Join-Path $backup 'task.xml') -Encoding UTF8 }
+if (Test-Path $guardPath) { Copy-Item $guardPath (Join-Path $backup 'guard.ps1') }
+try {
+    [IO.File]::WriteAllText($guardPath, $guard, [Text.UTF8Encoding]::new($false))
+    $exe = Join-Path $env:SystemRoot 'System32\WindowsPowerShell\v1.0\powershell.exe'
+    $run = New-ScheduledTaskAction -Execute $exe -Argument ('-NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass -File "' + $guardPath + '"')
+    $triggers = @((New-ScheduledTaskTrigger -AtStartup), (New-ScheduledTaskTrigger -Once -At (Get-Date).AddMinutes(1) -RepetitionInterval (New-TimeSpan -Minutes 1)))
+    $settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -MultipleInstances IgnoreNew -ExecutionTimeLimit (New-TimeSpan -Seconds 45) -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries
+    $identity = New-ScheduledTaskPrincipal -UserId 'SYSTEM' -LogonType ServiceAccount -RunLevel Highest
+    Register-ScheduledTask -TaskName $taskName -Action $run -Trigger $triggers -Settings $settings -Principal $identity -Force | Out-Null
+    $receipt = [ordered]@{node=$env:COMPUTERNAME;task=$taskName;installedHash=(Get-FileHash $guardPath -Algorithm SHA256).Hash;backup=$backup}
+    $receipt|ConvertTo-Json|Set-Content (Join-Path $backup 'receipt.json') -Encoding UTF8
+    Start-ScheduledTask -TaskName $taskName
+    $receipt|ConvertTo-Json -Compress
+} catch {
+    if ($previousTask) { Register-ScheduledTask -TaskName $taskName -Xml (Get-Content (Join-Path $backup 'task.xml') -Raw) -Force | Out-Null }
+    else { Unregister-ScheduledTask -TaskName $taskName -Confirm:$false -ErrorAction SilentlyContinue }
+    if (Test-Path (Join-Path $backup 'guard.ps1')) { Copy-Item (Join-Path $backup 'guard.ps1') $guardPath -Force }
+    else { Remove-Item $guardPath -ErrorAction SilentlyContinue }
+    throw
+}
@@ -0,0 +1,26 @@
+[CmdletBinding()]
+param([Parameter(Mandatory=$true)][string]$ExpectedNodeId)
+$ErrorActionPreference='Stop'
+$ProgressPreference='SilentlyContinue'
+if ($env:COMPUTERNAME -cne $ExpectedNodeId) { throw 'Worker identity mismatch' }
+$task=Get-ScheduledTask -TaskName 'ndc-mission-core-telemetry-recovery'
+$account=[Security.Principal.NTAccount]::new($task.Principal.UserId)
+$sid=$account.Translate([Security.Principal.SecurityIdentifier]).Value
+if (-not $task.Settings.Enabled -or $sid -ne 'S-1-5-18') { throw 'Recovery task not enabled as SYSTEM' }
+$before=Get-CimInstance Win32_Service -Filter "Name='telegraf'"
+if ($before.State -ne 'Running') { throw 'A running baseline is required' }
+$started=[DateTime]::UtcNow
+Stop-Service telegraf
+(Get-Service telegraf).WaitForStatus([ServiceProcess.ServiceControllerStatus]::Stopped,[TimeSpan]::FromSeconds(20))
+$automatic=$false
+try {
+    while (([DateTime]::UtcNow-$started).TotalSeconds -lt 80) {
+        Start-Sleep -Seconds 2
+        $after=Get-CimInstance Win32_Service -Filter "Name='telegraf'"
+        if ($after.State -eq 'Running' -and $after.ProcessId -ne $before.ProcessId) { $automatic=$true;break }
+    }
+    [ordered]@{node=$ExpectedNodeId;test='clean-service-stop';automaticRecovery=$automatic;elapsedSeconds=[Math]::Round(([DateTime]::UtcNow-$started).TotalSeconds,2);previousPid=$before.ProcessId;newPid=$after.ProcessId;state=$after.State}|ConvertTo-Json -Compress
+    if (-not $automatic) { throw 'Automatic recovery failed; restoring baseline' }
+} finally {
+    if ((Get-Service telegraf).Status -ne 'Running') { Start-Service telegraf }
+}
@@ -67,6 +67,10 @@ if (-not ($serviceEnvironmentCandidate | Where-Object {
    )
 }

+$recoveryInstaller = Join-Path $PSScriptRoot 'Install-NdcMissionCoreTelemetryRecovery.ps1'
+if ($env:MISSIONCORE_MQTT_HOST -eq '127.0.0.1' -and -not (Test-Path $recoveryInstaller -PathType Leaf)) {
+    throw 'Telemetry recovery installer missing from bundle'
+}
 $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"
@@ -127,6 +131,10 @@ try {
        if ($LASTEXITCODE -ne 0) {
            throw "Failed to enable Telegraf recovery for non-crash failures"
        }
+        $recoveryTask = $null
+        if ($env:MISSIONCORE_MQTT_HOST -eq '127.0.0.1') {
+            $recoveryTask = & $recoveryInstaller -ExpectedNodeId $env:COMPUTERNAME -Action Apply | ConvertFrom-Json
+        }
    }
    catch {
        Stop-Service -Name $serviceName -Force -ErrorAction SilentlyContinue
@@ -155,6 +163,7 @@ try {
        ServiceName = $serviceName
        Status = (Get-Service -Name $serviceName).Status.ToString()
        RecoveryConfigured = $true
+        RecoveryTask = $recoveryTask
        Configuration = $configurationPath
        Backup = $backupPath
        PipelineCollector = $collectorPath
@@ -0,0 +1,99 @@
+"""Bounded operator acceptance of installed telemetry recovery, no inference jobs.
+
+Each case injects one failure in an exact Mission Core telemetry component.
+An existing service's automatic supervisor must restore fresh worker telemetry.
+"""
+
+import argparse
+import json
+import os
+import re
+import subprocess
+import time
+import urllib.request
+from datetime import datetime
+
+DOCKER = "/Applications/Docker.app/Contents/Resources/bin/docker"
+BASE = "http://127.0.0.1:8000/api/v1/system/contours/worker-006/telemetry?history=1"
+
+
+def read():
+    return json.load(urllib.request.urlopen(BASE, timeout=5))
+
+
+def tunnel_pid():
+    r = subprocess.run(
+        ["launchctl", "print", f"gui/{os.getuid()}/com.nodedc.telemetry-tunnel.local"],
+        capture_output=True,
+        text=True,
+    )
+    m = re.search(r"\bpid = (\d+)", r.stdout)
+    return int(m.group(1)) if m else None
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("case", choices=["tunnel", "receiver"])
+    args = parser.parse_args()
+    baseline = read()["connection"]
+    if not baseline["reachable"] or not baseline["identity_matches"]:
+        raise RuntimeError("Fresh confirmed baseline required")
+    started = time.monotonic()
+    baseline_time = datetime.fromisoformat(baseline["observed_at_utc"].replace("Z", "+00:00"))
+    before_pid = tunnel_pid()
+    if args.case == "tunnel":
+        subprocess.run(
+            [
+                "launchctl",
+                "kill",
+                "SIGKILL",
+                f"gui/{os.getuid()}/com.nodedc.telemetry-tunnel.local",
+            ],
+            check=True,
+            capture_output=True,
+        )
+    else:
+        subprocess.run(
+            [DOCKER, "stop", "--time", "5", "ndc-mission-core-telemetry-normalizer"],
+            check=True,
+            capture_output=True,
+            timeout=15,
+        )
+    seen = set()
+    while time.monotonic() - started < 90:
+        time.sleep(2)
+        try:
+            probe = read()["connection"]
+            code = probe.get("error_code")
+            seen.add(code or "fresh")
+            observed = datetime.fromisoformat(probe["observed_at_utc"].replace("Z", "+00:00"))
+            if (
+                probe["reachable"]
+                and probe["identity_matches"]
+                and (observed - baseline_time).total_seconds() >= 10
+                and time.monotonic() - started >= 10
+                and (
+                    args.case != "tunnel"
+                    or (tunnel_pid() is not None and tunnel_pid() != before_pid)
+                )
+                and (args.case != "receiver" or "telemetry-receiver-unavailable" in seen)
+            ):
+                print(
+                    json.dumps(
+                        {
+                            "case": args.case,
+                            "automaticRecovery": True,
+                            "elapsedSeconds": round(time.monotonic() - started, 2),
+                            "observedStates": sorted(seen),
+                            "observedAt": probe["observed_at_utc"],
+                        }
+                    )
+                )
+                return
+        except (OSError, ValueError):
+            seen.add("query-unavailable")
+    raise RuntimeError("Automatic telemetry recovery not accepted within 90 seconds")
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,333 @@
+#!/usr/bin/env python3
+"""Versioned macOS telemetry startup: plan/apply/rollback, no worker job launch.
+
+The prepared stack owns credentials and volumes. SSH provides only a loopback
+MQTT forward through an existing strictly pinned Tailscale SSH profile.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import plistlib
+import re
+import subprocess
+import time
+import urllib.request
+from pathlib import Path
+
+from manage_mission_core_launch_agent import _wait_for_health, _write_atomic
+from migrate_worker_tunnel_logs import reload_agent
+
+from k1link.launchd_logs import launchd_log_path, prepare_launchd_log
+from k1link.web.compute_contour_network import _replace_environment_value
+
+STARTUP = "com.nodedc.telemetry-startup.local"
+TUNNEL = "com.nodedc.telemetry-tunnel.local"
+CORE = "com.nodedc.mission-core.local"
+DOCKER = "/Applications/Docker.app/Contents/Resources/bin/docker"
+
+
+def command(args, timeout=30):
+    # Never echo Compose output: expanded environments can contain credentials.
+    return subprocess.run(args, capture_output=True, timeout=timeout, check=False)
+
+
+def receiver_ready():
+    try:
+        with urllib.request.urlopen("http://127.0.0.1:18030/health", timeout=3) as r:
+            d = json.loads(r.read())
+        return (
+            d.get("ok") is True
+            and d.get("mqtt_connected") is True
+            and d.get("database_reachable") is True
+        )
+    except (OSError, ValueError):
+        return False
+
+
+def broker_loopback():
+    result = command(
+        [
+            DOCKER,
+            "inspect",
+            "--format",
+            "{{json .NetworkSettings.Ports}}",
+            "ndc-mission-core-mqtt-broker",
+        ],
+        10,
+    )
+    try:
+        bindings = json.loads(result.stdout).get("1883/tcp", [])
+        return result.returncode == 0 and bindings == [{"HostIp": "127.0.0.1", "HostPort": "1883"}]
+    except (ValueError, AttributeError):
+        return False
+
+
+def reconcile(stack: Path):
+    """One bounded retry, repeated by launchd after delayed Docker/network start."""
+    if receiver_ready() and broker_loopback():
+        return "ready"
+    if command([DOCKER, "info", "--format", "{{.ServerVersion}}"], 10).returncode:
+        # Desktop's own login setting is not a dependency manager. No VM/resource
+        # setting is changed and no user inference job is started here.
+        command(["/usr/bin/open", "-g", "-a", "/Applications/Docker.app"], 10)
+        return "waiting-for-docker"
+    compose = [
+        DOCKER,
+        "compose",
+        "--project-directory",
+        str(stack),
+        "--env-file",
+        str(stack / ".env"),
+        "-f",
+        str(stack / "compose.yaml"),
+    ]
+    if command([*compose, "config", "--quiet"], 20).returncode:
+        return "configuration-unavailable"
+    if command(
+        [
+            *compose,
+            "up",
+            "-d",
+            "--no-build",
+            "--pull",
+            "never",
+            "--wait",
+            "--wait-timeout",
+            "45",
+            "broker",
+            "timescale",
+            "normalizer",
+        ],
+        60,
+    ).returncode:
+        return "waiting-for-receiver"
+    return "ready" if receiver_ready() and broker_loopback() else "waiting-for-receiver"
+
+
+def private_file(path):
+    if path.is_symlink() or not path.is_file() or path.stat().st_uid != os.getuid():
+        raise ValueError("Expected owned regular configuration file")
+    return path.read_bytes()
+
+
+def plan(stack: Path, ssh_alias: str, repository: Path, agents: Path):
+    if not stack.is_absolute() or stack.resolve() != stack or not repository.is_absolute():
+        raise ValueError("Canonical absolute installation paths required")
+    if re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,80}", ssh_alias) is None:
+        raise ValueError("Invalid SSH profile")
+    for name in (
+        "compose.yaml",
+        "runtime/agents.json",
+        "runtime/mosquitto/acl",
+        "runtime/mosquitto/passwords",
+    ):
+        private_file(stack / name)
+    env = private_file(stack / ".env")
+    public = dict(
+        line.split("=", 1)
+        for line in env.decode().splitlines()
+        if "=" in line and not line.startswith("#")
+    )
+    if (
+        public.get("MISSIONCORE_MQTT_PORT", "1883") != "1883"
+        or public.get("MISSIONCORE_TELEMETRY_QUERY_PORT", "18030") != "18030"
+    ):
+        raise ValueError("This deployment profile requires MQTT 1883 and query 18030")
+    # Verify the existing trust boundary; never enroll a host or accept a new key.
+    ssh = command(["/usr/bin/ssh", "-G", ssh_alias])
+    settings = dict(line.split(" ", 1) for line in ssh.stdout.decode().splitlines() if " " in line)
+    if (
+        ssh.returncode
+        or settings.get("stricthostkeychecking") not in ("true", "yes")
+        or not settings.get("hostname", "").endswith(".ts.net")
+    ):
+        raise ValueError("A strictly pinned Tailscale SSH profile is required")
+    python = repository / ".venv/bin/python"
+    if not python.is_file() or not Path(DOCKER).is_file():
+        raise ValueError("Prepared application Python and Docker Desktop are required")
+    common = dict(
+        RunAtLoad=True, AbandonProcessGroup=False, ProcessType="Background", ExitTimeOut=15
+    )
+    startup = dict(
+        common,
+        Label=STARTUP,
+        StartInterval=30,
+        ProgramArguments=[
+            str(python),
+            str(repository / "scripts/manage_telemetry_startup.py"),
+            "reconcile",
+            "--stack-root",
+            str(stack),
+        ],
+        StandardOutPath="/dev/null",
+        StandardErrorPath=str(launchd_log_path("telemetry-startup.log")),
+    )
+    tunnel = dict(
+        common,
+        Label=TUNNEL,
+        KeepAlive=True,
+        ThrottleInterval=10,
+        ProgramArguments=[
+            "/usr/bin/ssh",
+            "-o",
+            "BatchMode=yes",
+            "-o",
+            "StrictHostKeyChecking=yes",
+            "-o",
+            "ExitOnForwardFailure=yes",
+            "-o",
+            "ConnectTimeout=8",
+            "-o",
+            "ServerAliveInterval=10",
+            "-o",
+            "ServerAliveCountMax=3",
+            "-N",
+            "-T",
+            "-R",
+            "127.0.0.1:1883:127.0.0.1:1883",
+            ssh_alias,
+        ],
+        StandardOutPath="/dev/null",
+        StandardErrorPath=str(launchd_log_path("telemetry-tunnel.log")),
+    )
+    core_path = agents / (CORE + ".plist")
+    core = plistlib.loads(private_file(core_path))
+    if core.get("Label") != CORE or core.get("WorkingDirectory") != str(repository):
+        raise ValueError("Canonical Mission Core installation changed")
+    core["EnvironmentVariables"]["MISSIONCORE_TELEMETRY_PLANE_ROOT"] = str(stack)
+    desired = {
+        stack / ".env": _replace_environment_value(
+            env.decode(), "MISSIONCORE_MQTT_BIND_ADDRESS", "127.0.0.1"
+        ).encode()
+    }
+    for label, doc in ((STARTUP, startup), (TUNNEL, tunnel), (CORE, core)):
+        path = agents / (label + ".plist")
+        if path.exists() and plistlib.loads(private_file(path)).get("Label") != label:
+            raise ValueError("Foreign LaunchAgent at target path")
+        desired[path] = plistlib.dumps(doc, sort_keys=True)
+    changes = []
+    for path, data in desired.items():
+        before = private_file(path) if path.exists() else None
+        changes.append(
+            dict(
+                path=str(path),
+                before=hashlib.sha256(before).hexdigest() if before is not None else None,
+                after=hashlib.sha256(data).hexdigest(),
+            )
+        )
+    document = dict(
+        schema_version="missioncore.telemetry-startup-plan/v1",
+        changes=changes,
+        stack_root=str(stack),
+        ssh_profile=ssh_alias,
+        mqtt="loopback-over-ssh-tailscale",
+        startup="macOS user login; retry every 30s",
+        worker_jobs_started=False,
+    )
+    document["artifact_sha256"] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
+    document["compose_sha256"] = hashlib.sha256((stack / "compose.yaml").read_bytes()).hexdigest()
+    document["sha256"] = hashlib.sha256(json.dumps(document, sort_keys=True).encode()).hexdigest()
+    return document, desired
+
+
+def restore(backup: Path, agents: Path):
+    manifest = json.loads(private_file(backup / "manifest.json"))
+    for item in manifest["changes"]:
+        path = Path(item["path"])
+        current = hashlib.sha256(private_file(path)).hexdigest() if path.exists() else None
+        if current not in (item["before"], item["after"]):
+            raise ValueError("Installed configuration changed; refusing rollback")
+    for label in (STARTUP, TUNNEL):
+        command(["launchctl", "bootout", f"gui/{os.getuid()}/{label}"])
+    for i, item in enumerate(manifest["changes"]):
+        path = Path(item["path"])
+        if item["before"] is None:
+            path.unlink(missing_ok=True)
+        else:
+            old = private_file(backup / str(i))
+            if hashlib.sha256(old).hexdigest() != item["before"]:
+                raise ValueError("Rollback payload changed")
+            _write_atomic(path, old)
+    for label in (STARTUP, TUNNEL, CORE):
+        path = agents / (label + ".plist")
+        if path.exists():
+            reload_agent(path, label)
+    if not _wait_for_health(45):
+        raise RuntimeError("Core health not accepted after rollback")
+    return {
+        "restored": True,
+        "container_state": "not changed; declared endpoint restored on next reconciliation",
+    }
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("action", choices=["plan", "apply", "reconcile", "rollback"])
+    p.add_argument("--stack-root", type=Path)
+    p.add_argument("--repository-root", type=Path, default=Path(__file__).resolve().parents[1])
+    p.add_argument("--ssh-alias", default="mission-gpu")
+    p.add_argument("--expected-sha256")
+    p.add_argument("--backup", type=Path)
+    a = p.parse_args()
+    agents = Path.home() / "Library/LaunchAgents"
+    if a.action == "rollback":
+        if a.backup is None:
+            p.error("--backup required")
+        print(json.dumps(restore(a.backup, agents)))
+        return
+    if a.stack_root is None:
+        p.error("--stack-root required")
+    if a.action == "reconcile":
+        try:
+            phase = reconcile(a.stack_root)
+        except (OSError, subprocess.SubprocessError, ValueError):
+            phase = "receiver-unavailable"
+        # No repeating stdout log; health remains independently observable.
+        print(json.dumps({"phase": phase}))
+        return
+    document, desired = plan(a.stack_root, a.ssh_alias, a.repository_root, agents)
+    if a.action == "plan":
+        print(json.dumps(document, indent=2))
+        return
+    if a.expected_sha256 != document["sha256"]:
+        raise ValueError("Startup plan changed before apply")
+    backup = (
+        a.repository_root
+        / ".runtime/mission-core/telemetry-service-backups"
+        / (str(time.time_ns()))
+    )
+    backup.mkdir(parents=True, mode=0o700)
+    _write_atomic(backup / "manifest.json", json.dumps(document).encode())
+    for i, path in enumerate(desired):
+        if path.exists():
+            _write_atomic(backup / str(i), private_file(path))
+    for name in ("telemetry-startup.log", "telemetry-tunnel.log"):
+        prepare_launchd_log(launchd_log_path(name))
+    try:
+        for path, data in desired.items():
+            _write_atomic(path, data)
+        for label in (STARTUP, TUNNEL, CORE):
+            reload_agent(agents / (label + ".plist"), label)
+        if not _wait_for_health(45):
+            raise RuntimeError("Canonical Core failed health acceptance")
+    except BaseException:
+        restore(backup, agents)
+        raise
+    print(
+        json.dumps(
+            {
+                "installed": True,
+                "backup": str(backup),
+                "plan_sha256": document["sha256"],
+                "telemetry_acceptance": "pending fresh worker sample",
+            }
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()
@@ -423,7 +423,9 @@ def build_compute_contour_router(*, root_provider: RootProvider) -> APIRouter:
        try:
            contour = store.get(contour_id)
            telemetry_plane_root = (
-                Path(__file__).resolve().parents[3] / "deploy" / "telemetry-plane"
+                Path(os.environ["MISSIONCORE_TELEMETRY_PLANE_ROOT"])
+                if os.environ.get("MISSIONCORE_TELEMETRY_PLANE_ROOT")
+                else Path(__file__).resolve().parents[3] / "deploy" / "telemetry-plane"
            )
            return apply_broker_network(
                _network_target(contour),
@@ -144,3 +144,17 @@ def test_contour_router_returns_404_for_unknown_contour(tmp_path: Path) -> None:
    with pytest.raises(HTTPException) as error:
        install("missing")
    assert error.value.status_code == 404
+
+
+def test_broker_apply_uses_explicit_installation_root(tmp_path, monkeypatch):
+    from k1link.web import compute_contour_api as api
+    root = tmp_path / 'prepared-stack'
+    monkeypatch.setenv('MISSIONCORE_TELEMETRY_PLANE_ROOT', str(root))
+    seen = []
+    def apply(target, installation):
+        seen.append(installation)
+        return {'ready': True}
+    monkeypatch.setattr(api, 'apply_broker_network', apply)
+    router = api.build_compute_contour_router(root_provider=lambda: tmp_path / 'system')
+    _endpoint(router, '/api/v1/system/contours/{contour_id}/network/broker', 'POST')('worker-006')
+    assert seen == [root]
@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import plistlib
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+SCRIPTS = Path(__file__).parents[1] / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+try:
+    spec = importlib.util.spec_from_file_location(
+        "telemetry_startup", SCRIPTS / "manage_telemetry_startup.py"
+    )
+    manager = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(manager)
+finally:
+    sys.path.pop(0)
+
+
+def setup(tmp_path, monkeypatch):
+    stack = tmp_path / "state"
+    repo = tmp_path / "release"
+    agents = tmp_path / "agents"
+    for p in (stack, repo, agents):
+        p.mkdir()
+    for name in (
+        "compose.yaml",
+        "runtime/agents.json",
+        "runtime/mosquitto/acl",
+        "runtime/mosquitto/passwords",
+    ):
+        p = stack / name
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text("fixture")
+    (stack / ".env").write_text("MISSIONCORE_MQTT_BIND_ADDRESS=192.168.1.5\nSECRET=preserve-me\n")
+    (repo / ".venv/bin").mkdir(parents=True)
+    (repo / ".venv/bin/python").write_text("fixture")
+    docker = tmp_path / "docker"
+    docker.write_text("fixture")
+    monkeypatch.setattr(manager, "DOCKER", str(docker))
+    (agents / (manager.CORE + ".plist")).write_bytes(
+        plistlib.dumps(
+            {
+                "Label": manager.CORE,
+                "WorkingDirectory": str(repo),
+                "EnvironmentVariables": {"MISSIONCORE_DATA_DIR": "/existing-data"},
+            }
+        )
+    )
+    monkeypatch.setattr(
+        manager,
+        "command",
+        lambda *args: subprocess.CompletedProcess(
+            [], 0, b"hostname worker.example.ts.net\nstricthostkeychecking true\n", b""
+        ),
+    )
+    return stack, repo, agents
+
+
+def test_plan_preserves_credentials_and_pins_loopback_transport(tmp_path, monkeypatch):
+    stack, repo, agents = setup(tmp_path, monkeypatch)
+    before = (stack / ".env").read_bytes()
+    doc, files = manager.plan(stack, "worker-test", repo, agents)
+    assert (stack / ".env").read_bytes() == before
+    assert files[stack / ".env"] == b"MISSIONCORE_MQTT_BIND_ADDRESS=127.0.0.1\nSECRET=preserve-me\n"
+    assert "preserve-me" not in json.dumps(doc)
+    startup = plistlib.loads(files[agents / (manager.STARTUP + ".plist")])
+    assert startup["RunAtLoad"] and startup["StartInterval"] == 30
+    tunnel = plistlib.loads(files[agents / (manager.TUNNEL + ".plist")])
+    assert "127.0.0.1:1883:127.0.0.1:1883" in tunnel["ProgramArguments"]
+    assert "StrictHostKeyChecking=yes" in tunnel["ProgramArguments"]
+    assert tunnel["KeepAlive"] and tunnel["ThrottleInterval"] >= 5
+    core = plistlib.loads(files[agents / (manager.CORE + ".plist")])
+    assert core["EnvironmentVariables"]["MISSIONCORE_DATA_DIR"] == "/existing-data"
+    assert core["EnvironmentVariables"]["MISSIONCORE_TELEMETRY_PLANE_ROOT"] == str(stack)
+
+
+def test_plan_refuses_non_tailscale_and_changed_input(tmp_path, monkeypatch):
+    stack, repo, agents = setup(tmp_path, monkeypatch)
+    before, _ = manager.plan(stack, "worker-test", repo, agents)
+    with (stack / ".env").open("a") as f:
+        f.write("NEW=setting\n")
+    after, _ = manager.plan(stack, "worker-test", repo, agents)
+    assert before["sha256"] != after["sha256"]
+    monkeypatch.setattr(
+        manager,
+        "command",
+        lambda *args: subprocess.CompletedProcess(
+            [], 0, b"hostname worker.local\nstricthostkeychecking true\n", b""
+        ),
+    )
+    with pytest.raises(ValueError, match="Tailscale"):
+        manager.plan(stack, "worker-test", repo, agents)
+
+
+def test_delayed_docker_retries_without_compose_or_workload_launch(monkeypatch, tmp_path):
+    calls = []
+    monkeypatch.setattr(manager, "receiver_ready", lambda: False)
+
+    def run(args, timeout):
+        calls.append(args)
+        return subprocess.CompletedProcess(args, 1 if "info" in args else 0)
+
+    monkeypatch.setattr(manager, "command", run)
+    assert manager.reconcile(tmp_path) == "waiting-for-docker"
+    assert len(calls) == 2 and calls[1][0] == "/usr/bin/open"
+    assert not any("compose" in c for c in calls)
+
+
+def test_ready_receiver_is_not_restarted(monkeypatch, tmp_path):
+    monkeypatch.setattr(manager, "receiver_ready", lambda: True)
+    monkeypatch.setattr(manager, "broker_loopback", lambda: True)
+    monkeypatch.setattr(
+        manager, "command", lambda *args: pytest.fail("Must not restart healthy services")
+    )
+    assert manager.reconcile(tmp_path) == "ready"
+
+
+def test_old_lan_binding_is_reconciled_even_when_receiver_healthy(monkeypatch, tmp_path):
+    calls = []
+    bindings = iter((False, True))
+    monkeypatch.setattr(manager, "receiver_ready", lambda: True)
+    monkeypatch.setattr(manager, "broker_loopback", lambda: next(bindings))
+
+    def run(args, timeout):
+        calls.append(args)
+        return subprocess.CompletedProcess(args, 0)
+
+    monkeypatch.setattr(manager, "command", run)
+    assert manager.reconcile(tmp_path) == "ready"
+    assert calls[-1][-3:] == ["broker", "timescale", "normalizer"]
+    assert "--no-build" in calls[-1] and "--pull" in calls[-1] and "never" in calls[-1]
+
+
+def test_failed_compose_reports_waiting_without_ready(monkeypatch, tmp_path):
+    monkeypatch.setattr(manager, "receiver_ready", lambda: False)
+    monkeypatch.setattr(
+        manager,
+        "command",
+        lambda args, timeout: subprocess.CompletedProcess(args, int("up" in args)),
+    )
+    assert manager.reconcile(tmp_path) == "waiting-for-receiver"