🐛 fix: refuse automated claims on needs-human issues (takeover)

changelog.d/fixed-needs-human-issue-claims.md

@@ -0,0 +1 @@
+- Refuse automated claims on `needs-human` or held issues and release existing automated claims on the next cleanup tick, without disturbing human claims ([#10526](https://github.com/hivecommons/hive/issues/10526)).

src/cmd/hive/claims_admission_test.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+
+	"github.com/hivecommons/hive/pkg/claims"
+	"github.com/hivecommons/hive/pkg/config"
+	"github.com/hivecommons/hive/pkg/dashboard"
+	"github.com/hivecommons/hive/pkg/github"
+)
+
+func TestKickClaimsRespectNeedsHumanAndReleaseOnTick(t *testing.T) {
+	var label atomic.Value
+	label.Store("needs-human")
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, `{"number":1,"labels":[{"name":%q}]}`, label.Load())
+	}))
+	defer api.Close()
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	client := github.NewClientForTest(api.URL, "o", []string{"r"}, logger)
+	ledger := testClaimsLedger(t)
+	installClaimAdmissionCheck(context.Background(), ledger, func() *github.Client { return client })
+	srv := dashboard.NewServer(0, logger)
+	srv.RegisterAPI(&dashboard.Dependencies{Config: &config.Config{}, IssueClaims: ledger})
+	t.Cleanup(srv.CloseContributeHub)
+
+	recordAgentKickListings(srv, "o", "scanner", []string{"r#1"}, logger)
+	if _, listed := ledger.Listed("o/r", 1); !listed {
+		t.Fatal("kick did not record listing")
+	}
+	recordAgentStart(ledger, "o", "scanner", "r", 1, "comment", logger)
+	if _, ok := ledger.Lookup("o/r", 1); ok {
+		t.Fatal("needs-human issue claimed on start")
+	}
+	label.Store("bug")
+	recordAgentStart(ledger, "o", "scanner", "r", 1, "comment", logger)
+	if c, ok := ledger.Lookup("o/r", 1); !ok || c.Kind != claims.KindAgent {
+		t.Fatal("ordinary issue not claimed")
+	}
+	label.Store("needs-human")
+	if n := ledger.ReleaseBlocked(); n != 1 {
+		t.Fatalf("cleanup tick released %d claims, want 1", n)
+	}
+	if _, held := claimsInflightLookup(ledger, "o")(github.Issue{Repo: "r", Number: 1}); held {
+		t.Fatal("released claim still reported In Flight")
+	}
+}

src/cmd/hive/claimswire.go

@@ -56,6 +56,24 @@ func buildClaimsLedger(cfg *config.Config, logger *slog.Logger) *claims.Ledger {
 	return ledger
 }
 
+// installClaimAdmissionCheck gates every automated ledger producer on current
+// labels, including renewals and forced takeovers. Resolve the client per call
+// because App setup can finish after boot.
+func installClaimAdmissionCheck(ctx context.Context, ledger *claims.Ledger, client func() *github.Client) {
+	if ledger == nil || client == nil {
+		return
+	}
+	ledger.SetAdmissionCheck(func(req claims.Request) (string, error) {
+		ctx, cancel := context.WithTimeout(ctx, claimsGitHubTimeout)
+		defer cancel()
+		gh := client()
+		if gh == nil {
+			return "", github.ErrNoGitHubClient
+		}
+		return gh.IssueClaimBlockReason(ctx, req.Repo, req.Issue)
+	})
+}
+
 // githubClaimHooks returns the hooks that mirror ledger transitions onto the
 // GitHub issue. Every call is best-effort: a failed comment or label never
 // affects the ledger, which is authoritative for the hub's own decisions.

src/cmd/hive/main.go

