cmd/orchestra-worker/main.go

@@ -54,6 +54,7 @@ type worker struct {
 	registration                           federation.Worker
 	lastError                              string
 	lastErrorAt                            time.Time
+	observations                           []federation.Observation
 	soft                                   float64
 	window                                 int64
 }
@@ -68,12 +69,38 @@ func (w *worker) executionBackend() herdr.Backend {
 	return nil
 }
 
+// observationRing bounds how many distinct failures a worker reports at once.
+const observationRing = 16
+
 func (w *worker) recordError(err error) {
 	if err == nil {
 		return
 	}
-	w.lastError = err.Error()
-	w.lastErrorAt = time.Now().UTC()
+	msg, now := err.Error(), time.Now().UTC()
+	w.lastError, w.lastErrorAt = msg, now
+	// Collapse by message rather than by position. A retry loop interleaved
+	// with other failures would otherwise still flush the ring, which is the
+	// whole reason one slot was not enough.
+	for i, o := range w.observations {
+		if o.Message == msg {
+			w.observations[i].Count++
+			w.observations[i].Last = now
+			return
+		}
+	}
+	if len(w.observations) >= observationRing {
+		// Evict the least recently seen. A loop keeps its slot, but it carries
+		// a count that says so, and a one-off from an hour ago is the entry
+		// worth losing first.
+		oldest := 0
+		for i, o := range w.observations {
+			if o.Last.Before(w.observations[oldest].Last) {
+				oldest = i
+			}
+		}
+		w.observations = append(w.observations[:oldest], w.observations[oldest+1:]...)
+	}
+	w.observations = append(w.observations, federation.Observation{Message: msg, Count: 1, First: now, Last: now})
 }
 
 func (w *worker) health(ctx context.Context) federation.WorkerHealth {
@@ -101,6 +128,9 @@ func (w *worker) health(ctx context.Context) federation.WorkerHealth {
 		}
 	}
 	h.LastError, h.ErrorAt = w.lastError, w.lastErrorAt
+	if len(w.observations) > 0 {
+		h.Observations = append([]federation.Observation(nil), w.observations...)
+	}
 	return h
 }
 

cmd/orchestra-worker/main_test.go

@@ -1268,3 +1268,62 @@ func TestFailedTaskDropsItsReleaseTransaction(t *testing.T) {
 		t.Fatalf("terminal task kept its release: releases=%v sessions=%v", w.releases, w.sessions)
 	}
 }
+
+// F18. One last_error slot destroyed causal evidence twice: run 7 kept only
+// the last of four failures, and in run 11 a five-second retry loop on a dead
+// task pinned the slot for twenty-six minutes while the live task's own expiry
+// went unrecorded. The ring has to survive exactly that interleaving.
+func TestObservationRingSurvivesARetryLoop(t *testing.T) {
+	w := &worker{}
+	w.recordError(errors.New("release dead-task commit: 409 lease not owned"))
+	w.recordError(errors.New("lease live-task not renewed: agent status idle"))
+	for i := 0; i < 300; i++ {
+		w.recordError(errors.New("release dead-task commit: 409 lease not owned"))
+	}
+	if len(w.observations) != 2 {
+		t.Fatalf("want 2 distinct observations, got %d: %v", len(w.observations), w.observations)
+	}
+	loop, other := w.observations[0], w.observations[1]
+	if loop.Count != 301 {
+		t.Fatalf("loop count %d, want 301", loop.Count)
+	}
+	if other.Message != "lease live-task not renewed: agent status idle" || other.Count != 1 {
+		t.Fatalf("the live task's failure was evicted by the loop: %v", other)
+	}
+	if !loop.First.Before(loop.Last) && loop.First != loop.Last {
+		t.Fatalf("first/last not tracked: %v", loop)
+	}
+	// last_error keeps its meaning: the newest failure, not the ring's head.
+	if got := w.health(context.Background()); got.LastError != loop.Message || len(got.Observations) != 2 {
+		t.Fatalf("health projection wrong: %+v", got)
+	}
+}
+
+// The ring is bounded, and the entry it gives up is the one nobody has seen
+// for longest.
+func TestObservationRingEvictsLeastRecentlySeen(t *testing.T) {
+	w := &worker{}
+	for i := 0; i < observationRing; i++ {
+		w.recordError(fmt.Errorf("failure %d", i))
+	}
+	w.recordError(errors.New("failure 0")) // refresh the oldest
+	w.recordError(errors.New("one more distinct failure"))
+	if len(w.observations) != observationRing {
+		t.Fatalf("ring unbounded at %d", len(w.observations))
+	}
+	var kept0, kept1 bool
+	for _, o := range w.observations {
+		switch o.Message {
+		case "failure 0":
+			kept0 = true
+		case "failure 1":
+			kept1 = true
+		}
+	}
+	if !kept0 {
+		t.Fatal("refreshed entry evicted")
+	}
+	if kept1 {
+		t.Fatal("least recently seen entry survived")
+	}
+}

internal/federation/federation.go

@@ -39,6 +39,21 @@ type WorkerHealth struct {
 	ActivePane  string    `json:"active_pane_id,omitempty"`
 	LastError   string    `json:"last_error,omitempty"`
 	ErrorAt     time.Time `json:"error_at,omitempty"`
+	// Observations is the bounded set of distinct failures behind LastError,
+	// which keeps its wire name and still reports only the newest.
+	Observations []Observation `json:"observations,omitempty"`
+}
+
+// Observation is one distinct worker failure with its repeat count. A single
+// last_error slot let one five-second retry loop erase the cause of everything
+// around it: run 7 lost three of four failures, and in run 11 the slot was
+// pinned to a different, blocked task for twenty-six minutes. Repeats collapse
+// here so a loop cannot evict the failures beside it.
+type Observation struct {
+	Message string    `json:"message"`
+	Count   int       `json:"count"`
+	First   time.Time `json:"first"`
+	Last    time.Time `json:"last"`
 }
 
 // Capture is published by a worker that owns the pane. The coordinator never