# Per-Cluster Mesh Phase 5 — gRPC live-telemetry stream Implementation Plan

> **For Claude:** REQUIRED SUB-SKILL: execute this with superpowers-extended-cc:subagent-driven-development
> (chosen for the program's prior phases). Fresh subagent per task; classification-driven review chain.

**Goal:** Replace the DPS fan-out of the live-telemetry observability channels with one gRPC
server-streaming contract — **each driver node hosts the telemetry gRPC server (Kestrel h2c),
central dials in** — so AdminUI live panels keep working once the meshes split (Phase 6) and no
longer depend on a shared Akka gossip ring for fleet observability. Ships as a per-node **dark
switch** (`Telemetry:Mode` = `Dps` default | `Grpc`), authenticated fail-closed from day one.

**Architecture:** Mirror ScadaBridge's `SiteStreamService` shape (recon'd 2026-07-23), adapted to
OtOpcUa's substrate. Node side: the four telemetry publish seams also emit into a **node-local
in-process hub** (`ITelemetryLocalHub`) — DPS publishing is left untouched, so the switch is pure —
and a streaming gRPC service fans the hub to connected clients through per-subscriber bounded
`DropOldest` channels. Central side: a supervisor actor discovers driver-node gRPC endpoints from
`ClusterNode` rows (`Host` + `GrpcPort`, added in Phase 1), keeps one reconnecting dialer per node,
converts streamed events back to the domain records, and feeds **the exact same in-process sinks the
DPS SignalR bridges feed today** (`IInProcessBroadcaster<AlarmTransitionEvent>`,
`IInProcessBroadcaster<ScriptLogEntry>`, `IDriverStatusSnapshotStore`,
`IDriverResilienceStatusStore`). The AdminUI components are untouched: only the bridge's *upstream*
swaps.

**Tech Stack:** .NET 10, `Grpc.AspNetCore` / `Grpc.Net.Client` (already in-repo via Phase 3),
`Grpc.Tools` codegen, Akka.NET, `System.Threading.Channels`, xUnit + Shouldly.

---

## Scope (settled with the user 2026-07-23)

**In scope — migrate these four node→central observability channels** to the stream as four `oneof`
event kinds:

| DPS topic today | Message record | Central sink fed today |
|---|---|---|
| `alerts` | `AlarmTransitionEvent` | `IInProcessBroadcaster<AlarmTransitionEvent>` (+ `AlertHub`) |
| `script-logs` | `ScriptLogEntry` | `IInProcessBroadcaster<ScriptLogEntry>` (+ `ScriptLogHub`) |
| `driver-health` | `DriverHealthChanged` | `IDriverStatusSnapshotStore` (+ `DriverStatusHub`) |
| `driver-resilience-status` | `DriverResilienceStatusChanged` | `IDriverResilienceStatusStore` (no hub) |

**Explicitly deferred, with rationale (do NOT migrate in Phase 5):**

- **`redundancy-state`** — bidirectional, built from `Cluster.State`, **pair-local control-plane**
  that drives ServiceLevel + the Primary gate (consumed by `OpcUaPublishActor`,
  `ScriptedAlarmHostActor`, `DriverHostActor`, `HistorianAdapterActor`). It stays on DPS in both
  MeshTransport modes today, and under Phase 6 it is pair-local and works in-mesh. Central's *display*
  of each pair's redundancy is a Phase 6 cross-mesh concern (possibly a later added event kind), not a
  Phase 5 observability panel.
- **`fleet-status`** — **central-internal**: `FleetStatusBroadcaster` (admin singleton) builds it
  from the admin node's own cluster membership/reachability/leader events; `Fleet.razor` **polls the
  ConfigDB** and ignores the feed entirely. It is not a node→central stream, and its live UI path is
  already DB-polled. Revisit in Phase 6 when central loses gossip visibility of site nodes.
- **`deployment-acks`** — already rides the Phase 2 ClusterClient transport when
  `MeshTransport:Mode=ClusterClient`; it is a command-plane reply, not telemetry.

This narrowing mirrors how Phases 1 and 3 honestly scoped down from the program sketch. The program
doc's Phase 5 line lists all seven; this plan records the four that are genuinely live node→central
observability and defers the rest with reasons above. **Update the program doc + design §6.3 in
Task 10.**

## Direction & the dark switch (read before any task)

- **Node = server, central = client** (the load-bearing ScadaBridge inversion). Telemetry
  originates on driver nodes; central/admin consumes it. A fused `admin,driver` node both hosts (as
  driver) and dials (as admin) — it dials itself plus its pair peer, same as central dials site nodes.
- **`Telemetry:Mode` is read at startup; both code paths are compiled into every binary.** Flipping
  the flag is an appsettings/env change + restart — NOT a rebuild — exactly the Phase 2/3 dark-switch
  discipline (`OTOPCUA_CONFIG_MODE` on the rig). The node **always** hosts the gRPC server when
  `Telemetry:GrpcListenPort > 0` and **always** emits into the local hub AND publishes DPS, in both
  modes — so central can ingest from either side without a node redeploy. Only **central's ingest
  source** switches: `Dps` → today's four DPS SignalR bridges subscribe and feed the sinks; `Grpc` →
  those four bridges are NOT spawned and the dial supervisor feeds the identical sinks instead.
