diff --git a/backend/internal/observe/scm/observer.go b/backend/internal/observe/scm/observer.go index 1ca0fa10e..a23ff6017 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 @@ -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 { diff --git a/backend/internal/observe/scm/observer_test.go b/backend/internal/observe/scm/observer_test.go index dec26750a..e5ab060e7 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() @@ -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. diff --git a/backend/internal/session_manager/manager.go b/backend/internal/session_manager/manager.go index bdc3a9ff2..f7137c1fa 100644 --- a/backend/internal/session_manager/manager.go +++ b/backend/internal/session_manager/manager.go @@ -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 } diff --git a/backend/internal/session_manager/manager_test.go b/backend/internal/session_manager/manager_test.go index 64a3383aa..5d6ccbe4b 100644 --- a/backend/internal/session_manager/manager_test.go +++ b/backend/internal/session_manager/manager_test.go @@ -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 {