feat: per-submission accept/reject/close buttons and fix stale-fork filter

julianknutsen · claude · julianknutsen · commit cdaa5089a3fd · 2026-03-06T06:21:57.000+01:00
Add reject-upstream and close-upstream endpoints so posters can act on
each competing submission individually. Reject closes the fork's DoltHub
PR without modifying local state; close adopts the fork's completion
without creating a stamp.

Fix the stale-fork filter that was hiding submissions (e.g. nullpriest):
the claimed_by != author check now only applies to "claimed" status,
since in_review/completed items represent intentional action.

Synthesize main-branch completions into the submissions list when an
item is in_review, so all submissions appear in one place. The separate
Completion section now only renders for completed (accepted) items.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/cmd/wl/cmd_review.go b/cmd/wl/cmd_review.go
@@ -965,3 +965,30 @@ func branchURLCallback(cfg *federation.Config) func(string) string {
 		return nil
 	}
 }
+
+// closeUpstreamPRCallback returns a callback that closes an upstream PR by its web URL.
+func closeUpstreamPRCallback(cfg *federation.Config) func(string) error {
+	switch cfg.ResolveProviderType() {
+	case "dolthub":
+		token := os.Getenv("DOLTHUB_TOKEN")
+		if token == "" {
+			return nil
+		}
+		upstreamOrg, db, err := federation.ParseUpstream(cfg.Upstream)
+		if err != nil {
+			return nil
+		}
+		provider := remote.NewDoltHubProvider(token)
+		return func(prURL string) error {
+			// Extract PR ID from URL like ".../pulls/123"
+			idx := strings.LastIndex(prURL, "/pulls/")
+			if idx < 0 {
+				return fmt.Errorf("cannot extract PR ID from URL: %s", prURL)
+			}
+			prID := prURL[idx+len("/pulls/"):]
+			return provider.ClosePR(upstreamOrg, db, prID)
+		}
+	default:
+		return nil
+	}
+}
diff --git a/cmd/wl/cmd_serve.go b/cmd/wl/cmd_serve.go
@@ -206,6 +206,7 @@ func runServe(cmd *cobra.Command, stdout, stderr io.Writer) error {
 		},
 		ListPendingItems: listPendingItemsFromPRs(cfg),
 		BranchURL:        branchURLCallback(cfg),
+		CloseUpstreamPR:  closeUpstreamPRCallback(cfg),
 	})
 
 	server := api.New(client)
diff --git a/cmd/wl/cmd_tui.go b/cmd/wl/cmd_tui.go
@@ -123,6 +123,7 @@ func runTUI(cmd *cobra.Command, _, stderr io.Writer) error {
 		},
 		ListPendingItems: listPendingItemsFromPRs(cfg),
 		BranchURL:        branchURLCallback(cfg),
+		CloseUpstreamPR:  closeUpstreamPRCallback(cfg),
 	})
 
 	m := tui.New(tui.Config{
diff --git a/cmd/wl/sdk_factory.go b/cmd/wl/sdk_factory.go
@@ -33,5 +33,6 @@ var newSDKClient = func(cfg *federation.Config, noPush bool) (*sdk.Client, error
 		},
 		ListPendingItems: listPendingItemsFromPRs(cfg),
 		BranchURL:        branchURLCallback(cfg),
+		CloseUpstreamPR:  closeUpstreamPRCallback(cfg),
 	}), nil
 }
diff --git a/internal/api/handlers.go b/internal/api/handlers.go
@@ -428,6 +428,53 @@ func (s *Server) handleAcceptUpstream(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, toMutationResponse(result, client.Mode()))
 }
 