- **Auth from day one.** Reuse the fail-closed `FixedTimeEquals` bearer interceptor pattern
  (`ConfigServeAuthInterceptor` / `LocalDbSyncAuthInterceptor`). Shared node key
  `Telemetry:ApiKey` (serve side) == `TelemetryDial:ApiKey` (central side). This supersedes design
  §6.3's "unauthenticated for now" — ScadaBridge itself closed that gap with this same pattern.

## Reuse map (from recon 2026-07-23 — exact sites)

- **Proto + codegen:** add `telemetry.proto` beside `src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/deployment_artifact.proto`; register a `<Protobuf Include="Protos\telemetry.proto" GrpcServices="Both"/>` in `ZB.MOM.WW.OtOpcUa.Commons.csproj:29` (same block as the existing item). Generated types are shared by node (Runtime/Host) and central (AdminUI) from the one Commons reference.
- **Server hosting + Kestrel h2c:** the dedicated-listener block `Program.cs:434-533` and the `MapGrpcService` gate `Program.cs:577-580`. Add `telemetryListenPort` alongside `syncListenPort`/`configServeGrpcPort`.
- **Auth interceptor:** copy `src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ConfigServeAuthInterceptor.cs`; add to the shared `AddGrpc` pipeline at `Program.cs:425-431`.
- **Client dialing:** `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Deployment/GrpcDeploymentArtifactFetcher.cs` (channel-cache, Bearer metadata, h2c `GrpcChannel.ForAddress`, linked-CTS deadline).
- **Options + validator:** `ConfigSourceOptions`/`ConfigSourceOptionsValidator` in `src/Core/ZB.MOM.WW.OtOpcUa.Cluster/`; registered via `AddValidatedOptions` in `ServiceCollectionExtensions.AddOtOpcUaCluster` (`:32`).
- **Node discovery:** `CentralCommunicationActor.cs:198-219` already reads `ClusterNodes` (enabled, non-maintenance) selecting `NodeId, Host, AkkaPort` — extend the same query shape to `GrpcPort` for telemetry dial targets. `ClusterNode.GrpcPort` (nullable) already exists (`Entities/ClusterNode.cs:46`) explicitly for "the Phase 5 telemetry stream."
- **Central sinks (the untouched seam):** registered in `HubServiceCollectionExtensions.AddOtOpcUaDriverStatusServices` (`:29-35`); DPS bridges spawned in `WithOtOpcUaSignalRBridges` (`:53-83`).
- **ScadaBridge reference to mirror:** `SiteStreamGrpcServer.cs` (relay-actor + bounded `DropOldest` channel + lifecycle/cleanup + concurrency cap + max-stream-lifetime), `SiteStreamGrpcClient.cs` / `SiteStreamGrpcClientFactory.cs` (channel cache, keepalive), `SiteAlarmAggregatorActor.cs` (generation-stamped, budget-limited, self-healing reconnect), `ProtoContractTests.cs` (reflection-over-oneof contract lock).

---

## Tasks

### Task 0: Telemetry proto contract + codegen + contract-lock test

**Classification:** standard
**Estimated implement time:** ~5 min
**Parallelizable with:** Task 1, Task 2

**Files:**
- Create: `src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto`
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj` (add the `<Protobuf>` item next to line 29)
- Test: `tests/Core/ZB.MOM.WW.OtOpcUa.Commons.Tests/Protos/TelemetryProtoContractTests.cs` (create; match the existing Commons.Tests project layout — if there is no Commons.Tests project, add the test to the nearest existing Core test project that already references generated Commons types and note it)

**Step 1: Write `telemetry.proto`.** Package `telemetry.v1`, `csharp_namespace = ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1`. One server-streaming RPC + a `oneof` envelope with the four event kinds. Every enum carries a `*_UNSPECIFIED = 0` zero value; every field comment marks additive intent.

```proto
syntax = "proto3";
package telemetry.v1;
option csharp_namespace = "ZB.MOM.WW.OtOpcUa.Commons.Protos.Telemetry.V1";

import "google/protobuf/timestamp.proto";

// Central dials each driver node and opens Subscribe; the node streams its own live telemetry.
// Additive-only field evolution: never renumber/reuse a tag; a pre-field peer must read a new
// field's proto3 default correctly. Locked by TelemetryProtoContractTests.
service TelemetryStreamService {
  rpc Subscribe(TelemetryStreamRequest) returns (stream TelemetryEvent);
}

message TelemetryStreamRequest {
  string correlation_id = 1;   // safe Akka path element; becomes the relay actor name on the node
}

message TelemetryEvent {
  string correlation_id = 1;
  oneof event {
    AlarmTransition          alarm_transition  = 2;   // <- DPS topic "alerts"
    ScriptLog                script_log        = 3;   // <- DPS topic "script-logs"
    DriverHealth             driver_health     = 4;   // <- DPS topic "driver-health"
    DriverResilienceStatus   driver_resilience = 5;   // <- DPS topic "driver-resilience-status"
  }
}

