diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index b37bfa9..3bbc732 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -112,6 +112,9 @@ type lease struct { Version int `json:"version"` Until time.Time `json:"until"` UsageBaseline float64 `json:"usage_baseline,omitempty"` + // ProgressSHA hashes the pane capture taken at the last renewal. Renewal + // requires the pane to have changed since then, or the agent to be busy. + ProgressSHA string `json:"progress_sha,omitempty"` } type releaseTransaction struct { ID string `json:"id"` @@ -958,10 +961,31 @@ func (w *worker) renewLeases(ctx context.Context) { if !ok || l.Version == 0 || l.Until.After(now.Add(10*time.Minute)) { continue } - if _, err := (herdr.CLIAdapter{Backend: w.executionBackend(), Harness: w.harness}).PaneCapture(ctx, s, "recent"); err != nil { + adapter := herdr.CLIAdapter{Backend: w.executionBackend(), Harness: w.harness} + text, err := adapter.PaneCapture(ctx, s, "recent") + if err != nil { w.recordError(fmt.Errorf("validate lease %s: %w", taskID, err)) continue } + // A live pane is not progress. Renewing on pane existence alone let a + // pane that opened and never started hold its lease forever, which is + // what orphaned the July task once the launch itself had failed. + status, err := adapter.AgentStatus(ctx, s) + if err != nil { + w.recordError(fmt.Errorf("validate lease %s: %w", taskID, err)) + continue + } + progress := domain.Hash([]byte(text)) + switch { + case herdr.IsBusy(status): + case progress != l.ProgressSHA && l.ProgressSHA != "": + case l.ProgressSHA == "": + // First renewal has no baseline to compare against. Record one and + // allow this renewal; the next one must show real movement. + default: + w.recordError(fmt.Errorf("lease %s not renewed: agent status %s and pane unchanged since the last renewal", taskID, status)) + continue + } if err := w.api.Renew(ctx, taskID, l.Epoch, l.Version, int((30 * time.Minute).Seconds())); err != nil { w.recordError(fmt.Errorf("renew lease %s: %w", taskID, err)) log.Printf("renew lease %s: %v", taskID, err) @@ -970,6 +994,7 @@ func (w *worker) renewLeases(ctx context.Context) { // replay arrives so a release transaction uses the same version. l.Version++ l.Until = now.Add(30 * time.Minute) + l.ProgressSHA = progress w.leases[taskID] = l _ = w.save() } diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go index ff124aa..f52824d 100644 --- a/cmd/orchestra-worker/main_test.go +++ b/cmd/orchestra-worker/main_test.go @@ -371,6 +371,7 @@ func mustJSON(v any) []byte { b, _ := json.Marshal(v); return b } type recordingBackend struct { status string + capture string calls []string prompts []string } @@ -395,7 +396,43 @@ func (b *recordingBackend) AgentStatus(context.Context, herdr.Session) (string, return b.status, nil } func (b *recordingBackend) PaneCapture(context.Context, herdr.Session, string) (string, error) { - return "", nil + return b.capture, nil +} + +func TestRenewLeasesRequiresProgress(t *testing.T) { + renewals := 0 + api := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { + renewals++ + rw.Write([]byte(`{}`)) + })) + defer api.Close() + backend := &recordingBackend{status: "idle", capture: "same screen"} + w := &worker{ + api: federation.Client{BaseURL: api.URL, WorkerID: "h", Token: "t"}, + backend: backend, + harness: "claude", + sessions: map[string]herdr.Session{"task": {PaneID: "pane"}}, + leases: map[string]lease{"task": {Epoch: "e", Version: 1, Until: time.Now(), ProgressSHA: domain.Hash([]byte("same screen"))}}, + quarantined: map[string]bool{}, + statePath: filepath.Join(t.TempDir(), "state.json"), + } + w.renewLeases(context.Background()) + if renewals != 0 { + t.Fatalf("idle pane with an unchanged capture renewed its lease %d times", renewals) + } + backend.status = "busy" + w.renewLeases(context.Background()) + if renewals != 1 { + t.Fatalf("busy agent renewals=%d, want 1", renewals) + } + backend.status, backend.capture = "idle", "new output" + l := w.leases["task"] + l.Until = time.Now() + w.leases["task"] = l + w.renewLeases(context.Background()) + if renewals != 2 { + t.Fatalf("changed capture renewals=%d, want 2", renewals) + } } func (b *recordingBackend) SendText(_ context.Context, _ herdr.Session, text string) error { b.calls = append(b.calls, "text:"+text)