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 5bd288727f
commit e8a2438291
4 changed files with 167 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
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
@ -454,17 +455,27 @@ func (o *Observer) discoverSubjects(ctx context.Context) (map[string]*subject, [
projects[sess.ProjectID] = p
proj = p
}
repo, ok := o.provider.ParseRepository(proj.RepoOriginURL)
if !ok {
o.logger.Debug("scm observer: project has no supported SCM origin", "project", proj.ID, "origin", proj.RepoOriginURL)
repos, err := o.projectSCMRepos(ctx, proj)
if err != nil {
return nil, nil, err
}
if len(repos) == 0 {
o.logger.Debug("scm observer: project has no supported SCM origins", "project", proj.ID)
continue
}
sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, branch: branch})
for _, repo := range repos {
sessionRepos = append(sessionRepos, sessionRepo{session: sess, repo: repo, branch: branch})
}
prs, err := o.store.ListPRsBySession(ctx, sess.ID)
if err != nil {
return nil, nil, err
}
for _, pr := range openTrackedPRs(prs) {
repo, 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(repo, 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)
@ -476,6 +487,62 @@ func (o *Observer) discoverSubjects(ctx context.Context) (map[string]*subject, [
return out, sessionRepos, nil
}
func (o *Observer) projectSCMRepos(ctx context.Context, proj domain.ProjectRecord) ([]ports.SCMRepo, error) {
repos := make([]ports.SCMRepo, 0, 1)
seen := map[string]bool{}
addRepo := func(origin, label string) {
if strings.TrimSpace(origin) == "" {
return
}
repo, ok := o.provider.ParseRepository(origin)
if !ok {
o.logger.Debug("scm observer: unsupported SCM origin", "project", proj.ID, "repo", label, "origin", origin)
return
}
key := prKey(repo, 0)
if seen[key] {
return
}
seen[key] = true
repos = append(repos, repo)
}
addRepo(proj.RepoOriginURL, domain.RootWorkspaceRepoName)
if proj.Kind.WithDefault() != domain.ProjectKindWorkspace {
return repos, nil
}
childRepos, err := o.store.ListWorkspaceRepos(ctx, proj.ID)
if err != nil {
return nil, err
}
for _, child := range childRepos {
addRepo(child.RepoOriginURL, child.Name)
}
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 {
out := make([]domain.PullRequest, 0, len(prs))
for _, pr := range prs {

91
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()
@ -139,7 +149,16 @@ func (p *fakeProvider) SCMCredentialsAvailable(context.Context) (bool, error) {
}
func (p *fakeProvider) ParseRepository(remote string) (ports.SCMRepo, bool) {
return testRepo, remote != ""
if remote == "" {
return ports.SCMRepo{}, false
}
if strings.Contains(remote, "o/api") {
return ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "o", Name: "api", Repo: "o/api"}, true
}
if strings.Contains(remote, "o/web") {
return ports.SCMRepo{Provider: "github", Host: "github.com", Owner: "o", Name: "web", Repo: "o/web"}, true
}
return testRepo, true
}
func (p *fakeProvider) RepoPRListGuard(_ context.Context, repo ports.SCMRepo, _ string) (ports.SCMGuardResult, error) {
p.mu.Lock()
@ -518,6 +537,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.

13
backend/internal/session_manager/manager.go
View File

@ -512,12 +512,6 @@ func (m *Manager) Kill(ctx context.Context, id domain.SessionID) (bool, error) {
return false, fmt.Errorf("kill %s: %w", id, err)
}
// Clear the restore marker so the next boot's RestoreAll cannot resurrect a
// killed session (#2319). Best-effort: teardown below still matters.
if err := m.store.DeleteSessionWorktrees(ctx, id); err != nil {
m.logger.Warn("kill: delete restore marker failed", "sessionID", id, "error", err)
}
// Only tear down what exists. A session may have lost its handle after a
// crash or never acquired one if spawn failed partway.
if handle.ID != "" {
@ -546,6 +540,13 @@ func (m *Manager) Kill(ctx context.Context, id domain.SessionID) (bool, error) {
}
freed = true
}
// Clear the restore marker so the next boot's RestoreAll cannot resurrect a
// killed session (#2319). For workspace projects this must happen after
// teardown reads the rows; dirty-preserved rows return above and are left as
// non-restorable inventory.
if err := m.store.DeleteSessionWorktrees(ctx, id); err != nil {
m.logger.Warn("kill: delete restore marker failed", "sessionID", id, "error", err)
}
return freed, nil
}

10
backend/internal/session_manager/manager_test.go
View File

@ -696,8 +696,8 @@ func TestSpawn_WorkspaceProjectRollsBackAllWorktreesOnRuntimeFailure(t *testing.
if ws.destroyed != 0 {
t.Fatalf("single-workspace destroy calls = %d, want 0", ws.destroyed)
}
if !st.sessions["mer-1"].IsTerminated {
t.Fatal("orphaned spawn should be terminated")
if _, present := st.sessions["mer-1"]; present {
t.Fatal("seed row should be deleted after runtime creation failure")
}
}
@ -800,7 +800,7 @@ func TestKill_DeletesRestoreMarker(t *testing.T) {
m, st, _, _ := newManager()
st.sessions["mer-1"] = mkLive("mer-1")
// The session carries a leftover shutdown-saved marker from a prior cycle.
st.worktrees["mer-1"] = []domain.SessionWorktreeRecord{{SessionID: "mer-1", RepoName: "__root__"}}
st.worktrees["mer-1"] = []domain.SessionWorktreeRecord{{SessionID: "mer-1", RepoName: "__root__", State: "removed"}}
if _, err := m.Kill(ctx, "mer-1"); err != nil {
t.Fatalf("kill err = %v", err)
@ -2048,7 +2048,7 @@ func TestRestoreAll_DeletesMarkerAfterRelaunch(t *testing.T) {
Metadata: domain.SessionMetadata{WorkspacePath: "/ws/mer-1", Branch: "ao/mer-1/root", AgentSessionID: "agent-w"},
Activity: domain.Activity{State: domain.ActivityExited},
}
st.worktrees["mer-1"] = []domain.SessionWorktreeRecord{{SessionID: "mer-1", RepoName: "__root__"}}
st.worktrees["mer-1"] = []domain.SessionWorktreeRecord{{SessionID: "mer-1", RepoName: "__root__", State: "removed"}}
if err := m.RestoreAll(ctx); err != nil {
t.Fatalf("RestoreAll err = %v", err)
@ -2081,7 +2081,7 @@ func TestRestoreAll_KilledSessionNotResurrectedOnSecondBoot(t *testing.T) {
Metadata: domain.SessionMetadata{WorkspacePath: "/ws/mer-1", Branch: "ao/mer-1/root", AgentSessionID: "agent-w"},
Activity: domain.Activity{State: domain.ActivityExited},
}
st.worktrees["mer-1"] = []domain.SessionWorktreeRecord{{SessionID: "mer-1", RepoName: "__root__"}}
st.worktrees["mer-1"] = []domain.SessionWorktreeRecord{{SessionID: "mer-1", RepoName: "__root__", State: "removed"}}
// First boot: marker present, session relaunches once.
if err := m.RestoreAll(ctx); err != nil {