@@ -3391,6 +3391,7 @@ func (b *boot) bootDashboardAPIWith(deps bootDashboardAPIDeps) {
 	wireSpektacularRunner(b.cfg, b.dashSrv, b.logger)
 	// #8380: chain GitHub comments/labels behind the relay yank on takeover.
 	b.dashSrv.InstallClaimHooks(githubClaimHooks(b.ctx, b.cfg, func() *github.Client { return b.ghClient }, b.logger))
+	installClaimAdmissionCheck(b.ctx, b.issueClaims, func() *github.Client { return b.ghClient })
 	// Forge App tab inventory: the resolved active key path and the per-app-id
 	// PVC keys live here in cmd/hive, so they are injected as a provider (the
 	// SetGitHubAppRecheckFn pattern). Fingerprints and paths only — the

src/pkg/claims/admission_test.go

@@ -0,0 +1,93 @@
+package claims
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestAdmissionRefusesAutomatedClaims(t *testing.T) {
+	for _, reason := range []string{"hold", "needs-human"} {
+		for _, kind := range []Kind{KindAgent, KindContributor, KindExternal} {
+			t.Run(reason+"/"+string(kind), func(t *testing.T) {
+				writes := 0
+				l, _ := newTestLedger(t, Hooks{OnClaimed: func(Claim, Outcome) { writes++ }})
+				l.SetAdmissionCheck(func(Request) (string, error) { return reason, nil })
+				res, err := l.Claim(Request{Repo: "o/r", Issue: 1, Holder: "worker", Kind: kind, Force: true})
+				if err != nil || res.Outcome != OutcomeRefused || res.Reason != reason {
+					t.Fatalf("claim = %+v, %v", res, err)
+				}
+				if len(l.List()) != 0 || writes != 0 {
+					t.Fatal("refused claim mutated ledger or fired write hook")
+				}
+			})
+		}
+	}
+}
+
+func TestAdmissionRefusesRenewalAndTakeover(t *testing.T) {
+	l, _ := newTestLedger(t, Hooks{})
+	req := Request{Repo: "o/r", Issue: 1, Holder: "worker", Kind: KindContributor}
+	first, _ := l.Claim(req)
+	l.SetAdmissionCheck(func(Request) (string, error) { return "needs-human", nil })
+	for _, next := range []Request{req, {Repo: "o/r", Issue: 1, Holder: "scanner", Kind: KindAgent, Force: true}} {
+		res, err := l.Claim(next)
+		if err != nil || res.Outcome != OutcomeRefused {
+			t.Fatalf("claim = %+v, %v", res, err)
+		}
+		if got, _ := l.Lookup("o/r", 1); got != first.Claim {
+			t.Fatal("refusal changed existing claim")
+		}
+	}
+}
+
+func TestReleaseBlockedClaims(t *testing.T) {
+	released := 0
+	l, _ := newTestLedger(t, Hooks{OnReleased: func(c Claim, reason string) {
+		if c.Kind == KindHuman || reason != "needs-human" {
+			t.Errorf("unexpected release: %+v, %q", c, reason)
+		}
+		released++
+	}})
+	for i, kind := range []Kind{KindAgent, KindContributor, KindExternal, KindHuman} {
+		_, _ = l.Claim(Request{Repo: "o/r", Issue: i + 1, Holder: "worker", Kind: kind})
+	}
+	l.SetAdmissionCheck(func(req Request) (string, error) {
+		if req.Issue == 3 {
+			return "", errors.New("forge unavailable")
+		}
+		return "needs-human", nil
+	})
+	if n := l.ReleaseBlocked(); n != 2 || released != 2 {
+		t.Fatalf("released %d, hooks %d; want 2", n, released)
+	}
+	if _, ok := l.Lookup("o/r", 3); !ok {
+		t.Fatal("lookup error released a claim")
+	}
+	if _, ok := l.Lookup("o/r", 4); !ok {
+		t.Fatal("needs-human released a human claim")
+	}
+	res, err := l.Claim(Request{Repo: "o/r", Issue: 5, Holder: "person", Kind: KindHuman})
+	if err != nil || res.Outcome != OutcomeClaimed {
+		t.Fatalf("human claim refused: %+v, %v", res, err)
+	}
+	res, err = l.Claim(Request{Repo: "o/r", Issue: 3, Holder: "other", Kind: KindAgent})
+	if err == nil || res.Outcome.Changed() {
+		t.Fatalf("lookup failure admitted claim: %+v, %v", res, err)
+	}
+}
+
+func TestReleaseBlockedPreservesConcurrentTakeover(t *testing.T) {
+	l, _ := newTestLedger(t, Hooks{})
+	_, _ = l.Claim(Request{Repo: "o/r", Issue: 1, Holder: "scanner", Kind: KindAgent})
+	l.SetAdmissionCheck(func(req Request) (string, error) {
+		// Human takeover while the forge check is in progress.
+		_, _ = l.Claim(Request{Repo: req.Repo, Issue: req.Issue, Holder: "person", Kind: KindHuman})
+		return "needs-human", nil
+	})
+	if n := l.ReleaseBlocked(); n != 0 {
+		t.Fatalf("released %d concurrent claims", n)
+	}
+	if c, ok := l.Lookup("o/r", 1); !ok || c.Kind != KindHuman {
+		t.Fatalf("human takeover lost: %+v, %v", c, ok)
+	}
+}

src/pkg/claims/claims.go

@@ -237,6 +237,8 @@ type Result struct {
 	// Previous is the displaced (taken_over) or blocking (held/refused)
 	// holder's claim, when there was one.
 	Previous *Claim
+	// Reason explains an admission refusal, which has no blocking holder.
+	Reason string
 }
 
 // Hooks are the hub-side side effects, all optional. They run after the
@@ -346,6 +348,7 @@ type Ledger struct {
 	hooks  Hooks
 	now    func() time.Time
 	hive   string
+	admit  func(Request) (string, error)
 }
 
 // ErrInvalid is returned for requests missing a repo, issue, holder or kind.
@@ -388,6 +391,15 @@ func (l *Ledger) SetHooks(h Hooks) {
 	l.mu.Unlock()
 }
 
+// SetAdmissionCheck installs a non-human claim gate. An empty reason admits
+// the request; an error leaves the ledger unchanged. The check runs outside
+// the ledger lock so a forge lookup never blocks other ledger operations.
+func (l *Ledger) SetAdmissionCheck(check func(Request) (string, error)) {
+	l.mu.Lock()
+	l.admit = check
+	l.mu.Unlock()
+}
+
 // SetNow overrides the clock. Intended for tests.
 func (l *Ledger) SetNow(fn func() time.Time) {
 	l.mu.Lock()
@@ -404,6 +416,18 @@ func (l *Ledger) Claim(req Request) (Result, error) {
 		return Result{}, ErrInvalid
 	}
 	l.mu.Lock()
+	check := l.admit
+	l.mu.Unlock()
+	if check != nil && req.Kind != KindHuman {
+		reason, err := check(req)
+		if err != nil {
+			return Result{}, err
+		}
+		if reason != "" {
+			return Result{Outcome: OutcomeRefused, Reason: reason}, nil
+		}
+	}
+	l.mu.Lock()
 	now := l.now()
 	expired := l.expireLocked(now)
 	key := Key(req.Repo, req.Issue)
@@ -551,6 +575,46 @@ func (l *Ledger) ForceRelease(repo string, issue int, reason string) (Claim, boo
 	return c, ok
 }
 
+// ReleaseBlocked rechecks non-human claims against the admission gate. Failed
+// lookups preserve claims. A concurrent renewal or takeover also preserves the
+// new claim: only the exact snapshot checked outside the lock may be removed.
+func (l *Ledger) ReleaseBlocked() int {
+	if l == nil {
+		return 0
+	}
+	l.mu.Lock()
+	check := l.admit
+	l.mu.Unlock()
+	if check == nil {
+		return 0
+	}
+	n := 0
+	for _, c := range l.List() {
+		if c.Kind == KindHuman {
+			continue
+		}
+		reason, err := check(Request{Repo: c.Repo, Issue: c.Issue, Holder: c.Holder, HolderID: c.HolderID, Kind: c.Kind})
+		if err != nil || reason == "" {
+			continue
+		}
+		l.mu.Lock()
+		current, ok := l.claims[c.Key()]
+		if !ok || current != c {
+			l.mu.Unlock()
+			continue
+		}
+		delete(l.claims, c.Key())
+		_ = l.saveLocked()
+		hooks := l.hooks
+		l.mu.Unlock()
+		n++
+		if hooks.OnReleased != nil {
+			hooks.OnReleased(c, reason)
+		}
+	}
+	return n
+}
+
 // ReleaseByHolderID drops every claim whose HolderID matches — used when a
 // relay lease is revoked or expires so the claim goes with it. Returns the
 // released claims. A non-empty onlyKey restricts the release to that one

src/pkg/dashboard/blocked_claims_loop_test.go

@@ -0,0 +1,32 @@
+package dashboard
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hivecommons/hive/internal/testutil"
+	"github.com/hivecommons/hive/pkg/claims"
+)
+
+func TestBlockedClaimsLoopReleasesAndStops(t *testing.T) {
+	hub, s, ledger := claimsHub(t)
+	if _, err := ledger.Claim(claims.Request{Repo: "o/r", Issue: 1, Holder: "worker", Kind: claims.KindAgent}); err != nil {
+		t.Fatal(err)
+	}
+	ledger.SetAdmissionCheck(func(claims.Request) (string, error) { return "needs-human", nil })
+	worker := &ContributeWSHub{server: s, logger: hub.logger, stopCh: make(chan struct{})}
+	done := make(chan struct{})
+	go func() { defer close(done); worker.blockedClaimsLoop(time.Millisecond) }()
+	defer func() {
+		close(worker.stopCh)
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Error("poller did not stop")
+		}
+	}()
+	testutil.Eventually(t, time.Second, func() bool {
+		_, ok := ledger.Lookup("o/r", 1)
+		return !ok
+	}, "blocked claim was not released")
+}