+func (s *Server) handleRejectUpstream(w http.ResponseWriter, r *http.Request) {
+	client, ok := s.resolveClient(w, r)
+	if !ok {
+		return
+	}
+	id := r.PathValue("id")
+	var req RejectUpstreamRequest
+	if err := decodeJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON: "+err.Error())
+		return
+	}
+	if req.RigHandle == "" {
+		writeError(w, http.StatusBadRequest, "rig_handle is required")
+		return
+	}
+	if err := client.RejectUpstream(id, req.RigHandle); err != nil {
+		writeMutationError(w, err)
+		return
+	}
+	s.invalidateReadCaches(id)
+	writeJSON(w, http.StatusOK, map[string]string{"status": "rejected"})
+}
+
+func (s *Server) handleCloseUpstream(w http.ResponseWriter, r *http.Request) {
+	client, ok := s.resolveClient(w, r)
+	if !ok {
+		return
+	}
+	id := r.PathValue("id")
+	var req CloseUpstreamRequest
+	if err := decodeJSON(r, &req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid JSON: "+err.Error())
+		return
+	}
+	if req.RigHandle == "" {
+		writeError(w, http.StatusBadRequest, "rig_handle is required")
+		return
+	}
+	result, err := client.CloseUpstream(id, req.RigHandle)
+	if err != nil {
+		writeMutationError(w, err)
+		return
+	}
+	s.invalidateReadCaches(id)
+	writeJSON(w, http.StatusOK, toMutationResponse(result, client.Mode()))
+}
+
 func (s *Server) handleReject(w http.ResponseWriter, r *http.Request) {
 	client, ok := s.resolveClient(w, r)
 	if !ok {
diff --git a/internal/api/routes.go b/internal/api/routes.go
@@ -18,6 +18,8 @@ func (s *Server) registerRoutes() {
 	s.mux.HandleFunc("POST /api/wanted/{id}/done", s.handleDone)
 	s.mux.HandleFunc("POST /api/wanted/{id}/accept", s.handleAccept)
 	s.mux.HandleFunc("POST /api/wanted/{id}/accept-upstream", s.handleAcceptUpstream)
+	s.mux.HandleFunc("POST /api/wanted/{id}/reject-upstream", s.handleRejectUpstream)
+	s.mux.HandleFunc("POST /api/wanted/{id}/close-upstream", s.handleCloseUpstream)
 	s.mux.HandleFunc("POST /api/wanted/{id}/reject", s.handleReject)
 	s.mux.HandleFunc("POST /api/wanted/{id}/close", s.handleClose)
 
diff --git a/internal/api/types.go b/internal/api/types.go
@@ -199,6 +199,16 @@ type AcceptUpstreamRequest struct {
 	Message     string   `json:"message"`
 }
 
+// RejectUpstreamRequest is the JSON body for POST /api/wanted/{id}/reject-upstream.
+type RejectUpstreamRequest struct {
+	RigHandle string `json:"rig_handle"`
+}
+
+// CloseUpstreamRequest is the JSON body for POST /api/wanted/{id}/close-upstream.
+type CloseUpstreamRequest struct {
+	RigHandle string `json:"rig_handle"`
+}
+
 // RejectRequest is the JSON body for POST /api/wanted/{id}/reject.
 type RejectRequest struct {
 	Reason string `json:"reason"`
@@ -284,6 +294,18 @@ func toDetailResponse(d *sdk.DetailResult, mode string) *DetailResponse {
 		actions[i] = commons.TransitionName(t)
 	}
 	var upstreamPRs []UpstreamPRJSON
+
+	// If the item is in_review and has a completion on main, include it as
+	// the first entry so the poster sees all submissions in one place.
+	if d.Item != nil && d.Item.Status == "in_review" && d.Completion != nil {
+		upstreamPRs = append(upstreamPRs, UpstreamPRJSON{
+			RigHandle:   d.Completion.CompletedBy,
+			Status:      "in_review",
+			CompletedBy: d.Completion.CompletedBy,
+			Evidence:    d.Completion.Evidence,
+		})
+	}
+
 	for _, p := range d.UpstreamPRs {
 		delta := ""
 		if p.Status != "" && d.Item != nil && p.Status != d.Item.Status {
diff --git a/internal/commons/commons.go b/internal/commons/commons.go
@@ -693,6 +693,27 @@ func AcceptUpstreamDML(wantedID, completionID, completedBy, evidence, rigHandle,
 	return []string{deleteCompletion, insertCompletion, updateWanted, insertStamp, updateCompletion}
 }
 
+// CloseUpstreamDML returns the pure DML statements for adopting a fork submission
+// without creating a stamp. Statements: DELETE existing completion, INSERT fork
+// completion, UPDATE wanted to completed.
+func CloseUpstreamDML(wantedID, completionID, completedBy, evidence, hopURI string) []string {
+	hopField := "NULL"
+	if hopURI != "" {
+		hopField = fmt.Sprintf("'%s'", EscapeSQL(hopURI))
+	}
+
+	deleteCompletion := fmt.Sprintf(`DELETE FROM completions WHERE wanted_id='%s'`,
+		EscapeSQL(wantedID))
+
+	insertCompletion := fmt.Sprintf(`INSERT IGNORE INTO completions (id, wanted_id, completed_by, evidence, hop_uri, completed_at) VALUES ('%s', '%s', '%s', '%s', %s, NOW())`,
+		EscapeSQL(completionID), EscapeSQL(wantedID), EscapeSQL(completedBy), EscapeSQL(evidence), hopField)
+
+	updateWanted := fmt.Sprintf(`UPDATE wanted SET status='completed', claimed_by='%s', evidence_url='%s', updated_at=NOW() WHERE id='%s'`,
+		EscapeSQL(completedBy), EscapeSQL(evidence), EscapeSQL(wantedID))
+
+	return []string{deleteCompletion, insertCompletion, updateWanted}
+}
+
 // AcceptCompletion validates a completion, creates a stamp, and marks the item completed.
 func AcceptCompletion(db DB, wantedID, completionID, rigHandle, hopURI string, stamp *Stamp, signed bool) error {
 	stmts := AcceptCompletionDML(wantedID, completionID, rigHandle, hopURI, stamp)
diff --git a/internal/hosted/resolver.go b/internal/hosted/resolver.go
@@ -243,6 +243,13 @@ func (wr *WorkspaceResolver) buildClient(wl *WastelandConfig, rigHandle, connect
 			}
 			return provider.ClosePR(upOrg, upDB, prID)
 		},
+		CloseUpstreamPR: func(prURL string) error {
+			prID := extractPRID(prURL)
+			if prID == "" {
+				return fmt.Errorf("cannot extract PR ID from URL: %s", prURL)
+			}
+			return provider.ClosePR(upOrg, upDB, prID)
+		},
 		ListPendingItems: wr.getOrCreatePendingCache(provider, upOrg, upDB).Get,
 		BranchURL:        branchURL,
 		Signing:          wl.Signing,
@@ -277,3 +284,13 @@ func extractWantedIDFromBranch(branch string) string {
 	}
 	return branch
 }
+
+// extractPRID extracts the pull request ID from a DoltHub PR URL like
+// "https://www.dolthub.com/repositories/org/db/pulls/123".
+func extractPRID(prURL string) string {
+	idx := strings.LastIndex(prURL, "/pulls/")
+	if idx < 0 {
+		return ""
+	}
+	return prURL[idx+len("/pulls/"):]
+}
diff --git a/internal/remote/dolthub.go b/internal/remote/dolthub.go
@@ -794,12 +794,14 @@ func (d *DoltHubProvider) ListPendingWantedIDs(upstreamOrg, db string) (map[stri
 			//
 			// Filter rules:
 			// 1. status "open" = untouched item (stale copy)
-			// 2. claimed_by set to someone other than the PR author =
-			//    inherited claim from a previous upstream state
+			// 2. status "claimed" with claimed_by set to someone other than
+			//    the PR author = inherited claim from a previous upstream state
+			// Items at "in_review" or "completed" always pass — those statuses
+			// require intentional action (submitting evidence / accepting).
 			if e.state.Status == "open" {
 				continue
 			}
-			if e.state.ClaimedBy != "" && e.state.ClaimedBy != e.author {
+			if e.state.Status == "claimed" && e.state.ClaimedBy != "" && e.state.ClaimedBy != e.author {
 				continue
 			}
 			ids[e.wantedID] = append(ids[e.wantedID], e.state)
diff --git a/internal/sdk/mutations.go b/internal/sdk/mutations.go
@@ -155,6 +155,79 @@ func (c *Client) AcceptUpstream(wantedID, submitterHandle string, input AcceptIn
 	return c.mutateLocked(wantedID, "wl accept-upstream: "+wantedID, stmts...)
 }
 
+// RejectUpstream declines a fork submission by closing its upstream DoltHub PR.
+// No local state is modified — the item remains in its current status.
+func (c *Client) RejectUpstream(wantedID, submitterHandle string) error {
+	match, err := c.findUpstreamSubmission(wantedID, submitterHandle)
+	if err != nil {
+		return err
+	}
+	if match.PRURL == "" {
+		return fmt.Errorf("submission has no upstream PR to close")
+	}
+	if c.CloseUpstreamPR == nil {
+		return fmt.Errorf("upstream PR closing not available")
+	}
+	return c.CloseUpstreamPR(match.PRURL)
+}
+
+// CloseUpstream adopts a fork submission without creating a stamp, then closes
+// the upstream DoltHub PR.
+func (c *Client) CloseUpstream(wantedID, submitterHandle string) (*MutationResult, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if result := c.prIdempotentLocked(wantedID, "completed"); result != nil {
+		return result, nil
+	}
+
+	match, err := c.findUpstreamSubmission(wantedID, submitterHandle)
+	if err != nil {
+		return nil, err
+	}
+
+	if match.Status != "in_review" {
+		return nil, fmt.Errorf("submission is not in review")
+	}
+	if match.CompletedBy == "" {
+		return nil, fmt.Errorf("submission has no completion data")
+	}
+	if submitterHandle == c.rigHandle {
+		return nil, fmt.Errorf("cannot close your own completion")
+	}
+
+	completionID := commons.GeneratePrefixedID("c", wantedID, match.CompletedBy)
+	stmts := commons.CloseUpstreamDML(wantedID, completionID, match.CompletedBy, match.Evidence, c.hopURI)
+	result, err := c.mutateLocked(wantedID, "wl close-upstream: "+wantedID, stmts...)
+	if err != nil {
+		return nil, err
+	}
+
+	// Best-effort close the upstream PR.
+	if match.PRURL != "" && c.CloseUpstreamPR != nil {
+		_ = c.CloseUpstreamPR(match.PRURL)
+	}
+	return result, nil
+}
+
+// findUpstreamSubmission looks up a pending upstream submission by rig handle.
+func (c *Client) findUpstreamSubmission(wantedID, submitterHandle string) (*PendingItem, error) {
+	if c.ListPendingItems == nil {
+		return nil, fmt.Errorf("upstream PR listing not available")
+	}
+	pending, err := c.ListPendingItems()
+	if err != nil {
+		return nil, fmt.Errorf("listing pending items: %w", err)
+	}
+	items := pending[wantedID]
+	for i := range items {
+		if items[i].RigHandle == submitterHandle {
+			return &items[i], nil
+		}
+	}
+	return nil, fmt.Errorf("no pending submission from %s", submitterHandle)
+}
+
 // Reject rejects a completion, reverting the item from in_review to claimed.
 func (c *Client) Reject(wantedID, reason string) (*MutationResult, error) {
 	if result := c.prIdempotent(wantedID, "claimed"); result != nil {
diff --git a/internal/sdk/sdk.go b/internal/sdk/sdk.go
@@ -27,6 +27,7 @@ type ClientConfig struct {
 	SaveConfig       func(mode string, signing bool) error
 	ListPendingItems func() (map[string][]PendingItem, error) // returns wanted IDs with pending upstream PR state
 	BranchURL        func(branch string) string               // returns a web URL for the branch
+	CloseUpstreamPR  func(prURL string) error                 // close an upstream PR by its web URL
 }
 
 // Client provides mode-aware operations against the Wasteland wanted board.
@@ -53,6 +54,8 @@ type Client struct {
 	ListPendingItems func() (map[string][]PendingItem, error)
 	// BranchURL returns a web URL for the given branch. Nil disables the feature.
 	BranchURL func(branch string) string
+	// CloseUpstreamPR closes an upstream PR by its web URL. Nil disables the feature.
+	CloseUpstreamPR func(prURL string) error
 }
 
 // New creates a Client from the given config.
@@ -71,6 +74,7 @@ func New(cfg ClientConfig) *Client {
 		SaveConfig:       cfg.SaveConfig,
 		ListPendingItems: cfg.ListPendingItems,
 		BranchURL:        cfg.BranchURL,
+		CloseUpstreamPR:  cfg.CloseUpstreamPR,
 	}
 }
 
@@ -99,5 +103,6 @@ func (c *Client) WithRigHandle(handle string) *Client {
 		SaveConfig:       c.SaveConfig,
 		ListPendingItems: c.ListPendingItems,
 		BranchURL:        c.BranchURL,
+		CloseUpstreamPR:  c.CloseUpstreamPR,
 	}
 }
diff --git a/web/src/api/client.ts b/web/src/api/client.ts
@@ -230,6 +230,22 @@ export async function acceptUpstream(
   });
 }
 
+export async function rejectUpstream(id: string, rigHandle: string): Promise<void> {
+  await request<Record<string, string>>(`/api/wanted/${id}/reject-upstream`, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ rig_handle: rigHandle }),
+  });
+}
+
+export async function closeUpstream(id: string, rigHandle: string): Promise<MutationResponse> {
+  return request<MutationResponse>(`/api/wanted/${id}/close-upstream`, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ rig_handle: rigHandle }),
+  });
+}
+
 export async function deleteItem(id: string): Promise<MutationResponse> {
   return request<MutationResponse>(`/api/wanted/${id}`, { method: "DELETE" });
 }
diff --git a/web/src/components/DetailView.tsx b/web/src/components/DetailView.tsx

Original file line number	Diff line number	Diff line change
`@@ -33,5 +33,6 @@ var newSDKClient = func(cfg federation.Config, noPush bool) (sdk.Client, error`
`33`	`33`	`},`
`34`	`34`	`ListPendingItems: listPendingItemsFromPRs(cfg),`
`35`	`35`	`BranchURL: branchURLCallback(cfg),`
	`36`	`+ CloseUpstreamPR: closeUpstreamPRCallback(cfg),`
`36`	`37`	`}), nil`
`37`	`38`	`}`