// The four event bodies below mirror the C# records field-for-field. Encode the domain records'
// enums as proto enums (UNSPECIFIED=0), and DateTimes as google.protobuf.Timestamp.
message AlarmTransition        { /* fields mirroring Commons/Messages/Alerts/AlarmTransitionEvent.cs */ }
message ScriptLog             { /* fields mirroring Commons/Messages/Logging/ScriptLogEntry.cs */ }
message DriverHealth          { /* fields mirroring Commons/Messages/Drivers/DriverHealthChanged.cs */ }
message DriverResilienceStatus{ /* fields mirroring Commons/Messages/Drivers/DriverResilienceStatusChanged.cs */ }
```

The implementer MUST open each of the four domain records (paths in the scope table above) and
transcribe every field into the corresponding message with a stable tag number, choosing
`Timestamp`/`Int32`/`string`/`bool`/enum as the type dictates. Where a record enum exists, define a
matching proto enum with `_UNSPECIFIED = 0`. This is the contract — get it complete, because the
mapping tasks (7) and the lock test below depend on it.

**Step 2: Register codegen.** In the `.csproj`, next to the existing `deployment_artifact.proto` item:
```xml
<Protobuf Include="Protos\telemetry.proto" GrpcServices="Both" />
```

**Step 3: Write the contract-lock test (fails first).** Mirror ScadaBridge `ProtoContractTests.AllOneofVariants_HaveConversionHandlers`: reflect over `Enum.GetValues<TelemetryEvent.EventOneofCase>()` minus `None`, assert count- and membership-equality against a hand-maintained `HandledCases` array that Task 7's converter will also key off. This fails until the proto compiles and the array is filled.

```csharp
[Fact]
public void EveryOneofVariant_IsAccountedFor()
{
    var variants = Enum.GetValues<TelemetryEvent.EventOneofCase>()
        .Where(c => c != TelemetryEvent.EventOneofCase.None).ToArray();
    variants.ShouldBe(TelemetryProtoContract.HandledCases, ignoreOrder: true);
}
```
Add a tiny `TelemetryProtoContract.HandledCases` constant array in Commons (the single source both
this test and the Task-7 converter reference) listing the four cases.

**Step 4: `dotnet build ZB.MOM.WW.OtOpcUa.slnx`** — expect the generated types to appear; test goes green.

**Step 5: Commit.** `feat(mesh-phase5): telemetry.proto contract + oneof + contract-lock test`

---

### Task 1: Telemetry options (serve + dial) + validator + registration

**Classification:** small
**Estimated implement time:** ~4 min
**Parallelizable with:** Task 0, Task 2

**Files:**
- Create: `src/Core/ZB.MOM.WW.OtOpcUa.Cluster/TelemetryOptions.cs`
- Modify: `src/Core/ZB.MOM.WW.OtOpcUa.Cluster/ServiceCollectionExtensions.cs` (register in `AddOtOpcUaCluster`, ~lines 40-49 block)
- Test: `tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests/TelemetryOptionsValidatorTests.cs`

**Step 1:** Define two options classes and one validator, mirroring `ConfigSourceOptions`/`ConfigServeOptions` + `ConfigSourceOptionsValidator` (which reads roles from `IConfiguration`, NOT `IClusterRoleInfo`).

```csharp
public sealed class TelemetryOptions           // section "Telemetry" (serve side, node)
{
    public const string SectionName = "Telemetry";
    public string Mode { get; set; } = "Dps";  // Dps | Grpc  (central-ingest selector; harmless on node)
    public int GrpcListenPort { get; set; }     // 0 = disabled; driver node's telemetry h2c port
    public string ApiKey { get; set; } = "";    // shared node key; supply via ${secret:}/env
}

public sealed class TelemetryDialOptions        // section "TelemetryDial" (central)
{
    public const string SectionName = "TelemetryDial";
    public string Mode { get; set; } = "Dps";   // Dps | Grpc
    public string ApiKey { get; set; } = "";     // must equal the nodes' Telemetry:ApiKey
    public int ContactRefreshSeconds { get; set; } = 60;
    public int CallTimeoutSeconds { get; set; } = 30;
}
```

**Step 2: Validator rules** (`TelemetryOptionsValidator : IValidateOptions<TelemetryOptions>`, ctor-inject `IConfiguration`, read `Cluster:Roles` like `ConfigSourceOptionsValidator:98`):
- `Mode` must be `Dps` or `Grpc` (case-insensitive) — else Fail.
- A **driver-role** node with `Mode=Grpc` must have `GrpcListenPort > 0` — else Fail ("nothing to serve").
- `Mode=Grpc` with a non-empty role set must have a non-empty `ApiKey` (fail-closed: refuse to host an un-keyed telemetry surface) — else Fail.
- Add a sibling `TelemetryDialOptionsValidator`: `Mode` in {Dps,Grpc}; `Mode=Grpc` ⇒ `ApiKey` non-empty.

**Step 3: Register** via `AddValidatedOptions<TelemetryOptions, TelemetryOptionsValidator>(configuration, TelemetryOptions.SectionName)` and the dial equivalent, in `AddOtOpcUaCluster`.

**Step 4:** Tests — Grpc+driver+port0 fails; Grpc+empty-key fails; Dps passes with defaults; unknown Mode fails. Run `dotnet test tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests`.

**Step 5: Commit.** `feat(mesh-phase5): Telemetry/TelemetryDial options + fail-closed validators`

---

### Task 2: Node-local in-process telemetry hub

**Classification:** standard
**Estimated implement time:** ~5 min
**Parallelizable with:** Task 0, Task 1

**Files:**
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs`
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs`
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs`

