diff --git a/internal/app/mcp_scalable_operations.go b/internal/app/mcp_scalable_operations.go index 3d1b512..9573c5f 100644 --- a/internal/app/mcp_scalable_operations.go +++ b/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{ diff --git a/internal/app/mcp_stdio_e2e_test.go b/internal/app/mcp_stdio_e2e_test.go index 0bef9f6..a1705a9 100644 --- a/internal/app/mcp_stdio_e2e_test.go +++ b/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"): diff --git a/internal/corpus/corpus_test.go b/internal/corpus/corpus_test.go index a3fed2b..b78d715 100644 --- a/internal/corpus/corpus_test.go +++ b/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() diff --git a/internal/corpus/observations.go b/internal/corpus/observations.go index 2ed1021..0607752 100644 --- a/internal/corpus/observations.go +++ b/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) } diff --git a/internal/mcpserver/capabilities_test.go b/internal/mcpserver/capabilities_test.go index 482bb3c..1bf6e75 100644 --- a/internal/mcpserver/capabilities_test.go +++ b/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) { diff --git a/internal/mcpserver/catalog_test.go b/internal/mcpserver/catalog_test.go index b5e2ff3..628dc54 100644 --- a/internal/mcpserver/catalog_test.go +++ b/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() diff --git a/internal/mcpserver/scalable.go b/internal/mcpserver/scalable.go index a22507d..a2fecb3 100644 --- a/internal/mcpserver/scalable.go +++ b/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")