diff --git a/backend/internal/observe/scm/observer.go b/backend/internal/observe/scm/observer.go index 780e669fc..f61d3d6ad 100644 --- a/backend/internal/observe/scm/observer.go +++ b/backend/internal/observe/scm/observer.go @@ -51,6 +51,7 @@ type Store interface { ListAllSessions(ctx context.Context) ([]domain.SessionRecord, error) GetProject(ctx context.Context, id string) (domain.ProjectRecord, bool, error) UpsertProject(ctx context.Context, row domain.ProjectRecord) error + ListWorkspaceRepos(ctx context.Context, projectID string) ([]domain.WorkspaceRepoRecord, error) ListPRsBySession(ctx context.Context, sessionID domain.SessionID) ([]domain.PullRequest, error) ListChecks(ctx context.Context, prURL string) ([]domain.PullRequestCheck, error) WriteSCMObservation(ctx context.Context, pr domain.PullRequest, checks []domain.PullRequestCheck, reviews []domain.PullRequestReview, threads []domain.PullRequestReviewThread, comments []domain.PullRequestComment, reviewMode ports.ReviewWriteMode) error @@ -466,23 +467,35 @@ func (o *Observer) discoverSubjects(ctx context.Context) (map[string]*subject, [ scanRepos[sess.ProjectID] = o.resolveScanRepos(p, origin) } } - 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 + repos := make([]ports.SCMRepo, 0, len(scanRepos[sess.ProjectID])) + if origin, ok := originRepos[sess.ProjectID]; ok { + for _, repo := range scanRepos[sess.ProjectID] { + sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, headRepo: origin, branch: branch}) + repos = append(repos, repo) + } } - for _, repo := range scanRepos[sess.ProjectID] { - sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, headRepo: origin, branch: branch}) + childRepos, err := o.workspaceSCMRepos(ctx, proj) + if err != nil { + return nil, nil, err + } + for _, repo := range childRepos { + sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, headRepo: repo, branch: branch}) + repos = append(repos, repo) + } + if len(repos) == 0 { + o.logger.Debug("scm observer: project has no supported SCM origins", "project", proj.ID) + continue } prs, err := o.store.ListPRsBySession(ctx, sess.ID) if err != nil { return nil, nil, err } for _, pr := range openTrackedPRs(prs) { - // 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) + prRepo, ok := repoForTrackedPR(pr, repos) + if !ok { + o.logger.Warn("scm observer: tracked PR repo no longer belongs to project", "session", sess.ID, "pr", pr.URL, "repo", pr.Repo) + continue + } 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) @@ -525,21 +538,55 @@ func (o *Observer) resolveScanRepos(proj domain.ProjectRecord, origin ports.SCMR 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 +func (o *Observer) workspaceSCMRepos(ctx context.Context, proj domain.ProjectRecord) ([]ports.SCMRepo, error) { + if proj.Kind.WithDefault() != domain.ProjectKindWorkspace { + return nil, nil } - 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), + childRepos, err := o.store.ListWorkspaceRepos(ctx, proj.ID) + if err != nil { + return nil, err } + repos := make([]ports.SCMRepo, 0, len(childRepos)) + seen := map[string]bool{} + for _, child := range childRepos { + if strings.TrimSpace(child.RepoOriginURL) == "" { + continue + } + repo, ok := o.provider.ParseRepository(child.RepoOriginURL) + if !ok { + o.logger.Debug("scm observer: unsupported SCM origin", "project", proj.ID, "repo", child.Name, "origin", child.RepoOriginURL) + continue + } + key := prKey(repo, 0) + if seen[key] { + continue + } + seen[key] = true + repos = append(repos, repo) + } + return repos, nil +} + +func repoForTrackedPR(pr domain.PullRequest, repos []ports.SCMRepo) (ports.SCMRepo, bool) { + if pr.Provider != "" || pr.Host != "" || pr.Repo != "" { + for _, repo := range repos { + if strings.EqualFold(pr.Provider, repo.Provider) && + strings.EqualFold(pr.Host, repo.Host) && + strings.EqualFold(pr.Repo, repoFullName(repo)) { + return repo, true + } + } + return ports.SCMRepo{}, false + } + if len(repos) == 1 { + return repos[0], true + } + for _, repo := range repos { + if strings.EqualFold(repo.Repo, repos[0].Repo) { + return repo, true + } + } + return repos[0], len(repos) > 0 } func openTrackedPRs(prs []domain.PullRequest) []domain.PullRequest { diff --git a/backend/internal/observe/scm/observer_test.go b/backend/internal/observe/scm/observer_test.go index d75d217a7..17761e143 100644 --- a/backend/internal/observe/scm/observer_test.go +++ b/backend/internal/observe/scm/observer_test.go @@ -21,16 +21,20 @@ import ( "github.com/aoagents/agent-orchestrator/backend/internal/ports" ) -var testRepo = ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "o", Name: "r", Repo: "o/r"} +var ( + testRepo = ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "o", Name: "r", Repo: "o/r"} + testAPIRepo = ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "o", Name: "api", Repo: "o/api"} +) type fakeStore struct { mu sync.Mutex - sessions []domain.SessionRecord - projects map[string]domain.ProjectRecord - prs map[domain.SessionID][]domain.PullRequest - checks map[string][]domain.PullRequestCheck - writeErr error + sessions []domain.SessionRecord + projects map[string]domain.ProjectRecord + workspaceRepos map[string][]domain.WorkspaceRepoRecord + prs map[domain.SessionID][]domain.PullRequest + checks map[string][]domain.PullRequestCheck + writeErr error writes []fakeWrite @@ -83,6 +87,12 @@ func (s *fakeStore) UpsertProject(_ context.Context, row domain.ProjectRecord) e return nil } +func (s *fakeStore) ListWorkspaceRepos(_ context.Context, projectID string) ([]domain.WorkspaceRepoRecord, error) { + s.mu.Lock() + defer s.mu.Unlock() + return append([]domain.WorkspaceRepoRecord(nil), s.workspaceRepos[projectID]...), nil +} + func (s *fakeStore) ListPRsBySession(_ context.Context, id domain.SessionID) ([]domain.PullRequest, error) { s.mu.Lock() defer s.mu.Unlock() @@ -529,6 +539,64 @@ func TestPoll_DiscoversSiblingUnderRootSessionNamespace(t *testing.T) { } } +func TestPoll_DiscoversWorkspaceChildRepoPR(t *testing.T) { + store := testStoreWithSession() + store.sessions[0].Metadata.Branch = "ao/p-1/root" + store.projects["p"] = domain.ProjectRecord{ + ID: "p", + RepoOriginURL: "https://github.com/o/r.git", + Kind: domain.ProjectKindWorkspace, + } + store.workspaceRepos = map[string][]domain.WorkspaceRepoRecord{ + "p": {{ProjectID: "p", Name: "api", RelativePath: "api", RepoOriginURL: "https://github.com/o/api.git"}}, + } + prObs := testObs(12) + prObs.Repo = "o/api" + prObs.PR.URL = "https://github.com/o/api/pull/12" + prObs.PR.HTMLURL = "https://github.com/o/api/pull/12" + prObs.PR.Number = 12 + prObs.PR.SourceBranch = "ao/p-1/api-billing" + prObs.PR.TargetBranch = "main" + prObs.PR.HeadSHA = "api-sha" + prObs.CI.HeadSHA = "api-sha" + provider := &fakeProvider{ + repoGuards: map[string]ports.SCMGuardResult{ + prKey(testRepo, 0): {ETag: "root-v2"}, + prKey(testAPIRepo, 0): {ETag: "api-v2"}, + }, + openPRs: map[string][]ports.SCMPRObservation{ + prKey(testAPIRepo, 0): { + {URL: "https://github.com/o/api/pull/12", Number: 12, SourceBranch: "ao/p-1/api-billing", HeadRepo: "o/api", TargetBranch: "main", HeadSHA: "api-sha"}, + }, + }, + observations: map[string]ports.SCMObservation{prKey(testAPIRepo, 12): prObs}, + } + 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(provider.fetchBatches) != 1 || len(provider.fetchBatches[0]) != 1 { + t.Fatalf("child repo PR not refreshed: %#v", provider.fetchBatches) + } + ref := provider.fetchBatches[0][0] + if ref.Repo.Repo != "o/api" || ref.Number != 12 { + t.Fatalf("fetched ref = %#v, want o/api#12", ref) + } + if len(store.writes) == 0 { + t.Fatal("expected child repo PR write") + } + if got := store.writes[0].pr.Repo; got != "o/api" { + t.Fatalf("persisted repo = %q, want o/api", got) + } + if got := store.writes[0].pr.SessionID; got != "p-1" { + t.Fatalf("session id = %q, want p-1", got) + } + if len(lc.observed) != 1 { + t.Fatalf("lifecycle observations = %d, want 1", len(lc.observed)) + } +} + // A PR whose head branch matches a session branch but lives in a fork (its head // repo differs from the project repo) must not be auto-attributed: its commits // are not the session's work. It is neither fetched nor persisted.