From a32d507eb8b9f9e99bc9d3e94d2d7c3977d66821 Mon Sep 17 00:00:00 2001 From: Harshit Singh Bhandari Date: Fri, 3 Jul 2026 20:50:41 +0530 Subject: [PATCH] feat(scm): detect cross-fork PRs by scanning all project remotes (#2368) The SCM observer only polled a project's push origin for open PRs, so a PR opened from the fork against an upstream base repo (the standard fork -> upstream flow) was never discovered: GitHub lists a PR under its base repo, not its head, so an origin-only scan can't see it. Sessions that opened such PRs showed no PR on the dashboard. Scan every GitHub remote in the project checkout (origin + upstreams / mirrors) for open PRs, and attribute a PR to a session only when its head branch lives in that session's push origin. This surfaces cross-fork PRs while preserving the no-misattribution guarantee: a stranger's fork PR has a head repo no session owns and is dropped. Tracked cross-fork PRs are keyed for refresh by their own recorded repo (the upstream base) so the GraphQL refetch targets the right repo. --- backend/internal/observe/scm/observer.go | 142 +++++++++++++++--- backend/internal/observe/scm/observer_test.go | 118 ++++++++++++++- 2 files changed, 240 insertions(+), 20 deletions(-) diff --git a/backend/internal/observe/scm/observer.go b/backend/internal/observe/scm/observer.go index 1ca0fa10e..780e669fc 100644 --- a/backend/internal/observe/scm/observer.go +++ b/backend/internal/observe/scm/observer.go @@ -191,12 +191,18 @@ type subject struct { hasPR bool } -// sessionRepo pairs a live session with its parsed repo and branch for per-repo -// branch-prefix discovery of new (including stacked) pull requests. +// sessionRepo pairs a live session with a repo to scan and its branch for +// per-repo branch-prefix discovery of new (including stacked) pull requests. +// A session is scanned against its push origin plus every other remote in the +// project checkout, so repo is the repo whose open-PR list is listed while +// headRepo is the repo the session's head branch actually lives in (the push +// origin). For same-repo PRs repo == headRepo; for a cross-fork PR (fork head, +// upstream base) repo is the upstream base and headRepo is the fork origin. type sessionRepo struct { - session domain.SessionRecord - repo ports.SCMRepo - branch string + session domain.SessionRecord + repo ports.SCMRepo + headRepo ports.SCMRepo + branch string } type repoGuardState struct { @@ -424,6 +430,8 @@ func (o *Observer) discoverSubjects(ctx context.Context) (map[string]*subject, [ return nil, nil, err } projects := map[domain.ProjectID]domain.ProjectRecord{} + originRepos := map[domain.ProjectID]ports.SCMRepo{} + scanRepos := map[domain.ProjectID][]ports.SCMRepo{} out := map[string]*subject{} var sessionRepos []sessionRepo for _, sess := range sessions { @@ -453,29 +461,87 @@ func (o *Observer) discoverSubjects(ctx context.Context) (map[string]*subject, [ } projects[sess.ProjectID] = p proj = p + if origin, ok := o.provider.ParseRepository(p.RepoOriginURL); ok { + originRepos[sess.ProjectID] = origin + scanRepos[sess.ProjectID] = o.resolveScanRepos(p, origin) + } } - repo, ok := o.provider.ParseRepository(proj.RepoOriginURL) + origin, ok := originRepos[sess.ProjectID] if !ok { o.logger.Debug("scm observer: project has no supported SCM origin", "project", proj.ID, "origin", proj.RepoOriginURL) continue } - sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, branch: branch}) + for _, repo := range scanRepos[sess.ProjectID] { + sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, headRepo: origin, branch: branch}) + } prs, err := o.store.ListPRsBySession(ctx, sess.ID) if err != nil { return nil, nil, err } for _, pr := range openTrackedPRs(prs) { - key := prKey(repo, pr.Number) + // A tracked PR may live on an upstream repo (cross-fork), so its + // refresh subject is keyed by the PR's own recorded repo, not the + // push origin, or the GraphQL refetch would target the wrong repo. + prRepo := subjectRepoForPR(pr, origin) + key := prKey(prRepo, pr.Number) if existing, ok := out[key]; ok { o.logger.Warn("scm observer: duplicate tracked PR ownership skipped", "pr", key, "kept_session", existing.session.ID, "skipped_session", sess.ID) continue } - out[key] = &subject{session: sess, repo: repo, branch: branch, known: pr, hasPR: true} + out[key] = &subject{session: sess, repo: prRepo, branch: branch, known: pr, hasPR: true} } } return out, sessionRepos, nil } +// resolveScanRepos returns the deduped set of repos whose open-PR lists should be +// scanned to attribute PRs to this project's sessions: the push origin plus every +// other GitHub remote configured in the project checkout (upstreams, mirrors). +// Attribution still requires a PR's head branch to live in the origin, so scanning +// extra remotes only surfaces cross-fork PRs (fork head, upstream base) and can +// never misattribute a stranger's PR. +// +// ponytail: remotes are read once per project per process (memoized by the +// caller); a remote added after the daemon started is picked up on restart. Move +// to a git-config watch if that latency ever matters. +func (o *Observer) resolveScanRepos(proj domain.ProjectRecord, origin ports.SCMRepo) []ports.SCMRepo { + repos := []ports.SCMRepo{origin} + if strings.TrimSpace(proj.Path) == "" { + return repos + } + seen := map[string]bool{prKey(origin, 0): true} + for _, url := range gitRemoteURLs(proj.Path) { + repo, ok := o.provider.ParseRepository(url) + if !ok { + continue + } + key := prKey(repo, 0) + if seen[key] { + continue + } + seen[key] = true + repos = append(repos, repo) + } + return repos +} + +// subjectRepoForPR resolves the repo that owns a tracked PR's number for refresh. +// A cross-fork PR lives on the base/upstream repo recorded in pr.Repo rather than +// the push origin, so the refresh/GraphQL fetch must target pr.Repo. Legacy rows +// written before pr.Repo was populated fall back to the origin. +func subjectRepoForPR(pr domain.PullRequest, origin ports.SCMRepo) ports.SCMRepo { + if strings.TrimSpace(pr.Repo) == "" { + return origin + } + return ports.SCMRepo{ + Provider: firstNonEmpty(pr.Provider, origin.Provider), + Host: firstNonEmpty(pr.Host, origin.Host), + Repo: pr.Repo, + Owner: ownerOf(pr.Repo), + Name: nameOf(pr.Repo), + } +} + func openTrackedPRs(prs []domain.PullRequest) []domain.PullRequest { out := make([]domain.PullRequest, 0, len(prs)) for _, pr := range prs { @@ -550,20 +616,19 @@ func (o *Observer) discoverNewPRs(ctx context.Context, sessionRepos []sessionRep if pr.Number <= 0 || pr.SourceBranch == "" { continue } - // Branch-prefix attribution must only claim PRs whose head branch - // lives in the project repo. A fork PR can carry a head branch whose - // name matches an AO session branch; its commits live in the fork, so - // auto-claiming it would misattribute work. Same-repo PRs always - // report the base repo's full name as their head repo, so anything - // else (including an empty head repo from a deleted fork) is skipped. - if !strings.EqualFold(pr.HeadRepo, repoFullName(repo)) { - continue - } key := prKey(repo, pr.Number) if _, ok := subjects[key]; ok { continue } - sr, ok := matchSession(byRepo[repoKey], pr.SourceBranch) + // Branch-prefix attribution must only claim PRs whose head branch + // lives in a session's push origin. A same-repo PR has head == origin + // == this scanned repo; a cross-fork PR (fork head, upstream base) has + // head == origin while this scanned repo is the upstream base. A + // stranger's fork PR carries a head repo no session owns and is + // dropped (as is an empty head repo from a deleted fork), preserving + // the no-misattribution guarantee. + eligible := candidatesForHeadRepo(byRepo[repoKey], pr.HeadRepo) + sr, ok := matchSession(eligible, pr.SourceBranch) if !ok { continue } @@ -612,6 +677,24 @@ func (o *Observer) discoverNewPRs(ctx context.Context, sessionRepos []sessionRep // branches are prefixes of the same source branch the longest (most specific) // one wins, so a child session claims its own stacked PRs rather than the // ancestor session. +// candidatesForHeadRepo narrows the scanned repo's session candidates to those +// whose head branch lives in headRepo (the PR's head repository full name). This +// is the fork guard: a PR is only attributable when its head repo equals a +// session's push origin, whether the PR was found on the origin itself or on a +// scanned upstream base repo. +func candidatesForHeadRepo(candidates []sessionRepo, headRepo string) []sessionRepo { + if strings.TrimSpace(headRepo) == "" { + return nil + } + var out []sessionRepo + for _, sr := range candidates { + if strings.EqualFold(repoFullName(sr.headRepo), headRepo) { + out = append(out, sr) + } + } + return out +} + func matchSession(candidates []sessionRepo, sourceBranch string) (sessionRepo, bool) { var best sessionRepo bestLen := -1 @@ -1232,6 +1315,27 @@ func resolveGitOriginURL(path string) string { return strings.TrimSpace(string(out)) } +// gitRemoteURLs lists the fetch URL of every git remote configured at path. It +// returns nil on any error (missing repo, no git, no remotes). The observer uses +// it to scan upstream/mirror remotes for cross-fork PRs in addition to origin. +func gitRemoteURLs(path string) []string { + out, err := aoprocess.Command("git", "-C", path, "remote").Output() + if err != nil { + return nil + } + var urls []string + for _, name := range strings.Fields(string(out)) { + u, err := aoprocess.Command("git", "-C", path, "remote", "get-url", name).Output() + if err != nil { + continue + } + if s := strings.TrimSpace(string(u)); s != "" { + urls = append(urls, s) + } + } + return urls +} + func scrubLine(s string) string { s = strings.ReplaceAll(s, "\n", " ") s = strings.ReplaceAll(s, "\r", " ") diff --git a/backend/internal/observe/scm/observer_test.go b/backend/internal/observe/scm/observer_test.go index dec26750a..f07b1fef5 100644 --- a/backend/internal/observe/scm/observer_test.go +++ b/backend/internal/observe/scm/observer_test.go @@ -139,7 +139,18 @@ func (p *fakeProvider) SCMCredentialsAvailable(context.Context) (bool, error) { } func (p *fakeProvider) ParseRepository(remote string) (ports.SCMRepo, bool) { - return testRepo, remote != "" + remote = strings.TrimSpace(remote) + if remote == "" { + return ports.SCMRepo{}, false + } + s := strings.TrimSuffix(remote, ".git") + s = strings.TrimPrefix(s, "https://github.com/") + s = strings.TrimPrefix(s, "git@github.com:") + parts := strings.Split(strings.Trim(s, "/"), "/") + if len(parts) != 2 || parts[0] == "" || parts[1] == "" { + return ports.SCMRepo{}, false + } + return ports.SCMRepo{Provider: "github", Host: "github.com", Owner: parts[0], Name: parts[1], Repo: parts[0] + "/" + parts[1]}, true } func (p *fakeProvider) RepoPRListGuard(_ context.Context, repo ports.SCMRepo, _ string) (ports.SCMGuardResult, error) { p.mu.Lock() @@ -540,6 +551,111 @@ func TestPoll_IgnoresForkPRWithMatchingBranch(t *testing.T) { } } +func mustGit(t *testing.T, args ...string) { + t.Helper() + if out, err := exec.Command("git", args...).CombinedOutput(); err != nil { + t.Fatalf("git %v: %v (%s)", args, err, out) + } +} + +// A PR opened from the fork push origin against an upstream base repo (the +// standard fork -> upstream contribution flow) must be discovered and attributed +// by scanning every remote in the project checkout, not just origin. Its head +// lives in origin (o/r) while its base lives in upstream (up/r); the persisted +// row records the upstream base repo so refresh refetches against it. +func TestPoll_DiscoversCrossForkPRFromUpstreamRemote(t *testing.T) { + dir := t.TempDir() + mustGit(t, "init", dir) + mustGit(t, "-C", dir, "remote", "add", "origin", "https://github.com/o/r.git") + mustGit(t, "-C", dir, "remote", "add", "upstream", "https://github.com/up/r.git") + + upstream := ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "up", Name: "r", Repo: "up/r"} + store := &fakeStore{ + sessions: []domain.SessionRecord{{ID: "p-1", ProjectID: "p", Metadata: domain.SessionMetadata{Branch: "ao/p-1/root"}}}, + projects: map[string]domain.ProjectRecord{"p": {ID: "p", Path: dir, RepoOriginURL: "https://github.com/o/r.git"}}, + prs: map[domain.SessionID][]domain.PullRequest{}, + checks: map[string][]domain.PullRequestCheck{}, + } + crossObs := ports.SCMObservation{ + Fetched: true, Provider: "github", Host: "github.com", Repo: "up/r", + PR: ports.SCMPRObservation{URL: "https://github.com/up/r/pull/1", Number: 1, State: "open", SourceBranch: "ao/p-1/feat", HeadRepo: "o/r", TargetBranch: "main", HeadSHA: "sha1", Title: "PR"}, + CI: ports.SCMCIObservation{Summary: string(domain.CIPassing), HeadSHA: "sha1"}, + Review: ports.SCMReviewObservation{Decision: string(domain.ReviewNone)}, + Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable), Mergeable: true}, + } + provider := &fakeProvider{ + repoGuards: map[string]ports.SCMGuardResult{prKey(testRepo, 0): {ETag: "origin"}, prKey(upstream, 0): {ETag: "up"}}, + openPRs: map[string][]ports.SCMPRObservation{ + prKey(upstream, 0): {{URL: "https://github.com/up/r/pull/1", Number: 1, SourceBranch: "ao/p-1/feat", HeadRepo: "o/r", TargetBranch: "main", HeadSHA: "sha1"}}, + }, + observations: map[string]ports.SCMObservation{prKey(upstream, 1): crossObs}, + } + lc := &fakeLifecycle{} + obs := newTestObserver(store, provider, lc, time.Unix(1, 0).UTC()) + if err := obs.Poll(context.Background()); err != nil { + t.Fatal(err) + } + if len(store.writes) == 0 { + t.Fatal("expected cross-fork PR write") + } + got := store.writes[0].pr + if got.SessionID != "p-1" { + t.Fatalf("session id = %q, want p-1", got.SessionID) + } + if got.Repo != "up/r" { + t.Fatalf("pr repo = %q, want up/r (upstream base)", got.Repo) + } + if got.SourceBranch != "ao/p-1/feat" { + t.Fatalf("source branch = %q, want ao/p-1/feat", got.SourceBranch) + } + fetched := false + for _, batch := range provider.fetchBatches { + for _, ref := range batch { + if ref.Repo.Repo == "up/r" && ref.Number == 1 { + fetched = true + } + } + } + if !fetched { + t.Fatalf("cross-fork PR must be refreshed against upstream, batches=%#v", provider.fetchBatches) + } +} + +// A PR on a scanned upstream remote whose head lives in some third-party fork +// (not this project's origin) must never be attributed, even though its branch +// name matches a session. Scanning extra remotes stays safe. +func TestPoll_IgnoresUpstreamPRFromForeignHead(t *testing.T) { + dir := t.TempDir() + mustGit(t, "init", dir) + mustGit(t, "-C", dir, "remote", "add", "origin", "https://github.com/o/r.git") + mustGit(t, "-C", dir, "remote", "add", "upstream", "https://github.com/up/r.git") + + upstream := ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "up", Name: "r", Repo: "up/r"} + store := &fakeStore{ + sessions: []domain.SessionRecord{{ID: "p-1", ProjectID: "p", Metadata: domain.SessionMetadata{Branch: "ao/p-1/root"}}}, + projects: map[string]domain.ProjectRecord{"p": {ID: "p", Path: dir, RepoOriginURL: "https://github.com/o/r.git"}}, + prs: map[domain.SessionID][]domain.PullRequest{}, + checks: map[string][]domain.PullRequestCheck{}, + } + provider := &fakeProvider{ + repoGuards: map[string]ports.SCMGuardResult{prKey(testRepo, 0): {ETag: "origin"}, prKey(upstream, 0): {ETag: "up"}}, + openPRs: map[string][]ports.SCMPRObservation{ + prKey(upstream, 0): {{URL: "https://github.com/up/r/pull/9", Number: 9, SourceBranch: "ao/p-1/feat", HeadRepo: "stranger/r", TargetBranch: "main", HeadSHA: "sha9"}}, + }, + observations: map[string]ports.SCMObservation{}, + } + obs := newTestObserver(store, provider, &fakeLifecycle{}, time.Unix(1, 0).UTC()) + if err := obs.Poll(context.Background()); err != nil { + t.Fatal(err) + } + if len(provider.fetchBatches) != 0 { + t.Fatalf("foreign-head upstream PR must not be fetched, got %#v", provider.fetchBatches) + } + if len(store.writes) != 0 { + t.Fatalf("foreign-head upstream PR must not be persisted, got %d writes", len(store.writes)) + } +} + // A newly discovered open PR is persisted as a baseline row during discovery, // before the refresh/lifecycle pass. This is what lets a same-poll terminal // observation for a sibling PR see the open PR in the store and avoid completing