fix(mcp): accept exact thread sync selectors

internal/app/mcp_scalable_operations.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"strings"
 	"sync"
 	"time"
@@ -58,11 +59,15 @@ func (r *MCPReader) SyncThreads(ctx context.Context, in mcpserver.SyncThreadsInp
 	if in.Selection == "threads" && (len(in.Threads) < 1 || len(in.Threads) > 100) {
 		return mcpserver.JobReference{}, errors.New("threads must contain 1 to 100 items")
 	}
-	if in.LimitPerRepository == 0 {
-		in.LimitPerRepository = 100
-	}
-	if in.LimitPerRepository < 1 || in.LimitPerRepository > 1000 {
-		return mcpserver.JobReference{}, errors.New("limit_per_repository must be between 1 and 1000")
+	if in.Selection == "repositories" {
+		if in.LimitPerRepository == 0 {
+			in.LimitPerRepository = 100
+		}
+		if in.LimitPerRepository < 1 || in.LimitPerRepository > 1000 {
+			return mcpserver.JobReference{}, errors.New("limit_per_repository must be between 1 and 1000")
+		}
+	} else if in.LimitPerRepository != 0 {
+		return mcpserver.JobReference{}, errors.New("limit_per_repository is only valid in repository selection mode")
 	}
 	var err error
 	in.MaxRequests, err = normalizeSyncBatchMaxRequests(in.MaxRequests)
@@ -84,10 +89,11 @@ func (r *MCPReader) SyncThreads(ctx context.Context, in mcpserver.SyncThreadsInp
 //nolint:gocognit
 func (s *Service) syncThreadsBatch(ctx context.Context, in mcpserver.SyncThreadsInput, report func(string, string) error) (map[string]any, error) {
 	type task struct {
-		key         string
-		ref         cli.RepoRef
-		numbers     []int
-		maxRequests int
+		key          string
+		ref          cli.RepoRef
+		numbers      []int
+		inputIndexes []int
+		maxRequests  int
 	}
 	var tasks []task
 	if in.Selection == "repositories" {
@@ -96,7 +102,7 @@ func (s *Service) syncThreadsBatch(ctx context.Context, in mcpserver.SyncThreads
 		}
 	} else {
 		grouped := make(map[string]int)
-		for _, thread := range in.Threads {
+		for inputIndex, thread := range in.Threads {
 			key := thread.Owner + "/" + thread.Repo
 			index, ok := grouped[key]
 			if !ok {
@@ -105,9 +111,14 @@ func (s *Service) syncThreadsBatch(ctx context.Context, in mcpserver.SyncThreads
 				index = len(tasks) - 1
 			}
 			tasks[index].numbers = append(tasks[index].numbers, thread.Number)
+			tasks[index].inputIndexes = append(tasks[index].inputIndexes, inputIndex)
 		}
 	}
-	if err := report("thread_headers", jobProgressCounts(0, len(tasks))); err != nil {
+	resultCount := len(tasks)
+	if in.Selection == "threads" {
+		resultCount = len(in.Threads)
+	}
+	if err := report("thread_headers", jobProgressCounts(0, resultCount)); err != nil {
 		return nil, err
 	}
 	state := in.State
@@ -130,7 +141,7 @@ func (s *Service) syncThreadsBatch(ctx context.Context, in mcpserver.SyncThreads
 		}
 		since = parsed
 	}
-	results := make([]map[string]any, len(tasks))
+	taskResults := make([]map[string]any, len(tasks))
 	remainingRequests := in.MaxRequests
 	plannedRequests := 0
 	runnable := make([]int, 0, len(tasks))
@@ -141,7 +152,7 @@ func (s *Service) syncThreadsBatch(ctx context.Context, in mcpserver.SyncThreads
 		}
 		required := syncFixedRequestCost() + threadRequests
 		if required > remainingRequests {
-			results[index] = syncRequestBudgetUnavailable(tasks[index].key, required, remainingRequests)
+			taskResults[index] = syncRequestBudgetUnavailable(tasks[index].key, required, remainingRequests)
 			continue
 		}
 		tasks[index].maxRequests = required
@@ -169,14 +180,14 @@ func (s *Service) syncThreadsBatch(ctx context.Context, in mcpserver.SyncThreads
 				res, err := s.SyncWithOptions(ctx, current.ref, opts)
 				if err != nil {
 					status, reason, message, retry := githubBatchError(err)
-					results[index] = map[string]any{"key": current.key, "status": status, "reason": reason, "message": message, "retry_after_ms": retry}
+					taskResults[index] = map[string]any{"key": current.key, "status": status, "reason": reason, "message": message, "retry_after_ms": retry}
 					continue
 				}
 				status := "complete"
 				if res.Capped {
 					status = "partial"
 				}
-				results[index] = map[string]any{"key": current.key, "status": status, "updated": res.Updated, "requests": res.Requests, "request_capped": res.Capped, "message": res.Message}
+				taskResults[index] = map[string]any{"key": current.key, "status": status, "updated": res.Updated, "requests": res.Requests, "request_capped": res.Capped, "message": res.Message}
 			}
 		}()
 	}
@@ -191,20 +202,36 @@ func (s *Service) syncThreadsBatch(ctx context.Context, in mcpserver.SyncThreads
 	}
 	close(jobs)
 	wg.Wait()
+	results := taskResults
+	if in.Selection == "threads" {
+		results = make([]map[string]any, len(in.Threads))
+		for taskIndex, current := range tasks {
+			for _, inputIndex := range current.inputIndexes {
+				item := maps.Clone(taskResults[taskIndex])
+				delete(item, "requests")
+				delete(item, "updated")
+				thread := in.Threads[inputIndex]
+				item["key"] = fmt.Sprintf("%s/%s#%d", thread.Owner, thread.Repo, thread.Number)
+				results[inputIndex] = item
+			}
+		}
+	}
 	status := "complete"
 	completed := 0
 	requests := 0
-	for _, result := range results {
+	for _, result := range taskResults {
 		if count, ok := result["requests"].(int); ok {
 			requests += count
 		}
+	}
+	for _, result := range results {
 		if result["status"] == "complete" {
 			completed++
 		} else {
 			status = "partial"
 		}
 	}
-	if err := report("thread_headers", jobProgressCounts(len(tasks), len(tasks))); err != nil {
+	if err := report("thread_headers", jobProgressCounts(resultCount, resultCount)); err != nil {
 		return nil, err
 	}
 	return map[string]any{

internal/app/mcp_stdio_e2e_test.go

@@ -189,6 +189,83 @@ func TestMCPStdioPullRequestPortfolioFlow(t *testing.T) {
 	}
 }
 
+func TestMCPStdioExactThreadSyncFlow(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	home := t.TempDir()
+	seedMCPStdioEmptyCorpus(ctx, t, home)
+	githubServer := newMCPGitHubServer(t)
+	defer githubServer.Close()
+	command := exec.Command(os.Args[0], "-test.run=^TestMCPStdioHelper$")
+	command.Env = append(os.Environ(), mcpE2EHomeEnv+"="+home, mcpE2EGitHubEnv+"="+githubServer.URL+"/")
+	client := mcp.NewClient(&mcp.Implementation{Name: "gitcontribute-e2e", Version: "test"}, nil)
+	session, err := client.Connect(ctx, &mcp.CommandTransport{Command: command}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer session.Close()
+
+	threads := []any{
+		map[string]any{"owner": "lab", "repo": "project", "kind": "issue", "number": 8},
+		map[string]any{"owner": "lab", "repo": "project", "kind": "pull_request", "number": 7},
+	}
+	for range 2 {
+		job := callMCPTool[mcpserver.JobReference](ctx, t, session, mcpserver.ToolSyncThreads, map[string]any{
+			"selection": "threads", "threads": threads,
+		})
+		waitMCPJob(ctx, t, session, job.ID)
+		detailed := callMCPTool[mcpserver.GetJobsOutput](ctx, t, session, mcpserver.ToolGetJob, map[string]any{
+			"ids": []string{job.ID}, "response_format": "detailed",
+		})
+		assertExactThreadJobItems(t, detailed, []string{"lab/project#8", "lab/project#7"})
+	}
+	inspection, err := New(config.NewPaths(&config.Env{Home: home}), "e2e", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	inventory, err := inspection.ListCorpusInventory(ctx)
+	if closeErr := inspection.Close(); err == nil {
+		err = closeErr
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(inventory.Repositories) != 1 || inventory.Repositories[0].ThreadObservations != 2 {
+		t.Fatalf("repeated exact sync duplicated observations: %+v", inventory.Repositories)
+	}
+
+	stored := callMCPTool[mcpserver.GetThreadsOutput](ctx, t, session, mcpserver.ToolGetThreads, map[string]any{"threads": threads})
+	if len(stored.Items) != 2 || stored.Items[0].Value == nil || stored.Items[0].Value.Kind != "issue" || stored.Items[1].Value == nil || stored.Items[1].Value.Kind != "pull_request" {
+		t.Fatalf("exact stored threads = %+v", stored)
+	}
+	status := callMCPTool[mcpserver.JobReference](ctx, t, session, mcpserver.ToolSyncPullRequestStatus, map[string]any{
+		"pull_requests": []any{map[string]any{"owner": "lab", "repo": "project", "kind": "pull_request", "number": 7}},
+	})
+	waitMCPJob(ctx, t, session, status.ID)
+}
+
+func assertExactThreadJobItems(t *testing.T, jobs mcpserver.GetJobsOutput, wantKeys []string) {
+	t.Helper()
+	if len(jobs.Items) != 1 || jobs.Items[0].Value == nil {
+		t.Fatalf("detailed job response = %+v", jobs)
+	}
+	result, ok := jobs.Items[0].Value.Result.(map[string]any)
+	if !ok {
+		t.Fatalf("job result type = %T", jobs.Items[0].Value.Result)
+	}
+	items, ok := result["items"].([]any)
+	if !ok || len(items) != len(wantKeys) {
+		t.Fatalf("job items = %#v", result["items"])
+	}
+	for i, raw := range items {
+		item, ok := raw.(map[string]any)
+		if !ok || item["key"] != wantKeys[i] || item["status"] != "complete" {
+			t.Fatalf("job item %d = %#v, want complete %s", i, raw, wantKeys[i])
+		}
+	}
+}
+
 func seedMCPStdioCorpus(ctx context.Context, t *testing.T, home string) {
 	t.Helper()
 	svc, err := New(config.NewPaths(&config.Env{Home: home}), "e2e", nil)
@@ -300,6 +377,14 @@ func newMCPGitHubServer(t *testing.T) *httptest.Server {
 				"pull_request":{"url":"https://api.github.test/repos/lab/project/pulls/7","html_url":"https://github.com/lab/project/pull/7"},
 				"created_at":"2026-07-15T10:00:00Z","updated_at":"2026-07-18T20:00:00Z"
 			}`))
+		case strings.HasSuffix(r.URL.Path, "/repos/lab/project/issues/8"):
+			_, _ = w.Write([]byte(`{
+				"id":800,"node_id":"I_8","number":8,"state":"open","title":"Document exact sync",
+				"body":"Keep refresh scope narrow","user":{"login":"morluto"},
+				"repository_url":"https://api.github.test/repos/lab/project",
+				"html_url":"https://github.com/lab/project/issues/8",
+				"created_at":"2026-07-15T10:00:00Z","updated_at":"2026-07-18T20:00:00Z"
+			}`))
 		case strings.HasSuffix(r.URL.Path, "/repos/lab/project/pulls/7/reviews"):
 			_, _ = w.Write([]byte(`[{"id":701,"node_id":"R_701","state":"APPROVED","user":{"login":"reviewer"},"commit_id":"head123","submitted_at":"2026-07-18T21:00:00Z"}]`))
 		case strings.HasSuffix(r.URL.Path, "/repos/lab/project/pulls/7"):

internal/corpus/corpus_test.go

@@ -410,6 +410,33 @@ func TestThreadDelayedObservations(t *testing.T) {
 	}
 }
 
+func TestThreadObservationReplayIsIdempotent(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	c, _ := openTestCorpus(t)
+	repo, err := c.ApplyRepositoryObservation(ctx, "owner", "repo", "1", time.Unix(1, 0).UTC(), `{}`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sourceUpdatedAt := time.Unix(1000, 0).UTC()
+	for range 2 {
+		if _, err := c.ApplyThreadObservation(ctx, repo.ID, ThreadKindIssue, 1, "open", "title", "body", "author", sourceUpdatedAt, `{"id":1}`); err != nil {
+			t.Fatal(err)
+		}
+	}
+	thread, err := c.GetThread(ctx, repo.ID, ThreadKindIssue, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	observations, err := c.ListThreadObservations(ctx, thread.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(observations) != 1 {
+		t.Fatalf("replayed observations = %d, want 1", len(observations))
+	}
+}
+
 func TestThreadEqualTimestampSequenceOrdering(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()

internal/corpus/observations.go

@@ -285,8 +285,12 @@ func (c *Corpus) UpsertThread(ctx context.Context, thread Thread, payload string
 
 	if _, err := tx.ExecContext(ctx, `
 		INSERT INTO thread_observations (thread_id, source_updated_at, observation_sequence, payload, observed_at)
-		VALUES (?, ?, ?, ?, ?)
-	`, threadID, srcSec, seq, payload, now); err != nil {
+		SELECT ?, ?, ?, ?, ?
+		WHERE NOT EXISTS (
+			SELECT 1 FROM thread_observations
+			WHERE thread_id = ? AND source_updated_at = ? AND payload = ?
+		)
+	`, threadID, srcSec, seq, payload, now, threadID, srcSec, payload); err != nil {
 		return nil, fmt.Errorf("insert thread observation: %w", err)
 	}
 

internal/mcpserver/capabilities_test.go

@@ -2,7 +2,10 @@ package mcpserver
 
 import "context"
 
-type fakeOptionalCapabilities struct{ base *fakeReader }
+type fakeOptionalCapabilities struct {
+	base             *fakeReader
+	syncThreadsInput SyncThreadsInput
+}
 
 func (*fakeOptionalCapabilities) FindNeighbors(context.Context, FindNeighborsInput) (FindNeighborsOutput, error) {
 	return FindNeighborsOutput{}, nil
@@ -42,7 +45,8 @@ func (f *fakeOptionalCapabilities) SearchGitHubRepositories(ctx context.Context,
 func (*fakeOptionalCapabilities) SyncRepositoryMetadata(context.Context, SyncRepositoryMetadataInput) (JobReference, error) {
 	return JobReference{ID: "job-metadata", Status: "queued"}, nil
 }
-func (*fakeOptionalCapabilities) SyncThreads(context.Context, SyncThreadsInput) (JobReference, error) {
+func (f *fakeOptionalCapabilities) SyncThreads(_ context.Context, in SyncThreadsInput) (JobReference, error) {
+	f.syncThreadsInput = in
 	return JobReference{ID: "job-threads", Status: "queued"}, nil
 }
 func (*fakeOptionalCapabilities) HydrateThreads(context.Context, HydrateThreadsInput) (JobReference, error) {

internal/mcpserver/catalog_test.go

@@ -412,6 +412,57 @@ func TestInvalidToolCallEvaluation(t *testing.T) {
 	}
 }
 
+func TestSyncThreadsDefaultsRemainSelectionSpecific(t *testing.T) {
+	base := &fakeReader{searchStarted: make(chan struct{})}
+	optional := &fakeOptionalCapabilities{base: base}
+	reader := completeTestReader{
+		Reader: base, NeighborReader: optional, ScalableReader: optional,
+		PortfolioReader: optional, GitHubOperator: optional, CodeIndexer: optional,
+		MergeConflictReader: optional, ResearchReader: optional,
+		PortfolioOperator: optional, Operator: base,
+	}
+	client, closeSessions := connect(t, reader)
+	defer closeSessions()
+
+	thread := map[string]any{"owner": "acme", "repo": "rocket", "kind": "pull_request", "number": 7}
+	result, err := client.CallTool(context.Background(), &mcp.CallToolParams{
+		Name: ToolSyncThreads, Arguments: map[string]any{"selection": "threads", "threads": []any{thread}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.IsError {
+		t.Fatalf("exact thread sync rejected: %+v", result.Content)
+	}
+	if optional.syncThreadsInput.LimitPerRepository != 0 {
+		t.Fatalf("thread mode received repository default: %+v", optional.syncThreadsInput)
+	}
+
+	result, err = client.CallTool(context.Background(), &mcp.CallToolParams{
+		Name: ToolSyncThreads, Arguments: map[string]any{
+			"selection": "threads", "threads": []any{thread}, "limit_per_repository": 10,
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !result.IsError {
+		t.Fatal("thread mode accepted explicit repository-only limit")
+	}
+
+	result, err = client.CallTool(context.Background(), &mcp.CallToolParams{
+		Name: ToolSyncThreads, Arguments: map[string]any{
+			"selection": "repositories", "repositories": []any{map[string]any{"owner": "acme", "repo": "rocket"}},
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.IsError || optional.syncThreadsInput.LimitPerRepository != 100 {
+		t.Fatalf("repository mode default = %+v, result = %+v", optional.syncThreadsInput, result.Content)
+	}
+}
+
 func TestSideEffectAuthorizationEvaluation(t *testing.T) {
 	tools, closeSessions := listedTools(t)
 	defer closeSessions()

internal/mcpserver/scalable.go

@@ -419,7 +419,6 @@ func (s *Server) registerScalable() {
 		setEnum(sc, "kind", "issue", "pull_request", "both")
 		setEnum(sc, "state", "open", "closed", "all")
 		setRange(sc, "limit_per_repository", 1, 1000)
-		setDefault(sc, "limit_per_repository", 100)
 		setRange(sc, "max_requests", 9, 1000)
 		setDefault(sc, "max_requests", 1000)
 	}), output: outputSchema[JobReference]("Reference to a bounded thread-header synchronization job."), handler: s.syncThreads})
@@ -630,6 +629,14 @@ func (s *Server) syncThreads(ctx context.Context, _ *mcp.CallToolRequest, in Syn
 	if in.Selection == "threads" && (len(in.Repositories) > 0 || in.Kind != "" || in.State != "" || in.UpdatedAfter != "" || in.LimitPerRepository != 0) {
 		return nil, JobReference{}, InvalidArgument("selection", "repository filters are not accepted in thread selection mode", nil)
 	}
+	for _, thread := range in.Threads {
+		if err := validateThreadRef(thread, false); err != nil {
+			return nil, JobReference{}, err
+		}
+	}
+	if in.Selection == "repositories" && in.LimitPerRepository == 0 {
+		in.LimitPerRepository = 100
+	}
 	op, ok := s.reader.(GitHubOperator)
 	if !ok {
 		return nil, JobReference{}, errors.New("batch thread sync is not available")