**Design:** a process-wide singleton carrying **this node's own** telemetry (never cluster-wide — that
is what keeps it correct on both the current single mesh and the Phase-6 pair mesh; DPS would leak
peers' events). It holds a **domain-typed** union (not proto) so the publish seams stay decoupled from
generated types. Two responsibilities:

1. `Emit(TelemetryItem)` — fan the item to every currently-subscribed writer through **bounded
   per-subscriber channels with `FullMode = DropOldest`** (live view is lossy under backpressure by
   design; a slow central never blocks the node).
2. **Snapshot replay for last-value channels.** Keep a last-value cache for `DriverHealth` (keyed by
   driver instance) and `DriverResilienceStatus` (keyed by `(instance, host)`); `alerts`/`script-logs`
   are append logs and are NOT cached. On `Subscribe`, first drain the cached snapshots to the new
   writer, then attach it for live deltas — the simplified equivalent of ScadaBridge's seed-then-stream,
   so a central reconnect immediately re-primes the driver-health/resilience stores.

```csharp
public abstract record TelemetryItem
{
    public sealed record Alarm(AlarmTransitionEvent E)               : TelemetryItem;
    public sealed record Script(ScriptLogEntry E)                    : TelemetryItem;
    public sealed record Health(DriverHealthChanged E)               : TelemetryItem;
    public sealed record Resilience(DriverResilienceStatusChanged E) : TelemetryItem;
}

public interface ITelemetryLocalHub
{
    void Emit(TelemetryItem item);
    // Returns a reader that first yields cached snapshots, then live deltas; disposing the
    // subscription detaches and completes the channel.
    ITelemetrySubscription Subscribe(int boundedCapacity);
}
```

`TelemetryLocalHub` uses a `ConcurrentDictionary<Guid, Channel<TelemetryItem>>` of subscribers +
two `ConcurrentDictionary` snapshot caches. `Emit` updates the snapshot cache (for Health/Resilience)
then `TryWrite`s to each subscriber (drop-oldest handled by the bounded channel). `Subscribe` snapshots
the caches into the new channel under a brief lock ordering guarantee (cache-then-attach) so no delta
is lost across the attach boundary.

**Register** as a singleton in the driver-role DI branch (Task 3 wires the producers; do the
`AddSingleton<ITelemetryLocalHub, TelemetryLocalHub>()` here in the Runtime `ServiceCollectionExtensions`
`hasDriver` path, or wherever driver-role Runtime services register — locate and match).

**Tests:** Emit-before-Subscribe is not seen except via snapshot (Health/Resilience last-value IS seen;
Alarm/Script are not); two subscribers each get their own copy; a full channel drops oldest not newest;
dispose detaches. Run the Runtime.Tests subset.

**Step 5: Commit.** `feat(mesh-phase5): node-local telemetry hub (snapshot-replay + drop-oldest fan-out)`

---

### Task 3: Tap the four publish seams into the hub

**Classification:** standard
**Estimated implement time:** ~5 min
**Parallelizable with:** none (needs Task 2)

**Files (all Modify) — add a hub `Emit` beside the existing DPS `Publish`, leaving DPS intact:**
- `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs:36` (driver-health)
- `src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs:80` (resilience)
- `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs:379` + `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs:1405` (alerts — two producers)
- `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs:261` + `src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Scripting/DpsScriptLogPublisher.cs:46` (script-logs — two producers)
- Test: extend the nearest existing publisher tests, or add `tests/.../Telemetry/PublishSeamEmitsToHubTests.cs`

**Approach:** inject `ITelemetryLocalHub` into each producer (constructor for the services/publishers;
for actors, pass via `Props` — match how each already receives dependencies). At each seam, immediately
after the existing `Mediator.Tell(new Publish(...))`, add `_telemetryHub.Emit(new TelemetryItem.Xxx(msg))`.
**Do NOT remove or gate the DPS publish** — it is the `Dps`-mode path and must remain unconditional so
the switch stays pure. The hub is a no-op sink until a client subscribes, so this is safe on every node
regardless of mode.

For actors constructed where a hub isn't readily resolvable, resolve the singleton once at spawn and
thread it through `Props` (do not `DependencyResolver` inside the actor per-message). Where a producer
is only present on driver-role nodes, the hub singleton is guaranteed registered (Task 2).

**Tests:** each producer, when driven, results in exactly one `hub.Emit` of the correct `TelemetryItem`
subtype carrying the same payload it published to DPS. Use a fake `ITelemetryLocalHub` capturing emits.

**Step 5: Commit.** `feat(mesh-phase5): tap the 4 telemetry publish seams into the local hub (DPS intact)`

---

### Task 4: Node-side gRPC streaming service

**Classification:** high-risk
**Estimated implement time:** ~5 min (mirror ScadaBridge closely)
**Parallelizable with:** none (needs Task 0, Task 2)

