GH-3519: restart a projection agent whose shard wedged out from under it (#3550)

Follow-up to GH-3535 (start-failure isolation) and GH-3520 (rebuild/rewind
resume). On multi-store, Wolverine-managed Marten hosts a projection shard can
start successfully and then die underneath its EventSubscriptionAgent wrapper --
a lost first-assignment start race that wedged daemon-side, a daemon-side stop,
or an execution-loop fault. Two gaps kept it dead for the process lifetime:

1. EventSubscriptionAgent.Status latched Running once started, so a shard that
   stopped underneath the wrapper kept reporting Running. NodeAgentController
   only restarts agents it can see are non-Running, so it saw nothing to fix.

2. NodeAgentController.StartAgentAsync short-circuited on a bare
   Agents.ContainsKey, treating any registered agent as healthy forever. The
   recurring Solo reevaluation therefore never resurrected a registered-but-dead
   agent -- it just sat in the observable 30s retry loop reporting a stale
   Running while its high-water climbed.

Fix both: Status now delegates to the live inner daemon agent once started (and
still reads Stopped before start / after an explicit stop), and StartAgentAsync
re-drives a registered agent whose shard has Stopped -- stopping it first to
release any lingering daemon-side shard state -- while leaving genuinely Running
and deliberately Paused (error backoff / blue-green gate) agents untouched.

Red-green regression test agent_restart_when_wedged in CoreTests. Also bumps to
6.22.0-alpha.3.

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

Author: jeremydmiller

Directory.Build.props

@@ -14,7 +14,7 @@
         <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
         <ImplicitUsings>true</ImplicitUsings>
         <Nullable>enable</Nullable>
-        <Version>6.22.0-alpha.2</Version>
+        <Version>6.22.0-alpha.3</Version>
         <RepositoryUrl>$(PackageProjectUrl)</RepositoryUrl>
         <PublishRepositoryUrl>true</PublishRepositoryUrl>
         <EmbedUntrackedSources>true</EmbedUntrackedSources>

src/Testing/CoreTests/Runtime/Agents/agent_restart_when_wedged.cs

@@ -0,0 +1,170 @@
+using JasperFx;
+using JasperFx.Core;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using Shouldly;
+using Wolverine.Runtime;
+using Wolverine.Runtime.Agents;
+using Xunit;
+
+namespace CoreTests.Runtime.Agents;
+
+/// <summary>
+/// Regression coverage for GH-3519. A projection / event-subscription shard that starts and then dies
+/// underneath its wrapper (a lost first-assignment start race that wedged at the daemon, a daemon-side
+/// stop, or an execution-loop fault) used to stay registered on the node reporting a stale Running.
+/// NodeAgentController.StartAgentAsync short-circuited on the bare ContainsKey, so the recurring Solo
+/// reevaluation never resurrected it -- it just sat in the observable 30s retry loop forever. The
+/// controller must now notice a registered-but-Stopped agent and re-drive its start.
+/// </summary>
+public class agent_restart_when_wedged
+{
+    private readonly WolverineOptions _options;
+    private readonly IWolverineRuntime _runtime;
+    private readonly CancellationTokenSource _cancellation = new();
+
+    public agent_restart_when_wedged()
+    {
+        _options = new WolverineOptions
+        {
+            ApplicationAssembly = GetType().Assembly
+        };
+        _options.Durability.Mode = DurabilityMode.Solo;
+        _options.Durability.DurabilityAgentEnabled = false; // skip MessageStoreCollection wiring
+        _options.Durability.CheckAssignmentPeriod = 1.Hours(); // keep the recurring loop out of the way
+
+        _runtime = Substitute.For<IWolverineRuntime>();
+        _runtime.Options.Returns(_options);
+        _runtime.DurabilitySettings.Returns(_options.Durability);
+        _runtime.Observer.Returns(Substitute.For<IWolverineObserver>());
+    }
+
+    private NodeAgentController controllerFor(params FakeAgent[] agents)
+    {
+        var family = new FakeAgentFamily("test-family");
+        foreach (var agent in agents)
+        {
+            family.Add(agent);
+        }
+
+        return new NodeAgentController(
+            _runtime,
+            Substitute.For<INodeAgentPersistence>(),
+            [family],
+            NullLogger<NodeAgentController>.Instance,
+            _cancellation.Token);
+    }
+
+    [Fact]
+    public async Task restarts_a_registered_agent_whose_shard_has_stopped_underneath_it()
+    {
+        var uri = new Uri("test-family://wedged");
+        var agent = new FakeAgent(uri);
+        var controller = controllerFor(agent);
+
+        await controller.StartAgentAsync(uri);
+        agent.StartCount.ShouldBe(1);
+        controller.Agents.ContainsKey(uri).ShouldBeTrue();
+
+        // Simulate the shard dying underneath the wrapper: the agent is still registered on the node,
+        // but its underlying status has flipped to Stopped (what EventSubscriptionAgent.Status now
+        // surfaces from the live daemon shard).
+        agent.SimulateShardStopped();
+
+        await controller.StartAgentAsync(uri);
+
+        // Before the fix this second call short-circuited on ContainsKey and left the shard dead. Now
+        // the wedged agent is stopped, evicted, and re-driven.
+        agent.StartCount.ShouldBe(2);
+        agent.Status.ShouldBe(AgentStatus.Running);
+        controller.Agents.ContainsKey(uri).ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task does_not_restart_a_genuinely_running_agent()
+    {
+        var uri = new Uri("test-family://healthy");
+        var agent = new FakeAgent(uri);
+        var controller = controllerFor(agent);
+
+        await controller.StartAgentAsync(uri);
+        await controller.StartAgentAsync(uri);
+        await controller.StartAgentAsync(uri);
+
+        // A running agent stays idempotent -- no needless stop/start churn.
+        agent.StartCount.ShouldBe(1);
+        agent.StopCount.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task leaves_a_paused_agent_alone()
+    {
+        var uri = new Uri("test-family://paused");
+        var agent = new FakeAgent(uri);
+        var controller = controllerFor(agent);
+
+        await controller.StartAgentAsync(uri);
+
+        // Paused is a deliberate error-backoff / blue-green state that owns its own resume schedule.
+        agent.SimulatePaused();
+        await controller.StartAgentAsync(uri);
+
+        agent.StartCount.ShouldBe(1);
+        agent.StopCount.ShouldBe(0);
+    }
+
+    private class FakeAgentFamily : IAgentFamily
+    {
+        private readonly Dictionary<Uri, FakeAgent> _agents = new();
+
+        public FakeAgentFamily(string scheme)
+        {
+            Scheme = scheme;
+        }
+
+        public void Add(FakeAgent agent) => _agents[agent.Uri] = agent;
+
+        public string Scheme { get; }
+
+        public ValueTask<IReadOnlyList<Uri>> AllKnownAgentsAsync()
+            => ValueTask.FromResult<IReadOnlyList<Uri>>(_agents.Keys.ToList());
+
+        public ValueTask<IAgent> BuildAgentAsync(Uri uri, IWolverineRuntime wolverineRuntime)
+            => ValueTask.FromResult<IAgent>(_agents[uri]);
+
+        public ValueTask<IReadOnlyList<Uri>> SupportedAgentsAsync()
+            => ValueTask.FromResult<IReadOnlyList<Uri>>(_agents.Keys.ToList());
+
+        public ValueTask EvaluateAssignmentsAsync(AssignmentGrid assignments) => ValueTask.CompletedTask;
+    }
+
+    private class FakeAgent : IAgent
+    {
+        public FakeAgent(Uri uri) => Uri = uri;
+
+        public int StartCount { get; private set; }
+        public int StopCount { get; private set; }
+
+        public Uri Uri { get; }
+        public AgentStatus Status { get; private set; } = AgentStatus.Stopped;
+
+        // Flip the underlying status without touching the controller registration, mimicking a daemon
+        // shard that stops out from under the wrapper.
+        public void SimulateShardStopped() => Status = AgentStatus.Stopped;
+        public void SimulatePaused() => Status = AgentStatus.Paused;
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            StartCount++;
+            Status = AgentStatus.Running;
+            return Task.CompletedTask;
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken)
+        {
+            StopCount++;
+            Status = AgentStatus.Stopped;
+            return Task.CompletedTask;
+        }
+    }
+}

src/Wolverine/Runtime/Agents/EventSubscriptionAgent.cs

@@ -124,8 +124,23 @@ await _daemon.RewindSubscriptionAsync(_shardName.Name, _shardName.TenantId, canc
 
     public Uri Uri { get; }
 
-    // Be nice for this to get the Paused too
-    public AgentStatus Status { get; private set; } = AgentStatus.Stopped;
+    private AgentStatus _status = AgentStatus.Stopped;
+
+    // GH-3519: reflect the LIVE daemon shard status once we have started rather than latching a value.
+    // The wedge in the field: a shard that stops or idles underneath this wrapper -- a lost
+    // first-assignment start race, a daemon-side stop, or an execution-loop fault -- used to keep
+    // reading Running because nothing flipped the latched field back. NodeAgentController only restarts
+    // agents it can see are non-Running, so a wrapper that lies "Running" over a dead shard froze at
+    // RegisteredIdle forever while its high-water climbed. Delegating to the inner agent (whose Status
+    // the daemon keeps current) lets the controller's reevaluation notice the dead shard and restart it.
+    // Before an inner agent exists, or after an explicit StopAsync (which sets _status = Stopped), fall
+    // back to the wrapper's own tracked value so a not-yet-started / deliberately-stopped agent still
+    // reads Stopped.
+    public AgentStatus Status
+    {
+        get => _status == AgentStatus.Stopped ? AgentStatus.Stopped : _innerAgent?.Status ?? _status;
+        private set => _status = value;
+    }
 
     public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context,
         CancellationToken cancellationToken = default)

src/Wolverine/Runtime/Agents/NodeAgentController.cs

@@ -171,9 +171,36 @@ private ValueTask<IAgent> findAgentAsync(Uri uri)
 
     public async Task StartAgentAsync(Uri agentUri)
     {
-        if (Agents.ContainsKey(agentUri))
+        if (Agents.TryGetValue(agentUri, out var existing))
         {
-            return;
+            // Idempotent for an agent that is genuinely still running -- or one deliberately Paused by an
+            // error backoff / blue-green side-effect gate, which owns its own resume schedule and must not
+            // be fought here.
+            if (existing.Status != AgentStatus.Stopped)
+            {
+                return;
+            }
+
+            // GH-3519: the agent is still registered on this node but its underlying shard has stopped
+            // (e.g. an event-subscription shard that lost a first-assignment startup race and wedged, or
+            // whose daemon execution loop faulted). The old blanket ContainsKey short-circuit treated any
+            // registered agent as healthy forever, so the recurring reevaluation never resurrected a
+            // wedged one -- it just sat in the 30s retry loop reporting a stale Running. Evict the dead
+            // registration -- stopping it first to release any lingering daemon-side shard state -- so the
+            // start below actually re-drives it.
+            _logger.LogInformation(
+                "Agent {AgentUri} is still registered on node {NodeNumber} but its shard is stopped; restarting it",
+                agentUri, _runtime.Options.Durability.AssignedNodeNumber);
+
+            Agents.TryRemove(agentUri, out _);
+            try
+            {
+                await existing.StopAsync(_cancellation.Token);
+            }
+            catch (Exception e)
+            {
+                _logger.LogDebug(e, "Error stopping wedged agent {AgentUri} before restarting it", agentUri);
+            }
         }
 
         var agent = await findAgentAsync(agentUri);