Push-driven team state via team_changed + consumer-loop resilience (#65)

* fix: keep event consumers alive across transient handler failures

Three faults observed live tonight, all the same class: an exception escaping
into a consumer's await-foreach ends that subscription for the rest of the
process, so the feature it drives goes silently dead until the bot restarts.

1. "Map connection-status loop faulted. InvalidOperationException: GetMap
   returned no data." — ConnectionSupervisor.GetMonumentsAsync forwarded the
   connection-level call (documented "Throws on failure") straight through a
   query seam whose own contract promised "or an empty list". It was the sole
   query in RustPlusSocketSource without a degrade-to-null/empty catch. It now
   catches, logs and returns [], and the failure message names the Rust+ error
   code instead of a bare "returned no data".

2. "ConnectionStatusChanged consumer faulted. TimeoutException" from a Discord
   REST edit — the Workspace consumer died, leaving the info channels stale.

3. The same shape existed, unguarded, in 21 consumer loops across 11 services.

Add EventBusConsumption.ConsumeAsync<TEvent>(handler, onHandlerFailure, ct):
it subscribes, guards each handler call, reports failures and keeps consuming;
only cancellation of the loop token ends the subscription. Abstractions stays
dependency-free (the callback is Action<Exception>, so each service keeps its
own logger). Every vulnerable consumer now routes through it; each service's
outer try/catch stays as a backstop. ClansHostedService and
CommandsHostedService already isolate per event inline and are left alone —
migrating them would lose the guild/server ids in their failure logs.

Also fixes the unrelated log line that started this: Rust+ sends the token
"apartmentcomplex", which no entry in MonumentTokenMap mapped, so apartment
complexes rendered with no icon ("No monument icon for token ...; mapped type:
null"). MonumentType.ApartmentsComplex and Apartments_Complex.svg were already
in RustMapsApi.Assets — only the token table had not caught up.

Tests: EventBusConsumptionTests (a failing handler costs its own event only;
cancellation still ends the subscription), WorkspaceConsumerResilienceTests and
MapHostedServiceTests (both written failing first, reproducing the live
exceptions), plus a monuments-seam and a token-map case.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(connections): use a query-specific log for the monuments seam

Copilot review: GetMonumentsAsync (the query seam used by map rendering) reused
LogMonumentsFetchFailed, whose message says "rig detection disabled for this
connection" — misleading for a non-rig caller. Give the seam its own generic
LogMonumentsQueryFailed and keep the rig-specific message on the rig path.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* test(map): stop the status-loop regression test flaking on slow CI

AdvancingClock jumped 1 minute per read while MapRefreshInterval is 30 minutes,
so MapRefreshThrottle gated every refresh after the first: the counter only
re-incremented once ~30 clock reads accumulated. Fast enough locally, but on
Windows CI fewer than 30 events were consumed inside the 5s deadline, so calls
stuck at 1 and the assert failed. Advance an hour per read (past the interval)
so the throttle passes every refresh and each event reaches the counter.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(spec): push-driven team state via team_changed (beta.6)

Design for moving team-state detection (connect/disconnect/death/respawn/
AFK) from the 2-5s poll onto RustPlusApi beta.6's OnTeamChanged event,
keeping a 30s team poll as an AFK safety tick. Marker/rig polling stays.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(plan): implementation plan for team_changed push migration

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* build(deps): bump RustPlusApi to 2.0.0-beta.6 (team_changed dispatch)

* refactor(connections): extract shared TeamInfoMapping.ToSnapshot

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(connections): surface team_changed as IRustServerConnection.TeamChanged

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(connections): add TeamPollInterval (AFK safety-tick cadence)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(connections): drive team state from team_changed push + slow AFK tick

* test(connections): cover AFK-during-silence; harden team_changed handler; docs

* style(connections): apply ReSharper formatting to new team tests

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: HandyS11

Directory.Packages.props

@@ -15,8 +15,8 @@
     <PackageVersion Include="Persistord.Core" Version="1.0.0-beta2" />
     <PackageVersion Include="RustMapsApi" Version="1.0.0-beta.3" />
     <PackageVersion Include="RustMapsApi.Assets" Version="1.0.0-beta.3" />
-    <PackageVersion Include="RustPlusApi" Version="2.0.0-beta.5" />
-    <PackageVersion Include="RustPlusApi.Fcm" Version="2.0.0-beta.5" />
+    <PackageVersion Include="RustPlusApi" Version="2.0.0-beta.6" />
+    <PackageVersion Include="RustPlusApi.Fcm" Version="2.0.0-beta.6" />
     <PackageVersion Include="Serilog.Extensions.Hosting" Version="10.0.0" />
     <PackageVersion Include="Serilog.Settings.Configuration" Version="10.0.1" />
     <PackageVersion Include="Serilog.Sinks.Console" Version="6.1.1" />

docs/superpowers/plans/2026-07-24-team-changed-push.md

@@ -0,0 +1,140 @@
+# Push-driven team state via `team_changed` (RustPlusApi beta.6)
+
+**Date:** 2026-07-24
+**Status:** Approved design
+**Branch:** `fix/consumer-loop-resilience` (or a dedicated follow-up branch)
+
+## 1. Problem & motivation
+
+Team-state detection (connect / disconnect / death / respawn / AFK) is currently
+derived by **polling** `GetTeamInfoAsync` inside `PollMarkersAsync`
+(`ConnectionSupervisor.cs:761`), on the marker cadence of every 2–5 s
+(`MarkerPollInterval` = 5 s, `MarkerPollFastInterval` = 2 s).
+
+Two problems motivate a change:
+
+1. **Efficiency / latency.** A `getTeamInfo` request every 2–5 s per connected
+   server is wasteful, and presence/death events lag by up to one poll interval.
+2. **Log noise.** RustPlusApi `beta.5` does not dispatch the `team_changed`
+   broadcast — its `ParseNotification` falls through to
+   `Logger.LogUnknownBroadcast`, producing recurring
+   `WRN Unknown broadcast received: RustPlusContracts.AppBroadcast`. `team_changed`
+   is the *only* unhandled broadcast field in `beta.5`, so every one of these
+   warnings is a dropped team update.
+
+RustPlusApi **beta.6** now parses `team_changed` and raises a new
+`OnTeamChanged` event carrying `TeamChangedEventArg { PlayerId, TeamInfo }`,
+where `TeamInfo` is produced by the **same `ToTeamInfo()` mapping** the polled
+`GetTeamInfoAsync` already uses. Full field parity (members with
+`X/Y/IsOnline/IsAlive/LastSpawnTime/LastDeathTime`, `LeaderSteamId`, `DeathNote`).
+
+## 2. Scope
+
+**In scope:** upgrade `beta.5 → beta.6`; move team-state detection from the fast
+poll to the `OnTeamChanged` push event; retain a low-frequency team poll purely as
+an AFK safety tick.
+
+**Explicitly out of scope:** marker/rig polling. The Rust+ protocol has **no
+broadcast** for cargo-ship / patrol-heli / chinook / travelling-vendor markers;
+the server only answers `getMapMarkers` on request. `PollMarkersAsync`'s marker
+and rig-activation work is unchanged. This is therefore a *partial* migration —
+"move team state to push", not "replace the polling system" wholesale.
+
+## 3. Architecture
+
+### 3.1 Connection layer — `IRustServerConnection` / `RustPlusSocketSource`
+
+- Add to the interface:
+  `event EventHandler<TeamInfoSnapshot>? TeamChanged;`
+- `RustPlusSocketSource`:
+  - Subscribe `_rustPlus.OnTeamChanged += OnTeamChanged` in the ctor event block
+    (near `RustPlusSocketSource.cs:172`); unsubscribe in `DisposeAsync`
+    (near line 768), symmetrically with the existing handlers.
+  - Extract the inline `TeamInfo → TeamInfoSnapshot` mapping currently in
+    `GetTeamInfoAsync` (lines 339–353) into a private static
+    `ToSnapshot(RustPlusApi.Data.TeamInfo)` helper. Both `GetTeamInfoAsync` and
+    the new handler reuse it:
+    `private void OnTeamChanged(object? s, TeamChangedEventArg e) => TeamChanged?.Invoke(this, ToSnapshot(e.TeamInfo));`
+  - Because the event and the poll share one mapping, there is zero behavioural
+    drift between pushed and polled team snapshots.
+- The test fake `IRustServerConnection` gains the `TeamChanged` event plus a test
+  hook to raise it.
+
+### 3.2 Supervisor layer — `ConnectionSupervisor`
+
+- **Push handler.** Wire an `OnTeamChanged` handler in `RunAsync` alongside the
+  existing `OnTeamMessage` / `OnClanChanged` wiring (~line 609). On each event it
+  calls `tracker.Diff(snapshot, clock.UtcNow, _options.AfkThreshold,
+  _options.AfkEpsilon)` and, when transitions are non-empty, fire-and-forget
+  publishes `PlayerStateChangedEvent` using the same `_ = PublishAsync(...)`
+  error-isolation pattern as the other handlers. Subscribe on setup, unsubscribe
+  in the `finally` (lines 655–659 block).
+- **Slow team poll (AFK tick + self-heal).** Split the team block (lines 761–768)
+  out of `PollMarkersAsync` into a new `PollTeamAsync` loop that runs the *same*
+  `Diff` + publish at a new `TeamPollInterval` (default 30 s). Its first iteration
+  runs immediately (to prime the baseline promptly), then it waits
+  `TeamPollInterval` between iterations. Launched as a fourth `Task.Run` next to
+  `markerPoll` / `reachabilityPoll` (line 624) and joined in the same
+  `Task.WhenAll` on exit (line 646).
+- **`PollMarkersAsync`** loses its `tracker` parameter and the team block; markers
+  and rig-activation detection are otherwise untouched.
+- **Shared `dims`.** `PlayerStateChangedEvent` needs `MapDimensions?`. `dims` stays
+  resolved lazily *off the critical connect path* (preserving the non-blocking
+  behaviour the comment at lines 731–735 protects), stored in a per-connected-window
+  holder that both the push handler and `PollTeamAsync` read. Events arriving before
+  `dims` resolves publish `dims: null`, which `PlayerStateChangedEvent` already
+  tolerates (nullable → renders without a grid reference).
+
+### 3.3 Options — `ConnectionOptions`
+
+- Add `public TimeSpan TeamPollInterval { get; set; } = TimeSpan.FromSeconds(30);`
+
+## 4. Concurrency & correctness
+
+- `TeamStateTracker.Diff` is now invoked from two threads: the RustPlusApi dispatch
+  thread (push) and the `PollTeamAsync` thread (tick). `TeamStateTracker` is already
+  guarded by `_gate` and swaps `_baseline` atomically, so concurrent calls serialize
+  and never corrupt shared state. Ordering is NOT guaranteed, though:
+  `PollTeamAsync` captures its snapshot before an `await` spanning a network
+  round-trip, so a concurrent push can commit a newer baseline first and the poll's
+  stale snapshot may then emit a rare, self-healing spurious presence transition
+  (corrected on the next tick). This is the accepted cost of the slow-poll
+  (Approach B) design; AFK timing is unaffected.
+- **No change to the AFK model.** `UpdateAfk` still flags `BecameAfk` when
+  `Clock - stillSince >= Threshold`, evaluated whenever `Diff` runs. The slow poll
+  exists solely to guarantee `Diff` runs periodically, so "became AFK" cannot stall
+  during broadcast silence (a stationary player in a quiet/solo team emits no
+  `team_changed`). Worst-case detection latency for BecameAfk is
+  `AfkThreshold + TeamPollInterval`; presence/death/respawn/returned-from-AFK remain
+  instant via push.
+- **Baseline priming.** Whichever of {first broadcast, first `PollTeamAsync`
+  iteration} runs first primes the tracker silently (`Diff` primes on first non-null
+  call); the other diffs against it. No spurious connect-event flood on connect.
+
+## 5. Testing
+
+- `RustPlusSocketSource`: raising `OnTeamChanged` surfaces a correctly-mapped
+  `TeamInfoSnapshot` — member fields, leader death note, offline and dead members.
+- `ConnectionSupervisor`: a `TeamChanged` event carrying a moved / disconnected /
+  died member publishes the expected `PlayerStateChangedEvent`; and the fast marker
+  cadence no longer issues any `getTeamInfo` request.
+- AFK-during-silence: with no broadcasts delivered, a stationary online player is
+  still flagged `BecameAfk` from the slow poll within
+  `AfkThreshold + TeamPollInterval`.
+- Regression: existing marker/rig and status-loop tests remain green (the marker
+  loop is untouched apart from removing the team block).
+
+## 6. Rollout
+
+- Bump `RustPlusApi` and `RustPlusApi.Fcm` to `2.0.0-beta.6` in
+  `Directory.Packages.props`.
+- The `Unknown broadcast received` warning disappears once beta.6 dispatches
+  `team_changed`; no code change is required to silence it.
+
+## 7. Out-of-scope follow-up (noted, not included here)
+
+The related investigation found that `RustPlusSocketSource.GetMapMarkersAsync`
+throws a bare `"GetMapMarkers returned no data."` that discards the server's
+`response.Error` code/message, making the `Marker poll ... failed` warning
+non-diagnostic. That is a separate, independent fix and is **not** part of this
+design.

docs/superpowers/specs/2026-07-24-team-changed-push-design.md

@@ -42,6 +42,14 @@ public sealed class ConnectionOptions
     /// <summary>Movement tolerance (world units) below which a member is considered still. Default 1.</summary>
     public float AfkEpsilon { get; set; } = 1f;
 
+    /// <summary>
+    /// How often to poll team info as an AFK safety tick and self-heal, now that live team changes arrive
+    /// via the pushed <c>team_changed</c> event. Presence/death events are instant; this only bounds how long
+    /// a still player in a broadcast-silent team can take to be flagged AFK (worst case AfkThreshold + this).
+    /// Default 30s.
+    /// </summary>
+    public TimeSpan TeamPollInterval { get; set; } = TimeSpan.FromSeconds(30);
+
     /// <summary>How often to poll managed devices for reachability changes while connected. Default 5m.</summary>
     public TimeSpan ReachabilityPollInterval { get; set; } = TimeSpan.FromMinutes(5);
 

src/RustPlusBot.Features.Connections/ConnectionOptions.cs

@@ -169,4 +169,11 @@ Task<IReadOnlyList<MonumentSnapshot>> GetMonumentsAsync(TimeSpan timeout,
 
     /// <summary>Raised when a managed storage monitor's contents change in-game; carries the entity id and the new contents.</summary>
     event EventHandler<StorageMonitorTrigger>? StorageMonitorTriggered;
+
+    /// <summary>
+    /// Raised when the server pushes a <c>team_changed</c> broadcast (member join/leave, online/offline,
+    /// death/respawn, movement, leader change). Carries the full team snapshot — the same shape a
+    /// <see cref="GetTeamInfoAsync"/> poll returns.
+    /// </summary>
+    event EventHandler<TeamInfoSnapshot>? TeamChanged;
 }

src/RustPlusBot.Features.Connections/Listening/IRustServerConnection.cs

@@ -138,6 +138,12 @@ public event EventHandler<StorageMonitorTrigger>? StorageMonitorTriggered
             remove { _ = value; }
         }
 
+        public event EventHandler<TeamInfoSnapshot>? TeamChanged
+        {
+            add { _ = value; }
+            remove { _ = value; }
+        }
+
         public ValueTask DisposeAsync() => ValueTask.CompletedTask;
     }
 