**Files:**
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryStreamGrpcService.cs` (place beside `DeploymentArtifactService.cs`; alias the generated base to avoid the name collision, as that file does at its lines 9-10)
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMap.Node.cs` (domain→proto mapping; the reverse of Task 7)
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Grpc/TelemetryStreamGrpcServiceTests.cs`

**Design — port `SiteStreamGrpcServer.RunSubscriptionStreamAsync` (`.cs:253-385`) shape:**

```csharp
public sealed class TelemetryStreamGrpcService : GeneratedTelemetryBase
{
    // ctor: ITelemetryLocalHub hub, ILogger, IOptions<TelemetryOptions> (for a GrpcMaxConcurrentStreams knob)
    public override async Task Subscribe(TelemetryStreamRequest request,
        IServerStreamWriter<TelemetryEvent> responseStream, ServerCallContext context)
    {
        // 1. Validate correlation_id is a safe id (it labels the subscription; reject empty/oversized).
        // 2. Concurrency cap (default 100) -> throw RpcException(ResourceExhausted) when exceeded.
        // 3. sub = _hub.Subscribe(boundedCapacity: 1000);   // snapshot-then-live, DropOldest inside the hub
        // 4. MaxStreamLifetime linked-CTS CancelAfter (default 4h) ORed with context.CancellationToken,
        //    so a zombie stream terminates even if h2c keepalive misses it.
        // 5. await foreach (item in sub.Reader.ReadAllAsync(ct))
        //        await responseStream.WriteAsync(TelemetryProtoMap.ToProto(item, request.CorrelationId), ct);
        // 6. finally: sub.Dispose(); balance any opened/closed gauge; swallow OperationCanceled as normal.
    }
}
```

Key correctness points (from the ScadaBridge recon — do not skip):
- The **bounded DropOldest channel lives in the hub** (Task 2), so the service just pumps its reader —
  a slow/blocked central cannot back-pressure the node's actor threads.
- Wrap the whole body so a client disconnect (`RpcException`/`OperationCanceledException`) exits cleanly
  and disposes the subscription; never let it bubble as a fault.
- No readiness race: the hub is a plain singleton available at host build, so unlike ScadaBridge's
  `SetReady(ActorSystem)` gate there is nothing to defer — but if the driver actor system isn't up yet
  the hub simply has no snapshots and no deltas, which is fine.

**`TelemetryProtoMap.ToProto`**: switch on `TelemetryItem` subtype → build the matching proto message,
wrap in `TelemetryEvent { CorrelationId, <oneof> = ... }`. Transcribe every field (the reverse of the
`.proto` transcription in Task 0). DateTimes → `Timestamp.FromDateTime(utc)`.

**Tests:** in-memory — subscribe, emit each of the four item kinds into the hub, assert the service
writes the correct proto event with fields intact; a cancelled `ServerCallContext` ends the stream and
disposes; the concurrency cap throws `ResourceExhausted` past the limit. (Full two-host wire test is
covered by the Task-12 live gate + optionally a Task-8 boundary test.)

**Step 5: Commit.** `feat(mesh-phase5): node-side TelemetryStreamService (hub -> server-streaming)`

---

### Task 5: Telemetry stream auth interceptor

**Classification:** small
**Estimated implement time:** ~3 min
**Parallelizable with:** none (needs Task 1)

**Files:**
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/TelemetryStreamAuthInterceptor.cs` (copy `ConfigServeAuthInterceptor.cs` verbatim, change the gated prefix + options source)
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/TelemetryStreamAuthInterceptorTests.cs`

**Steps:**
- `ServicePrefix = "/telemetry.v1.TelemetryStreamService/"`; read `IOptions<TelemetryOptions>.ApiKey`.
- Override all four handler kinds (especially **ServerStreaming** — that is the RPC shape here) so the
  gate holds; non-matching paths pass through; empty key ⇒ throw (fail-closed); `FixedTimeEquals` on the
  Bearer token; reject with `PermissionDenied`.
- **Keep exactly one public constructor** (the ScadaBridge recon flagged that `Grpc.AspNetCore` silently
  stops authorizing an interceptor with >1 ctor — pin it with a reflection test).

**Tests:** right key passes; wrong key `PermissionDenied`; empty configured key rejects all; a call to a
different service path passes through.

**Step 5: Commit.** `feat(mesh-phase5): fail-closed bearer interceptor for the telemetry stream`

---

### Task 6: Kestrel wiring + map the node telemetry server

**Classification:** standard
**Estimated implement time:** ~4 min
**Parallelizable with:** none (needs Task 1, Task 4, Task 5)

**Files:**
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs` (the `AddGrpc` block 425-431, the dedicated-listener block 434-533, and the map block 577-580)
- Test: none new (covered by Task 12 live gate); build must stay green

**Steps — extend, do not rewrite, the existing block:**

1. **Interceptor registration** (`Program.cs:425-431`): inside the `AddGrpc(o => …)`, add
   `if (hasDriver) o.Interceptors.Add<TelemetryStreamAuthInterceptor>();` (alongside the LocalDbSync one —
   both are driver-side, both path-scoped, harmless when their service is unmapped).

2. **Port resolution** (near `:453-454`):
   ```csharp
   var telemetryListenPort = hasDriver ? builder.Configuration.GetValue<int>("Telemetry:GrpcListenPort") : 0;
   ```
   Add `|| telemetryListenPort > 0` to the block guard at `:455`.

3. **Bind it** inside the `ConfigureKestrel` closure (`:517-526`): capture `var telemetryPortToBind = telemetryListenPort;` beside the other two, and add
   `if (telemetryPortToBind > 0) kestrel.ListenAnyIP(telemetryPortToBind, o => o.Protocols = HttpProtocols.Http2);`.