src/pkg/dashboard/contribute_claim_gate_test.go

@@ -0,0 +1,28 @@
+package dashboard
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/hivecommons/hive/pkg/github"
+)
+
+func TestRelayClaimMarkerRefusesNeedsHuman(t *testing.T) {
+	hub, s := covK2Hub(t)
+	enableIssueClaims(s)
+	recorder := &claimRecorder{}
+	hub.claimCommenter = recorder
+	api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, `{"number":42,"labels":[{"name":"needs-human"}]}`)
+	}))
+	defer api.Close()
+	s.deps.GHClient = github.NewClientForTest(api.URL, "o", []string{"r"}, hub.logger)
+	claim, posted := hub.recordAgentClaim(context.Background(), issueClaimConn("contributor"), "task-1", "o/r", 42, time.Now())
+	if posted || claim.Identity != "" || len(recorder.all()) != 0 {
+		t.Fatalf("needs-human received legacy claim marker: %+v, posted=%v", claim, posted)
+	}
+}

src/pkg/dashboard/contribute_claims.go

@@ -125,6 +125,17 @@ func (h *ContributeWSHub) recordAgentClaim(ctx context.Context, c *ContributorCo
 	if !h.claimsEnabled() || c == nil || repoFull == "" || number <= 0 {
 		return ghpkg.IssueClaimMark{}, false
 	}
+	// The queue may predate a label change. Recheck before posting the
+	// legacy claim marker too, not only when writing the worker ledger.
+	if h.server.deps.GHClient != nil {
+		checkCtx, cancel := context.WithTimeout(ctx, claimCommentTimeout)
+		reason, err := h.server.deps.GHClient.IssueClaimBlockReason(checkCtx, repoFull, number)
+		cancel()
+		if err != nil || reason != "" {
+			h.logger.Warn("[contribute-ws] issue claim refused", "repo", repoFull, "number", number, "reason", reason, "error", err)
+			return ghpkg.IssueClaimMark{}, false
+		}
+	}
 	claim := ghpkg.IssueClaimMark{
 		Identity:  claimIdentityFor(c),
 		StartedAt: now,

src/pkg/dashboard/contribute_ws.go

@@ -4181,7 +4181,26 @@ func (h *ContributeWSHub) checkEffortAllowed(backend, effort string) (bool, stri
 	return true, ""
 }
 
+// blockedClaimsLoop polls labels separately from connection cleanup. Five minutes
+// bounds steady-state traffic to 12 reads per claim/hour; slow GitHub calls
+// cannot delay heartbeat reaping, and passes never overlap.
+func (h *ContributeWSHub) blockedClaimsLoop(interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-h.stopCh:
+			return
+		case <-ticker.C:
+			if n := h.claimsLedger().ReleaseBlocked(); n > 0 {
+				h.logger.Info("[claims] blocked claims released", "count", n)
+			}
+		}
+	}
+}
+
 func (h *ContributeWSHub) cleanupLoop() {
+	go h.blockedClaimsLoop(5 * time.Minute)
 	if h.doneCh != nil {
 		defer close(h.doneCh)
 	}