@@ -170,6 +176,7 @@ public RustPlusServerConnection(string ip,
             _rustPlus.OnStorageMonitorTriggered += OnStorageMonitorTriggered;
             _rustPlus.OnClanChatReceived += OnClanChatReceived;
             _rustPlus.OnClanChanged += OnClanChanged;
+            _rustPlus.OnTeamChanged += OnTeamChanged;
         }
 
         /// <inheritdoc />
@@ -336,21 +343,7 @@ public async Task<HeartbeatResult> GetInfoAsync(TimeSpan timeout, CancellationTo
                     return null;
                 }
 
-                var members = (response.Data.Members ?? [])
-                    .Select(m => new TeamMemberSnapshot(
-                        m.SteamId,
-                        m.Name ?? string.Empty,
-                        m.X,
-                        m.Y,
-                        m.IsOnline,
-                        m.IsAlive,
-                        new DateTimeOffset(DateTime.SpecifyKind(m.LastSpawnTime, DateTimeKind.Utc)),
-                        new DateTimeOffset(DateTime.SpecifyKind(m.LastDeathTime, DateTimeKind.Utc))))
-                    .ToList();
-                var deathNote = response.Data.DeathNote is { } dn
-                    ? ((float X, float Y)?)(dn.X, dn.Y)
-                    : null;
-                return new TeamInfoSnapshot(response.Data.LeaderSteamId, members, deathNote);
+                return TeamInfoMapping.ToSnapshot(response.Data);
             }
             catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
             {
@@ -466,6 +459,8 @@ public async Task<bool> PromoteToLeaderAsync(ulong steamId,
 
         public event EventHandler<StorageMonitorTrigger>? StorageMonitorTriggered;
 
+        public event EventHandler<TeamInfoSnapshot>? TeamChanged;
+
         public async Task<DeviceReading> GetSmartDeviceInfoAsync(ulong entityId,
             SmartDeviceKind kind,
             TimeSpan timeout,
@@ -766,6 +761,7 @@ public async ValueTask DisposeAsync()
             _rustPlus.OnStorageMonitorTriggered -= OnStorageMonitorTriggered;
             _rustPlus.OnClanChatReceived -= OnClanChatReceived;
             _rustPlus.OnClanChanged -= OnClanChanged;
+            _rustPlus.OnTeamChanged -= OnTeamChanged;
             try
             {
                 // CONFIRMED: RustPlusSocket implements IAsyncDisposable in 2.0.0-beta.1.
@@ -835,6 +831,14 @@ private void OnClanChanged(object? sender, ClanChangedEventArg e) =>
             ClanChanged?.Invoke(this,
                 e.ClanInfo is { } info ? ClanProbeResult.From(ClanMapping.ToSnapshot(info)) : ClanProbeResult.NoClan);
 
+        private void OnTeamChanged(object? sender, RustPlusApi.Data.Events.TeamChangedEventArg e)
+        {
+            if (e.TeamInfo is { } teamInfo)
+            {
+                TeamChanged?.Invoke(this, TeamInfoMapping.ToSnapshot(teamInfo));
+            }
+        }
+
         [LoggerMessage(Level = LogLevel.Warning, Message = "Rust+ socket connect failed.")]
         private static partial void LogConnectFailed(ILogger logger, Exception ex);
 

src/RustPlusBot.Features.Connections/Listening/RustPlusSocketSource.cs

@@ -0,0 +1,33 @@
+using RustPlusBot.Abstractions.Connections;
+
+namespace RustPlusBot.Features.Connections.Listening;
+
+/// <summary>Maps a RustPlusApi <see cref="RustPlusApi.Data.TeamInfo"/> to the bot's decoupled
+/// <see cref="TeamInfoSnapshot"/>. Shared by the polled read and the pushed <c>team_changed</c> event so
+/// the two paths can never drift.</summary>
+internal static class TeamInfoMapping
+{
+    /// <summary>Projects a RustPlusApi team-info model onto a <see cref="TeamInfoSnapshot"/>.</summary>
+    /// <param name="teamInfo">The RustPlusApi team info (from a poll response or a team_changed broadcast).</param>
+    /// <returns>The decoupled snapshot the tracker and events consume.</returns>
+    public static TeamInfoSnapshot ToSnapshot(RustPlusApi.Data.TeamInfo teamInfo)
+    {
+        // CONFIRMED (2.0.0-beta.6): MemberInfo exposes SteamId/Name?/X/Y/IsOnline/IsAlive plus
+        // LastSpawnTime/LastDeathTime as Unspecified-kind DateTimes that are actually UTC.
+        var members = (teamInfo.Members ?? [])
+            .Select(m => new TeamMemberSnapshot(
+                m.SteamId,
+                m.Name ?? string.Empty,
+                m.X,
+                m.Y,
+                m.IsOnline,
+                m.IsAlive,
+                new DateTimeOffset(DateTime.SpecifyKind(m.LastSpawnTime, DateTimeKind.Utc)),
+                new DateTimeOffset(DateTime.SpecifyKind(m.LastDeathTime, DateTimeKind.Utc))))
+            .ToList();
+        var deathNote = teamInfo.DeathNote is { } dn
+            ? ((float X, float Y)?)(dn.X, dn.Y)
+            : null;
+        return new TeamInfoSnapshot(teamInfo.LeaderSteamId, members, deathNote);
+    }
+}