4. **HTTPS-refuse branch** (`:500-511`): set `telemetryListenPort = 0;` too, and add it to the log
   message (all dedicated listeners disable together when the host serves HTTPS).

5. **Map the service** (after `:580`):
   ```csharp
   if (hasDriver && telemetryListenPort > 0)
       app.MapGrpcService<TelemetryStreamGrpcService>();
   ```

6. Update the block's header comment (434-452) to mention the third dedicated port.

**Verify:** `dotnet build`; a fused node with all three ports set binds all three exactly once (the
"re-bind existing surface once" invariant already holds — the new port is just one more
`ListenAnyIP`). A driver-only node with only `Telemetry:GrpcListenPort` set binds only that.

**Step 5: Commit.** `feat(mesh-phase5): host the telemetry gRPC server on driver nodes (dedicated h2c port)`

---

### Task 7: Central per-node dialer client + proto→domain mapping

**Classification:** high-risk
**Estimated implement time:** ~5 min
**Parallelizable with:** none (needs Task 0)

**Files:**
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryStreamClient.cs`
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMap.Central.cs` (proto→domain; reverse of Task 4; keyed off `TelemetryProtoContract.HandledCases`)
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryProtoMapTests.cs`

**Design — mirror `SiteStreamGrpcClient.cs` + `GrpcDeploymentArtifactFetcher` channel handling:**

```csharp
public sealed class TelemetryStreamClient : IDisposable
{
    // ctor: string endpoint (http://host:port), string apiKey, IClientFactory seam for tests
    // - GrpcChannel.ForAddress(endpoint) over http:// => prior-knowledge h2c; cache one channel per endpoint.
    // - HTTP/2 keepalive: PingDelay 15s / Timeout 10s / Always (SocketsHttpHandler on the channel).
    public async Task RunAsync(string correlationId, Action<TelemetryItem> onEvent,
                               Action<Exception> onError, CancellationToken ct)
    {
        // headers: authorization: Bearer {apiKey}
        // using call = client.Subscribe(new TelemetryStreamRequest{CorrelationId=correlationId}, headers, ct)
        // await foreach (evt in call.ResponseStream.ReadAllAsync(ct)) onEvent(TelemetryProtoMap.ToDomain(evt));
        // RpcException(Cancelled) on shutdown => normal; anything else => onError (the reconnect trigger).
    }
}
```

`TelemetryProtoMap.ToDomain(TelemetryEvent)` switches on `evt.EventCase` and reconstructs the domain
record (`AlarmTransitionEvent` / `ScriptLogEntry` / `DriverHealthChanged` / `DriverResilienceStatusChanged`).
It MUST cover exactly `TelemetryProtoContract.HandledCases` — a `default: throw` on an unknown case makes
"new event added but not mapped" a loud runtime failure, and the Task-0 lock test makes it a compile-time-ish
guard.

**Tests:** round-trip each domain record → `TelemetryProtoMap.ToProto` (Task 4) → `ToDomain` → assert
field-equality across all four kinds and all enum values (wire-fidelity, like ScadaBridge
`ProtoRoundtripTests`); an unmapped `EventCase` throws.

**Step 5: Commit.** `feat(mesh-phase5): central telemetry dialer client + proto<->domain round-trip`

---

### Task 8: Central dial supervisor actor (discovery + reconnect + feed sinks)

**Classification:** high-risk
**Estimated implement time:** ~5 min
**Parallelizable with:** none (needs Task 1, Task 7)

**Files:**
- Create: `src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryDialSupervisor.cs`
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Telemetry/TelemetryDialSupervisorTests.cs`

**Design — an admin-role actor holding one dialer per driver node, feeding the existing central sinks:**

- **Discovery:** on `PreStart` and every `TelemetryDialOptions.ContactRefreshSeconds` (+ on an
  admin-change signal if one is already wired for `CentralCommunicationActor`), read enabled,
  non-maintenance `ClusterNode` rows selecting `NodeId, Host, GrpcPort` (extend the
  `CentralCommunicationActor.cs:198-219` query). A row with `GrpcPort == null` is skipped with a Warning
  (the node exposes no telemetry port — honest, not an error). Build `http://{Host}:{GrpcPort}` dial
  targets; add dialers for new nodes, stop dialers for removed ones.
- **Per-node dialer loop** (mirror `SiteAlarmAggregatorActor.OpenGrpcStream`/`HandleGrpcError`,
  simplified — one node per dialer, no NodeA/NodeB flip since each `ClusterNode` row is its own node):
  - Open `TelemetryStreamClient.RunAsync`; `onEvent` posts each `TelemetryItem` to the supervisor which
    routes to the sink (below); `onError` schedules a reconnect.
  - **Reconnect:** immediate first retry, then fixed `_reconnectDelay` (5s) backoff. A **monotonic
    generation stamp** per node makes late errors from a superseded stream ignorable. The dialer does
    **not** die on repeated failure — it keeps retrying (observability, not data plane); log the first
    failure + every Nth.
  - A `correlationId` per (node, generation) — a safe id string, e.g. `central-{NodeId}-{gen}`.
