An adopted engine that turns ready after registration is reported

The registration carries each shard's state as it is then. An adopted
engine is health-checked after that, and nothing reported the ready that
followed, so when the agent registered first the control plane kept the
replica starting while it served. After llmgw restart on the PoC two
replicas stayed so, one the only replica of qwen2.5-coder-14b. The
session's watcher now reports every shard the registration called not
ready once it is.

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

CHANGELOG.md

@@ -65,6 +65,12 @@ What was left of the field report from v0.10.6.
 
 ### Fixed
 
+- **A replica no longer stays "starting" while it serves after a restart**:
+  an agent that registered before its adopted engines had answered their
+  health check never said they had, so the control plane kept them starting
+  and routed nothing to them — on the reference cluster, after `llmgw
+  restart`, the only replica of a model. The agent now reports each shard
+  the registration called not ready once it is.
 - **A failover no longer replaces healthy replicas**: a standby that took
   over placed at once, before the agents had reconnected and said what runs,
   so it started a second copy of replicas it did not know of and then drained

internal/agent/agent.go

@@ -417,6 +417,17 @@ func (a *Agent) weights() *weights.Fetcher {
 }
 
 // runningShards describes what this agent is supervising, for registration.
+// unready are the shards a registration reported as not yet ready.
+func unready(running []*llmgwv1.RunningShard) map[string]bool {
+	out := map[string]bool{}
+	for _, r := range running {
+		if r.GetState() != agentsvc.StateToWire(supervisor.Ready) {
+			out[r.GetShardId()] = true
+		}
+	}
+	return out
+}
+
 func (a *Agent) runningShards() []*llmgwv1.RunningShard {
 	var out []*llmgwv1.RunningShard
 	for _, st := range a.sup.Statuses() {
@@ -571,7 +582,7 @@ func (a *Agent) session(ctx context.Context, server string) error {
 		a.sendLoop(ctx, stream, out)
 	}()
 	defer wg.Wait()
-	go a.watchExits(ctx, out)
+	go a.watchExits(ctx, out, unready(reg.RunningShards))
 	go a.forwardKV(ctx, out)
 	go a.reportCompute(ctx, out)
 
@@ -1191,7 +1202,14 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan<
 // reported the engine dying; and an adopted engine had no watcher at all —
 // on node06 one was killed and its replica stayed "ready" for the control
 // plane, routed to and failing, indefinitely.
-func (a *Agent) watchExits(ctx context.Context, out chan<- *llmgwv1.AgentMessage) {
+//
+// It also reports the shards the registration said were not ready yet once
+// they are. An adopted engine is health-checked after the agent registers,
+// and a shard started on the stream before is reported on that stream; so
+// when the registration came first nothing ever told the control plane, and
+// the replica stayed "starting" while it served — on node05, after a
+// control plane restart, the only replica of a model.
+func (a *Agent) watchExits(ctx context.Context, out chan<- *llmgwv1.AgentMessage, pending map[string]bool) {
 	t := time.NewTicker(time.Second)
 	defer t.Stop()
 	ready := map[string]bool{}
@@ -1200,6 +1218,24 @@ func (a *Agent) watchExits(ctx context.Context, out chan<- *llmgwv1.AgentMessage
 			switch st.State {
 			case supervisor.Ready:
 				ready[st.ID] = true
+				if !pending[st.ID] {
+					continue
+				}
+				delete(pending, st.ID)
+				sh, err := a.sup.Get(st.ID)
+				if err != nil {
+					continue
+				}
+				msg := &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Shard{Shard: &llmgwv1.ShardStatus{
+					NodeId: a.cfg.NodeID, ReplicaId: sh.Spec.ReplicaID, ShardId: st.ID,
+					State:   agentsvc.StateToWire(st.State),
+					Compute: computeToWire(a.sup.ComputeOf(st.ID)),
+				}}}
+				select {
+				case out <- msg:
+				case <-ctx.Done():
+					return
+				}
 			case supervisor.Failed:
 				if !ready[st.ID] {
 					continue // never ready here: its start reported it

internal/agent/watchexits_test.go

@@ -10,6 +10,7 @@ import (
 	llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines"
 	"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc"
 )
 
 type alwaysReady struct{}
@@ -40,7 +41,7 @@ func TestAnAdoptedEngineThatDiesIsReported(t *testing.T) {
 	out := make(chan *llmgwv1.AgentMessage, 4)
 	ctx, cancel := context.WithCancel(t.Context())
 	defer cancel()
-	go a.watchExits(ctx, out)
+	go a.watchExits(ctx, out, nil)
 	time.Sleep(1500 * time.Millisecond) // seen ready
 	_ = cmd.Process.Kill()
 
@@ -54,3 +55,50 @@ func TestAnAdoptedEngineThatDiesIsReported(t *testing.T) {
 		t.Fatal("the adopted engine died and nothing was reported")
 	}
 }
+
+// Registered while its health check still ran: the ready that follows is
+// reported, or the control plane keeps the replica starting while it serves.
+func TestAShardRegisteredStartingIsReportedReady(t *testing.T) {
+	sup, err := supervisor.New(supervisor.Config{WorkDir: t.TempDir(), RunAs: "-"}, alwaysReady{}, slog.New(slog.DiscardHandler))
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmd := exec.Command("sleep", "60")
+	if err := cmd.Start(); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }()
+	sh, err := sup.Adopt(supervisor.Orphan{ShardID: "m-r0", ReplicaID: "m", PID: cmd.Process.Pid,
+		Spec: engines.ShardSpec{ReplicaID: "m", ShardID: "m-r0", HTTPPort: 1}, Alive: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	a := &Agent{sup: sup, log: slog.New(slog.DiscardHandler)}
+	pending := unready(a.runningShards())
+	if !pending["m-r0"] {
+		t.Fatalf("registration reported it ready already: %v", pending)
+	}
+	out := make(chan *llmgwv1.AgentMessage, 4)
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	go a.watchExits(ctx, out, pending)
+	if err := sup.WaitReady(t.Context(), sh, "http://127.0.0.1:1", "m", 5*time.Second); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case m := <-out:
+		s := m.GetShard()
+		if s.GetShardId() != "m-r0" || s.GetState() != agentsvc.StateToWire(supervisor.Ready) {
+			t.Errorf("reported %+v", s)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("it became ready and nothing was reported")
+	}
+	select {
+	case m := <-out:
+		t.Errorf("reported twice: %+v", m.GetShard())
+	case <-time.After(2500 * time.Millisecond):
+	}
+}