diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index 3bbc732..b7aef84 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -1280,6 +1280,16 @@ func (w *worker) reconcileLeases(ctx context.Context) error { } } for taskID, l := range active { + // The coordinator is authoritative for the lease, not for what this + // worker has observed under it. Rebuilding the struct wholesale wiped + // every worker-local field each tick, which silently disabled the + // renewal progress check and reset the usage baseline. Carry them + // across, but only while the epoch is the same lease. + if prev, ok := w.leases[taskID]; ok && prev.Epoch == l.Epoch { + l.ProgressSHA = prev.ProgressSHA + l.UsageBaseline = prev.UsageBaseline + l.PickupAcknowledged = prev.PickupAcknowledged + } w.leases[taskID] = l } return nil diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index f52824d..d91955f 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -871,3 +871,47 @@ func TestInlineTransportSubmitsTheWholeInstruction(t *testing.T) { t.Fatalf("submitted %q, want the whole instruction", backend.prompts) } } + +// Found live during burn-in run 3: reconcileLeases rebuilt each lease from the +// coordinator's view every tick, which wiped ProgressSHA and made the renewal +// progress check renew unconditionally on its no-baseline branch. The fix in +// isolation is worthless if the live call path resets its input. +func TestReconcileLeasesKeepsWorkerLocalObservations(t *testing.T) { + until := time.Now().Add(30 * time.Minute) + epoch := "epoch-1" + api := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { + rw.Write(mustJSON([]domain.Task{{ + ID: "task", State: domain.StateLeased, Version: 12, + Lease: &domain.Lease{HarnessID: "h", Epoch: epoch, Until: until}, + }})) + })) + defer api.Close() + w := &worker{ + api: federation.Client{BaseURL: api.URL, WorkerID: "h", Token: "t"}, + harnessID: "h", + tasks: map[string]domain.Task{}, + leases: map[string]lease{"task": {Epoch: "epoch-1", Version: 11, ProgressSHA: "sha-1", UsageBaseline: 0.25, PickupAcknowledged: true}}, + statePath: filepath.Join(t.TempDir(), "state.json"), + } + if err := w.reconcileLeases(context.Background()); err != nil { + t.Fatal(err) + } + got := w.leases["task"] + if got.ProgressSHA != "sha-1" || got.UsageBaseline != 0.25 || !got.PickupAcknowledged { + t.Fatalf("reconcile dropped worker-local observations: %+v", got) + } + if got.Version != 12 { + t.Fatalf("version=%d, want the coordinator's 12", got.Version) + } + + // A different epoch is a different lease. Nothing observed under the old + // one may carry into it. + epoch = "epoch-2" + if err := w.reconcileLeases(context.Background()); err != nil { + t.Fatal(err) + } + got = w.leases["task"] + if got.ProgressSHA != "" || got.UsageBaseline != 0 || got.PickupAcknowledged { + t.Fatalf("observations survived a new epoch: %+v", got) + } +}