- **Feed the sinks (the untouched seam):** inject the four singletons and route by item type:
  - `Alarm` → `IInProcessBroadcaster<AlarmTransitionEvent>.Publish(e)`
  - `Script` → `IInProcessBroadcaster<ScriptLogEntry>.Publish(e)`
  - `Health` → `IDriverStatusSnapshotStore.Upsert(e)`
  - `Resilience` → `IDriverResilienceStatusStore.Upsert(e)`
- **Connection indicator:** drive the two broadcasters' `SetConnected(...)` off aggregate stream health
  — connected when **≥1** node stream is up, disconnected when all are down — matching today's
  DPS `SubscribeAck`/`PostStop` pill semantics on `/alerts` and `/script-log`. (The two store-backed
  panels have no connection flag today; leave their per-row staleness as-is — noted for Task 10 docs.)

**Registration** is done in Task 9 (mode-gated). This task only builds + unit-tests the actor with a
fake `TelemetryStreamClient` factory and fake sinks.

**Tests:** discovery adds/removes dialers as the (fake) `ClusterNode` set changes; a null `GrpcPort`
row is skipped; an emitted item of each kind lands on the correct sink; a stream error triggers a
scheduled reconnect with an incremented generation; a late event from a superseded generation is
dropped; broadcaster `IsConnected` flips true on first up / false on all-down.

**Step 5: Commit.** `feat(mesh-phase5): central telemetry dial supervisor (discover + reconnect + feed sinks)`

---

### Task 9: Wire the dark switch (central ingest source)

**Classification:** standard
**Estimated implement time:** ~4 min
**Parallelizable with:** none (needs Task 8)

**Files:**
- Modify: `src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/HubServiceCollectionExtensions.cs` (`WithOtOpcUaSignalRBridges`, lines 53-83)
- Modify: the AdminUI Akka configurator call site that invokes `WithOtOpcUaSignalRBridges` (locate; it's inside the `hasAdmin` branch per the method's own doc example) — pass or resolve the mode
- Test: `tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Hubs/TelemetryModeWiringTests.cs`

**Steps:**
- Inside `WithActors`, resolve `IOptions<TelemetryDialOptions>` and read `Mode`.
- **`Mode == Dps` (default):** spawn all five bridges exactly as today (unchanged behaviour).
- **`Mode == Grpc`:** spawn the **fleet-status bridge only** (out of Phase 5 scope, stays on DPS), and
  do **NOT** spawn the four telemetry DPS bridges (alert, script-log, driver-status, resilience).
  Instead spawn `TelemetryDialSupervisor` (Task 8), resolving the four sink singletons + the
  `IDbContextFactory<OtOpcUaConfigDbContext>` (admin nodes have ConfigDb) + `IOptions<TelemetryDialOptions>`.
- The sinks (`AddOtOpcUaDriverStatusServices`, `:29-35`) are registered identically in both modes — no
  change there; that is the whole point of the swap.

**Guard rails:**
- Keep the fleet-status bridge on DPS in both modes (deferred).
- A `Grpc`-mode admin node still needs the sinks registered (it does — same call), so the AdminUI
  components resolve them regardless.

**Tests:** with `Mode=Dps`, the registry has the four telemetry bridge keys and no supervisor; with
`Mode=Grpc`, it has the supervisor and the fleet-status bridge but none of the four telemetry bridge
keys. (Use the Akka.Hosting `ActorRegistry` + a TestKit or the existing bridge-registration test
harness if one exists.)

**Step 5: Commit.** `feat(mesh-phase5): dark-switch central telemetry ingest (Dps bridges | Grpc dialer)`

---

### Task 10: Docs — Telemetry section, supersede §6.3, program status

**Classification:** small
**Estimated implement time:** ~4 min
**Parallelizable with:** Task 11 (disjoint files)

**Files (Modify/Create):**
- Create: `docs/Telemetry.md` — the stream architecture (direction, dark switch, auth, the four channels, deferred three + why, reconnect story, per-panel connection indicator status).
- Modify: `docs/Configuration.md` — add `Telemetry` + `TelemetryDial` sections (keys table, the `Mode` dark switch, port + shared-key `${secret:}` guidance).
- Modify: `docs/Redundancy.md` — a short "Command/observability transport" cross-ref noting redundancy-state stays DPS (pair-local) and telemetry moved to gRPC.
- Modify: `docs/plans/2026-07-21-per-cluster-mesh-design.md` §6.3 — supersede note: "telemetry stream authenticated from day one (fail-closed bearer), superseding the earlier 'unauthenticated for now'; ScadaBridge closed the same gap identically."
- Modify: `docs/plans/2026-07-22-per-cluster-mesh-program.md` — Phase 5 section + Tracking row: record the four-channel scope, the three deferrals + rationale, the dark switch, auth-from-day-one.
- Modify: `CLAUDE.md` — a "Live telemetry transport (`Telemetry`/`TelemetryDial`)" section mirroring the MeshTransport/ConfigSource sections (default `Dps`, node hosts server, central dials, four channels, deferred three, fail-closed key).

**Step 5: Commit.** `docs(mesh-phase5): telemetry stream — Telemetry.md, config, supersede §6.3, program status`

---

### Task 11: Rig config — telemetry ports + keys + flip env

