bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: observe workspace project SCM repos

This commit is contained in:
Aditi Chauhan 2026-07-01 12:28:19 +05:30
parent 3f10578c1b
commit 642b6aea2e
2 changed files with 144 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
backend/internal/observe/scm/observer.go
View File

@ -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 {

80
backend/internal/observe/scm/observer_test.go
View File

@ -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.