**Classification:** small
**Estimated implement time:** ~4 min
**Parallelizable with:** Task 10 (disjoint files); needs Task 6 + Task 9 landed for the flip to mean anything

**Files:**
- Modify: `docker-dev/docker-compose.yml`
- Modify: any committed `appsettings.json` defaults that must carry the (default-OFF) keys — verify none are needed since `Mode` defaults to `Dps` and port defaults to 0.

**Steps:**
- Give every **driver-role** node (central-1/2 fused + the four site nodes) `Telemetry__GrpcListenPort`
  (pick a non-colliding port — verify against 4053 Akka / 4055 ConfigServe / the LocalDb sync port; use
  e.g. **4056**) and `Telemetry__ApiKey: "telemetry-docker-dev-key"` (committed dev-secret exception, like
  `configserve-docker-dev-key`).
- Give the **central (admin)** nodes `TelemetryDial__ApiKey: "telemetry-docker-dev-key"` (matching).
- Add each driver node's `GrpcPort` to its `ClusterNode` seed row (the rig's SQL seed / migration seed) so
  central discovers the telemetry endpoint — mirror how `AkkaPort` was seeded in Phase 1.
- Leave `Telemetry__Mode`/`TelemetryDial__Mode` unset (⇒ `Dps` default). Document the flip in the compose
  header: set `Telemetry__Mode=Grpc` on the driver nodes **and** `TelemetryDial__Mode=Grpc` on the central
  nodes at `docker compose up`, then recreate.
- `lmxopcua-fix`/rig note: this is the local `docker-dev` rig, not the shared driver-fixture host.

**Step 5: Commit.** `chore(mesh-phase5): rig telemetry ports + shared key + ClusterNode.GrpcPort seed`

---

### Task 12: Live gate

**Classification:** high-risk (gate, not a code change — but the phase is not done until it passes)
**Estimated implement time:** live rig run (not subagent wall-time)
**Parallelizable with:** none (needs everything)

**Files:**
- Create: `docs/plans/2026-07-23-mesh-phase5-live-gate.md` (record)

**Procedure (on the local `docker-dev` rig, matching prior phases' gates):**
1. Build the Phase-5 image (`--platform linux/amd64` build stage; `--progress=plain`) and recreate the rig in **default `Dps` mode** — confirm all six nodes up, `/alerts`, `/script-log`, `/hosts` panels live exactly as before (regression check: DPS path unchanged).
2. Flip: set `Telemetry__Mode=Grpc` on all driver nodes and `TelemetryDial__Mode=Grpc` on the central nodes; `docker compose up` (recreate).
3. **Exit gate leg A — panels green over gRPC:** drive a scripted-alarm transition and a script-log emission and a driver health/resilience change; confirm `/alerts`, `/script-log`, and the `/hosts` driver table update live — **with the four DPS telemetry bridges NOT spawned** (grep the central logs to confirm the Grpc branch ran and the four bridge keys are absent). Confirm the `/alerts` + `/script-log` connection pill reads "live".
4. **Exit gate leg B — kill-and-reconnect:** `docker kill` a site driver node (or drop its telemetry port), watch the central dialer log the error + retries and the pill flip to disconnected; restart the node and confirm the dialer reconnects, the driver-health/resilience stores re-prime from the node hub's snapshot replay, and the pill returns to live — **every stream recovers**.
5. **Optionally** confirm the auth gate: a dial with a wrong `TelemetryDial:ApiKey` gets `PermissionDenied` and no panel data (a deliberate misconfig probe).
6. Record all legs + evidence in the gate doc; note any deviations (as Phases 1-4 gates did).

**Exit gate (program):** all AdminUI live panels green against the rig with the telemetry DPS bridges
off; kill-and-reconnect of the central dialer recovers every stream.

**Step: Commit** the gate record; then finish per superpowers-extended-cc:finishing-a-development-branch
(merge `feat/mesh-phase5` to master locally, push, update the scadaproj umbrella index, update memory).

---

## Risks / watch-items carried into execution

- **Single-mesh correctness of the node-local hub.** The hub must carry only *this node's* events
  (Task 2) — never subscribe it to cluster DPS, or on the current single mesh every node would stream
  every other node's events and central would double-count. Snapshot replay + drop-oldest come from the
  hub, not the service.
- **Do not gate/remove the DPS publish** at the four seams (Task 3) — that is the `Dps`-mode path and
  the pure-dark-switch guarantee. The hub is a no-op until a client connects.
- **Interceptor single-ctor pin** (Task 5) — `Grpc.AspNetCore` silently disables an interceptor with
  >1 public ctor.
- **Kestrel "re-bind once" invariant** (Task 6) — the third dedicated port is just one more
  `ListenAnyIP`; the existing-surface re-bind still happens exactly once. Don't duplicate it.
- **Frame size is a non-issue** — server-streaming telemetry events are small and never cross Akka
  remoting; the 128 KB ClusterClient frame trap does not apply to this path.
- **Additive-only proto evolution** — new event kinds/fields only; the Task-0 lock test + the
  Task-7 `default: throw` are the guard. Never renumber a tag.
- **Deferred-channel honesty** — redundancy-state / fleet-status / deployment-acks are out by design,
  documented in Task 10; a future reviewer must not read "seven topics" in the program doc and think
  Phase 5 missed three.