From 3413acca33501dcab34f5beea71f66d309acde45 Mon Sep 17 00:00:00 2001 From: Harshit Singh Bhandari Date: Sat, 6 Jun 2026 00:01:03 +0530 Subject: [PATCH] feat: ao session claim-pr + spawn --claim-pr wiring (#101) * feat: add session PR claiming CLI and API * fix: tighten PR claim rollback and CDC facts * fix: align PR claim branch with latest main --- .../internal/adapters/scm/github/provider.go | 11 +- backend/internal/cdc/event.go | 1 + backend/internal/cli/dto_drift_e2e_test.go | 8 + backend/internal/cli/pr_ref.go | 90 ++++ backend/internal/cli/root.go | 23 +- backend/internal/cli/session.go | 119 ++++++ backend/internal/cli/session_test.go | 90 ++++ backend/internal/cli/spawn.go | 56 ++- backend/internal/cli/spawn_test.go | 96 +++++ backend/internal/daemon/lifecycle_wiring.go | 6 +- backend/internal/daemon/scm_wiring.go | 36 +- backend/internal/domain/pr.go | 1 + backend/internal/httpd/apispec/openapi.yaml | 183 ++++++++ .../internal/httpd/apispec/specgen/build.go | 30 ++ backend/internal/httpd/controllers/dto.go | 34 ++ .../internal/httpd/controllers/sessions.go | 95 +++++ .../httpd/controllers/sessions_test.go | 91 ++++ backend/internal/ports/outbound.go | 29 ++ backend/internal/ports/pr_observations.go | 16 +- backend/internal/service/pr/manager_test.go | 6 + backend/internal/service/session/claim_pr.go | 395 ++++++++++++++++++ backend/internal/service/session/service.go | 43 +- .../internal/service/session/service_test.go | 126 +++++- backend/internal/storage/sqlite/gen/pr.sql.go | 58 +++ .../0006_pr_session_changed_cdc.sql | 345 +++++++++++++++ .../internal/storage/sqlite/queries/pr.sql | 17 + .../storage/sqlite/store/pr_cdc_test.go | 124 ++++++ .../internal/storage/sqlite/store/pr_facts.go | 1 + .../internal/storage/sqlite/store/pr_store.go | 135 +++--- .../terminal/session_integration_test.go | 5 +- frontend/src/api/schema.ts | 192 +++++++++ package-lock.json | 2 +- 32 files changed, 2377 insertions(+), 87 deletions(-) create mode 100644 backend/internal/cli/pr_ref.go create mode 100644 backend/internal/service/session/claim_pr.go create mode 100644 backend/internal/storage/sqlite/migrations/0006_pr_session_changed_cdc.sql diff --git a/backend/internal/adapters/scm/github/provider.go b/backend/internal/adapters/scm/github/provider.go index f961cdaaf..8e96849ef 100644 --- a/backend/internal/adapters/scm/github/provider.go +++ b/backend/internal/adapters/scm/github/provider.go @@ -108,7 +108,7 @@ func (p *Provider) Observe(ctx context.Context, prURL string) (ports.PRObservati // Network/auth/rate-limit failures must surface as Fetched:false. // Stable terminal states like 404 also surface that way — the PR // Manager keeps the prior row rather than fabricating closed/merged. - return out, err + return out, scmObserveError(err) } out.Draft = rest.Draft @@ -117,7 +117,7 @@ func (p *Provider) Observe(ctx context.Context, prURL string) (ports.PRObservati gq, err := p.fetchGraphQL(ctx, owner, repo, number) if err != nil { - return out, err + return out, scmObserveError(err) } out.CI = ciSummaryFromGraphQL(gq) @@ -149,6 +149,13 @@ func (p *Provider) Observe(ctx context.Context, prURL string) (ports.PRObservati return out, nil } +func scmObserveError(err error) error { + if errors.Is(err, ErrNotFound) { + return fmt.Errorf("%w: %w", ports.ErrSCMPRNotFound, err) + } + return err +} + // --------------------------------------------------------------------------- // REST: pull payload // --------------------------------------------------------------------------- diff --git a/backend/internal/cdc/event.go b/backend/internal/cdc/event.go index 32c651b44..4f43d8f28 100644 --- a/backend/internal/cdc/event.go +++ b/backend/internal/cdc/event.go @@ -25,6 +25,7 @@ const ( EventPRCreated EventType = "pr_created" EventPRUpdated EventType = "pr_updated" EventPRCheckRecorded EventType = "pr_check_recorded" + EventPRSessionChanged EventType = "pr_session_changed" EventPRReviewThreadAdded EventType = "pr_review_thread_added" EventPRReviewThreadResolved EventType = "pr_review_thread_resolved" ) diff --git a/backend/internal/cli/dto_drift_e2e_test.go b/backend/internal/cli/dto_drift_e2e_test.go index 47a58f8ee..83739b571 100644 --- a/backend/internal/cli/dto_drift_e2e_test.go +++ b/backend/internal/cli/dto_drift_e2e_test.go @@ -90,6 +90,14 @@ func (f *fakeSessionService) Send(context.Context, domain.SessionID, string) err return nil } +func (f *fakeSessionService) ListPRs(context.Context, domain.SessionID) ([]domain.PRFacts, error) { + return nil, nil +} + +func (f *fakeSessionService) ClaimPR(context.Context, domain.SessionID, string, sessionsvc.ClaimPROptions) (sessionsvc.ClaimPRResult, error) { + return sessionsvc.ClaimPRResult{}, nil +} + // fakeProjectManager captures the project.AddInput the controller decodes from // the CLI's request body. Every other method is a no-op so it satisfies the // projectsvc.Manager interface. diff --git a/backend/internal/cli/pr_ref.go b/backend/internal/cli/pr_ref.go new file mode 100644 index 000000000..41603494a --- /dev/null +++ b/backend/internal/cli/pr_ref.go @@ -0,0 +1,90 @@ +package cli + +import ( + "context" + "errors" + "fmt" + "net/url" + "strconv" + "strings" +) + +func (c *commandContext) resolvePRRef(ctx context.Context, ref string, project projectDetails) (string, error) { + ref = strings.TrimSpace(ref) + if ref == "" { + return "", usageError{errors.New("PR reference must be a github.com PR URL or a number")} + } + if isNumericPRRef(ref) { + repo := strings.TrimSpace(project.Repo) + if repo == "" { + // The daemon must not shell out to external CLIs from its loopback API; + // when the durable project record lacks repo_origin_url, the thin CLI + // does the one-off gh lookup from the registered project checkout and + // sends the daemon a normalized URL. + out, err := c.deps.CommandOutputInDir(ctx, project.Path, "gh", "repo", "view", "--json", "url", "-q", ".url") + if err != nil || strings.TrimSpace(string(out)) == "" { + return "", usageError{errors.New("gh not available; pass the full PR URL")} + } + repo = strings.TrimSpace(string(out)) + } + owner, name, err := cliGitHubRepoFromURL(repo) + if err != nil { + return "", usageError{errors.New("PR reference must be a github.com PR URL or a number")} + } + n, _ := strconv.Atoi(strings.TrimPrefix(ref, "#")) + return fmt.Sprintf("https://github.com/%s/%s/pull/%d", owner, name, n), nil + } + owner, name, n, err := cliParseGitHubPRURL(ref) + if err != nil || owner == "" || name == "" || n <= 0 { + return "", usageError{errors.New("PR reference must be a github.com PR URL or a number")} + } + return fmt.Sprintf("https://github.com/%s/%s/pull/%d", owner, name, n), nil +} + +func isNumericPRRef(ref string) bool { + ref = strings.TrimPrefix(strings.TrimSpace(ref), "#") + n, err := strconv.Atoi(ref) + return err == nil && n > 0 +} + +func cliParseGitHubPRURL(raw string) (string, string, int, error) { + u, err := url.Parse(raw) + if err != nil { + return "", "", 0, err + } + if !strings.EqualFold(u.Scheme, "https") || !strings.EqualFold(u.Hostname(), "github.com") { + return "", "", 0, errors.New("not github") + } + parts := strings.Split(strings.Trim(u.Path, "/"), "/") + if len(parts) != 4 || parts[2] != "pull" { + return "", "", 0, errors.New("not pr") + } + n, err := strconv.Atoi(parts[3]) + if err != nil || n <= 0 { + return "", "", 0, errors.New("bad number") + } + return parts[0], strings.TrimSuffix(parts[1], ".git"), n, nil +} + +func cliGitHubRepoFromURL(raw string) (string, string, error) { + raw = strings.TrimSpace(raw) + if strings.HasPrefix(raw, "git@github.com:") { + parts := strings.Split(strings.TrimSuffix(strings.TrimPrefix(raw, "git@github.com:"), ".git"), "/") + if len(parts) == 2 && parts[0] != "" && parts[1] != "" { + return parts[0], parts[1], nil + } + return "", "", errors.New("bad repo") + } + u, err := url.Parse(raw) + if err != nil { + return "", "", err + } + if !strings.EqualFold(u.Hostname(), "github.com") { + return "", "", errors.New("not github") + } + parts := strings.Split(strings.Trim(u.Path, "/"), "/") + if len(parts) < 2 || parts[0] == "" || parts[1] == "" { + return "", "", errors.New("bad repo") + } + return parts[0], strings.TrimSuffix(parts[1], ".git"), nil +} diff --git a/backend/internal/cli/root.go b/backend/internal/cli/root.go index 7d5c0df2e..daabd5b81 100644 --- a/backend/internal/cli/root.go +++ b/backend/internal/cli/root.go @@ -50,12 +50,13 @@ type Deps struct { Out io.Writer Err io.Writer - HTTPClient *http.Client - Executable func() (string, error) - StartProcess func(processStartConfig) error - ProcessAlive func(pid int) bool - LookPath func(file string) (string, error) - CommandOutput func(ctx context.Context, name string, args ...string) ([]byte, error) + HTTPClient *http.Client + Executable func() (string, error) + StartProcess func(processStartConfig) error + ProcessAlive func(pid int) bool + LookPath func(file string) (string, error) + CommandOutput func(ctx context.Context, name string, args ...string) ([]byte, error) + CommandOutputInDir func(ctx context.Context, dir, name string, args ...string) ([]byte, error) // DoctorGitHubRESTBase lets tests point the doctor GitHub token probe at // httptest without mutating package-global state. DoctorGitHubRESTBase string @@ -75,6 +76,7 @@ func DefaultDeps() Deps { ProcessAlive: processalive.Alive, LookPath: exec.LookPath, CommandOutput: commandOutput, + CommandOutputInDir: commandOutputInDir, DoctorGitHubRESTBase: defaultDoctorGitHubRESTBase, Now: time.Now, Sleep: time.Sleep, @@ -85,6 +87,12 @@ func commandOutput(ctx context.Context, name string, args ...string) ([]byte, er return exec.CommandContext(ctx, name, args...).CombinedOutput() } +func commandOutputInDir(ctx context.Context, dir, name string, args ...string) ([]byte, error) { + cmd := exec.CommandContext(ctx, name, args...) + cmd.Dir = dir + return cmd.CombinedOutput() +} + func (d Deps) withDefaults() Deps { def := DefaultDeps() if d.In == nil { @@ -114,6 +122,9 @@ func (d Deps) withDefaults() Deps { if d.CommandOutput == nil { d.CommandOutput = def.CommandOutput } + if d.CommandOutputInDir == nil { + d.CommandOutputInDir = def.CommandOutputInDir + } if d.DoctorGitHubRESTBase == "" { d.DoctorGitHubRESTBase = def.DoctorGitHubRESTBase } diff --git a/backend/internal/cli/session.go b/backend/internal/cli/session.go index 9f6696263..2e0d84cc3 100644 --- a/backend/internal/cli/session.go +++ b/backend/internal/cli/session.go @@ -29,6 +29,12 @@ type sessionCleanupOptions struct { yes bool } +type sessionClaimPROptions struct { + project string + json bool + noTakeover bool +} + type sessionRenameRequest struct { DisplayName string `json:"displayName"` } @@ -78,6 +84,30 @@ type cleanupSessionsResponse struct { Cleaned []string `json:"cleaned"` } +type claimPRRequest struct { + PR string `json:"pr"` + AllowTakeover bool `json:"allowTakeover"` +} + +type sessionPRDTO struct { + URL string `json:"url"` + Number int `json:"number"` + State string `json:"state"` + CI string `json:"ci"` + Review string `json:"review"` + Mergeability string `json:"mergeability"` + ReviewComments bool `json:"reviewComments"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type claimPRResponse struct { + OK bool `json:"ok"` + SessionID string `json:"sessionId"` + PRs []sessionPRDTO `json:"prs"` + BranchChanged bool `json:"branchChanged"` + TakenOverFrom []string `json:"takenOverFrom"` +} + type sessionListEntry struct { ID string `json:"id"` ProjectID string `json:"projectId"` @@ -109,6 +139,7 @@ func newSessionCommand(ctx *commandContext) *cobra.Command { cmd.AddCommand(newSessionRestoreCommand(ctx)) cmd.AddCommand(newSessionRenameCommand(ctx)) cmd.AddCommand(newSessionCleanupCommand(ctx)) + cmd.AddCommand(newSessionClaimPRCommand(ctx)) return cmd } @@ -220,6 +251,34 @@ func newSessionCleanupCommand(ctx *commandContext) *cobra.Command { return cmd } +func newSessionClaimPRCommand(ctx *commandContext) *cobra.Command { + var opts sessionClaimPROptions + cmd := &cobra.Command{ + Use: "claim-pr ", + Short: "Attach an existing PR to a session", + Args: func(cmd *cobra.Command, args []string) error { + if err := cobra.ExactArgs(2)(cmd, args); err != nil { + return usageError{err} + } + if _, err := normalizeSessionID(args[0]); err != nil { + return err + } + return nil + }, + RunE: func(cmd *cobra.Command, args []string) error { + id, err := normalizeSessionID(args[0]) + if err != nil { + return err + } + return ctx.claimSessionPR(cmd.Context(), cmd, id, args[1], opts) + }, + } + addSessionProjectFlag(cmd.Flags(), &opts.project, "Project id to scope the lookup") + cmd.Flags().BoolVar(&opts.noTakeover, "no-takeover", false, "Refuse if another active session owns the PR") + cmd.Flags().BoolVar(&opts.json, "json", false, "Output as JSON") + return cmd +} + func addSessionProjectFlag(flags interface { StringVarP(*string, string, string, string, string) }, target *string, usage string) { @@ -249,6 +308,66 @@ func sessionRenameArgs(cmd *cobra.Command, args []string) error { return nil } +func (c *commandContext) claimSessionPR(ctx context.Context, cmd *cobra.Command, id, ref string, opts sessionClaimPROptions) error { + sess, err := c.fetchScopedSession(ctx, id, opts.project) + if err != nil { + return err + } + project, err := c.fetchProjectDetails(ctx, sess.ProjectID) + if err != nil { + return err + } + resolvedRef, err := c.resolvePRRef(ctx, ref, project) + if err != nil { + return err + } + var res claimPRResponse + req := claimPRRequest{PR: resolvedRef, AllowTakeover: !opts.noTakeover} + if err := c.postJSON(ctx, "sessions/"+url.PathEscape(id)+"/pr/claim", req, &res); err != nil { + return err + } + if opts.json { + return writeJSON(cmd.OutOrStdout(), res) + } + return writeClaimPRResult(cmd, res) +} + +func (c *commandContext) fetchProjectDetails(ctx context.Context, id string) (projectDetails, error) { + var res projectGetResult + if err := c.getJSON(ctx, "projects/"+url.PathEscape(id), &res); err != nil { + return projectDetails{}, err + } + return res.Project, nil +} + +func writeClaimPRResult(cmd *cobra.Command, res claimPRResponse) error { + out := cmd.OutOrStdout() + if len(res.PRs) == 0 { + _, err := fmt.Fprintf(out, "session %s claimed PR\n", res.SessionID) + return err + } + pr := res.PRs[0] + if _, err := fmt.Fprintf(out, "session %s claimed PR #%d\n", res.SessionID, pr.Number); err != nil { + return err + } + if _, err := fmt.Fprintf(out, " pr: %s\n", pr.URL); err != nil { + return err + } + checkout := "already on PR branch" + if res.BranchChanged { + checkout = "switched to PR branch" + } + if _, err := fmt.Fprintf(out, " checkout: %s\n", checkout); err != nil { + return err + } + for _, owner := range res.TakenOverFrom { + if _, err := fmt.Fprintf(out, " taking over from %s\n", owner); err != nil { + return err + } + } + return nil +} + func (c *commandContext) listSessions(ctx context.Context, cmd *cobra.Command, opts sessionListOptions) error { params := url.Values{} if opts.project != "" { diff --git a/backend/internal/cli/session_test.go b/backend/internal/cli/session_test.go index 45e2b283f..f33723901 100644 --- a/backend/internal/cli/session_test.go +++ b/backend/internal/cli/session_test.go @@ -1,6 +1,7 @@ package cli import ( + "context" "encoding/json" "io" "net/http" @@ -53,6 +54,20 @@ func sessionCommandServer(t *testing.T) (*httptest.Server, *sessionRequestLog) { } case r.Method == http.MethodGet && r.URL.Path == "/api/v1/sessions/demo-1": _, _ = io.WriteString(w, `{"session":`+sessionJSON("demo-1", "demo", "worker", "working", false)+`}`) + case r.Method == http.MethodGet && r.URL.Path == "/api/v1/projects/demo": + _, _ = io.WriteString(w, `{"status":"ok","project":{"id":"demo","name":"Demo","path":"/repo/demo","repo":"https://github.com/aoagents/agent-orchestrator","defaultBranch":"main"}}`) + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/demo-1/pr/claim": + var req claimPRRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if !req.AllowTakeover { + w.WriteHeader(http.StatusConflict) + _, _ = io.WriteString(w, `{"error":"conflict","code":"PR_CLAIMED_BY_ACTIVE_SESSION","message":"PR is already claimed by active session demo-2 (omit --no-takeover to steal)"}`) + return + } + _, _ = io.WriteString(w, `{"ok":true,"sessionId":"demo-1","prs":[{"url":`+jsonQuote(req.PR)+`,"number":142,"state":"open","ci":"passing","review":"review_required","mergeability":"mergeable","reviewComments":false,"updatedAt":"2026-06-04T12:00:00Z"}],"branchChanged":true,"takenOverFrom":["demo-0"]}`) case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/cleanup": _, _ = io.WriteString(w, `{"ok":true,"cleaned":["demo-old","demo-orch"]}`) case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/demo-1/kill": @@ -371,3 +386,78 @@ func TestSessionRename_ProjectMismatchDoesNotPatch(t *testing.T) { t.Fatalf("requests = %#v, want %#v", got, want) } } + +func TestSessionClaimPR_ProjectScopeMismatchIsUsage(t *testing.T) { + cfg := setConfigEnv(t) + srv, log := sessionCommandServer(t) + writeRunFileFor(t, cfg, srv) + + _, _, err := executeCLI(t, Deps{ProcessAlive: func(int) bool { return true }}, "session", "claim-pr", "demo-1", "https://github.com/aoagents/agent-orchestrator/pull/142", "-p", "other") + if err == nil || ExitCode(err) != 2 || !strings.Contains(err.Error(), "session demo-1 is not in project other") { + t.Fatalf("err=%v exit=%d, want project mismatch usage", err, ExitCode(err)) + } + want := []string{"GET /api/v1/sessions/demo-1"} + if got := log.all(); !reflect.DeepEqual(got, want) { + t.Fatalf("requests=%#v want %#v", got, want) + } +} + +func TestSessionClaimPR_JSONAndNoTakeoverError(t *testing.T) { + cfg := setConfigEnv(t) + srv, _ := sessionCommandServer(t) + writeRunFileFor(t, cfg, srv) + + out, errOut, err := executeCLI(t, Deps{ProcessAlive: func(int) bool { return true }}, "session", "claim-pr", "demo-1", "https://github.com/aoagents/agent-orchestrator/pull/142", "--json") + if err != nil { + t.Fatalf("claim-pr --json failed: %v stderr=%s", err, errOut) + } + var got claimPRResponse + if err := json.Unmarshal([]byte(out), &got); err != nil || got.SessionID != "demo-1" || len(got.PRs) != 1 || got.PRs[0].Number != 142 { + t.Fatalf("bad json err=%v got=%#v out=%s", err, got, out) + } + + _, _, err = executeCLI(t, Deps{ProcessAlive: func(int) bool { return true }}, "session", "claim-pr", "demo-1", "https://github.com/aoagents/agent-orchestrator/pull/142", "--no-takeover") + if err == nil || ExitCode(err) != 1 || !strings.Contains(err.Error(), "PR_CLAIMED_BY_ACTIVE_SESSION") { + t.Fatalf("err=%v exit=%d, want takeover refusal runtime error", err, ExitCode(err)) + } +} + +func TestSessionClaimPR_GHFallbackWhenProjectRepoMissing(t *testing.T) { + cfg := setConfigEnv(t) + log := &sessionRequestLog{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + log.append(r) + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodGet && r.URL.Path == "/api/v1/sessions/demo-1": + _, _ = io.WriteString(w, `{"session":`+sessionJSON("demo-1", "demo", "worker", "working", false)+`}`) + case r.Method == http.MethodGet && r.URL.Path == "/api/v1/projects/demo": + _, _ = io.WriteString(w, `{"status":"ok","project":{"id":"demo","name":"Demo","path":"/repo/demo","repo":"","defaultBranch":"main"}}`) + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/demo-1/pr/claim": + var req claimPRRequest + _ = json.NewDecoder(r.Body).Decode(&req) + _, _ = io.WriteString(w, `{"ok":true,"sessionId":"demo-1","prs":[{"url":`+jsonQuote(req.PR)+`,"number":142,"state":"open","ci":"passing","review":"review_required","mergeability":"mergeable","reviewComments":false,"updatedAt":"2026-06-04T12:00:00Z"}],"branchChanged":false,"takenOverFrom":[]}`) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(srv.Close) + writeRunFileFor(t, cfg, srv) + var ghDir string + out, _, err := executeCLI(t, Deps{ + ProcessAlive: func(int) bool { return true }, + CommandOutputInDir: func(_ context.Context, dir, name string, args ...string) ([]byte, error) { + ghDir = dir + if name != "gh" { + t.Fatalf("command name=%s", name) + } + return []byte("https://github.com/aoagents/agent-orchestrator\n"), nil + }, + }, "session", "claim-pr", "demo-1", "142") + if err != nil { + t.Fatalf("claim-pr fallback failed: %v", err) + } + if ghDir != "/repo/demo" || !strings.Contains(out, "claimed PR #142") { + t.Fatalf("ghDir=%q out=%s", ghDir, out) + } +} diff --git a/backend/internal/cli/spawn.go b/backend/internal/cli/spawn.go index 534fff78f..30cc7fad4 100644 --- a/backend/internal/cli/spawn.go +++ b/backend/internal/cli/spawn.go @@ -1,7 +1,9 @@ package cli import ( + "context" "fmt" + "net/url" "github.com/spf13/cobra" @@ -9,12 +11,14 @@ import ( ) type spawnOptions struct { - project string - harness string - branch string - prompt string - issue string - rules string + project string + harness string + branch string + prompt string + issue string + rules string + claimPR string + noTakeover bool } // spawnRequest mirrors the daemon's SpawnSessionRequest body for @@ -48,6 +52,20 @@ func newSpawnCommand(ctx *commandContext) *cobra.Command { if opts.project == "" { return usageError{fmt.Errorf("--project is required")} } + if opts.noTakeover && opts.claimPR == "" { + return usageError{fmt.Errorf("--no-takeover requires --claim-pr")} + } + claimRef := "" + if opts.claimPR != "" { + project, err := ctx.fetchProjectDetails(cmd.Context(), opts.project) + if err != nil { + return err + } + claimRef, err = ctx.resolvePRRef(cmd.Context(), opts.claimPR, project) + if err != nil { + return err + } + } req := spawnRequest{ ProjectID: opts.project, IssueID: opts.issue, @@ -60,8 +78,25 @@ func newSpawnCommand(ctx *commandContext) *cobra.Command { if err := ctx.postJSON(cmd.Context(), "sessions", req, &res); err != nil { return err } + claimed := "" + if opts.claimPR != "" { + var claim claimPRResponse + if err := ctx.postJSON(cmd.Context(), "sessions/"+url.PathEscape(res.Session.ID)+"/pr/claim", claimPRRequest{PR: claimRef, AllowTakeover: !opts.noTakeover}, &claim); err != nil { + if killErr := ctx.rollbackSpawnedSession(cmd.Context(), res.Session.ID); killErr != nil { + return fmt.Errorf("failed to claim PR %s: %w; rollback of session %s failed: %w", opts.claimPR, err, res.Session.ID, killErr) + } + return fmt.Errorf("failed to claim PR %s: %w; rolled back session %s", opts.claimPR, err, res.Session.ID) + } + if len(claim.PRs) > 0 { + claimed = claim.PRs[0].URL + } + } out := cmd.OutOrStdout() - if _, err := fmt.Fprintf(out, "spawned session %s (%s)\n", res.Session.ID, res.Session.Status); err != nil { + claimLabel := "" + if claimed != "" { + claimLabel = fmt.Sprintf(" (claimed %s)", claimed) + } + if _, err := fmt.Fprintf(out, "spawned session %s (%s)%s\n", res.Session.ID, res.Session.Status, claimLabel); err != nil { return err } // The daemon runs zellij under a short, non-default socket dir (see @@ -84,5 +119,12 @@ func newSpawnCommand(ctx *commandContext) *cobra.Command { f.StringVar(&opts.prompt, "prompt", "", "Initial prompt for the agent") f.StringVar(&opts.issue, "issue", "", "Issue id to associate with the session") f.StringVar(&opts.rules, "rules", "", "Agent rules appended to the prompt") + f.StringVar(&opts.claimPR, "claim-pr", "", "Immediately claim an existing PR for the spawned session") + f.BoolVar(&opts.noTakeover, "no-takeover", false, "Refuse if another active session owns the claimed PR (requires --claim-pr)") return cmd } + +func (c *commandContext) rollbackSpawnedSession(ctx context.Context, id string) error { + var res killSessionResponse + return c.postJSON(ctx, "sessions/"+url.PathEscape(id)+"/kill", struct{}{}, &res) +} diff --git a/backend/internal/cli/spawn_test.go b/backend/internal/cli/spawn_test.go index baca6e3ad..1076a2fe0 100644 --- a/backend/internal/cli/spawn_test.go +++ b/backend/internal/cli/spawn_test.go @@ -2,6 +2,11 @@ package cli import ( "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "reflect" "strings" "testing" ) @@ -35,3 +40,94 @@ func TestProjectAddCommand_RequiresPath(t *testing.T) { t.Fatalf("error = %v, want it to mention --path is required", err) } } + +func TestSpawnClaimPRWiring(t *testing.T) { + cfg := setConfigEnv(t) + var requests []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests = append(requests, r.Method+" "+r.URL.Path) + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodGet && r.URL.Path == "/api/v1/projects/demo": + _, _ = io.WriteString(w, `{"status":"ok","project":{"id":"demo","name":"Demo","path":"/repo/demo","repo":"https://github.com/aoagents/agent-orchestrator","defaultBranch":"main"}}`) + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions": + _, _ = io.WriteString(w, `{"session":{"id":"demo-9","status":"idle"}}`) + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/demo-9/pr/claim": + var req claimPRRequest + _ = json.NewDecoder(r.Body).Decode(&req) + if req.PR != "https://github.com/aoagents/agent-orchestrator/pull/142" || req.AllowTakeover { + t.Fatalf("claim request = %#v", req) + } + _, _ = io.WriteString(w, `{"ok":true,"sessionId":"demo-9","prs":[{"url":"https://github.com/aoagents/agent-orchestrator/pull/142","number":142,"state":"open","ci":"passing","review":"review_required","mergeability":"mergeable","reviewComments":false,"updatedAt":"2026-06-04T12:00:00Z"}],"branchChanged":false,"takenOverFrom":[]}`) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(srv.Close) + writeRunFileFor(t, cfg, srv) + + out, errOut, err := executeCLI(t, Deps{ProcessAlive: func(int) bool { return true }}, "spawn", "--project", "demo", "--claim-pr", "142", "--no-takeover") + if err != nil { + t.Fatalf("spawn claim-pr failed: %v stderr=%s", err, errOut) + } + if !strings.Contains(out, "claimed https://github.com/aoagents/agent-orchestrator/pull/142") { + t.Fatalf("output missing claimed label: %s", out) + } + want := []string{"GET /api/v1/projects/demo", "POST /api/v1/sessions", "POST /api/v1/sessions/demo-9/pr/claim"} + if !reflect.DeepEqual(requests, want) { + t.Fatalf("requests=%#v want %#v", requests, want) + } +} + +func TestSpawnClaimPRFailureRollsBackSession(t *testing.T) { + cfg := setConfigEnv(t) + var requests []string + sessions := map[string]bool{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests = append(requests, r.Method+" "+r.URL.Path) + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodGet && r.URL.Path == "/api/v1/projects/demo": + _, _ = io.WriteString(w, `{"status":"ok","project":{"id":"demo","name":"Demo","path":"/repo/demo","repo":"https://github.com/aoagents/agent-orchestrator","defaultBranch":"main"}}`) + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions": + sessions["demo-10"] = true + _, _ = io.WriteString(w, `{"session":{"id":"demo-10","status":"idle"}}`) + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/demo-10/pr/claim": + if !sessions["demo-10"] { + t.Fatal("claim called before session existed") + } + w.WriteHeader(http.StatusNotFound) + _, _ = io.WriteString(w, `{"error":"not_found","code":"PR_NOT_FOUND","message":"PR not found"}`) + case r.Method == http.MethodPost && r.URL.Path == "/api/v1/sessions/demo-10/kill": + delete(sessions, "demo-10") + _, _ = io.WriteString(w, `{"ok":true,"sessionId":"demo-10","freed":true}`) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(srv.Close) + writeRunFileFor(t, cfg, srv) + + _, _, err := executeCLI(t, Deps{ProcessAlive: func(int) bool { return true }}, "spawn", "--project", "demo", "--claim-pr", "142") + if err == nil { + t.Fatal("expected spawn claim failure") + } + msg := err.Error() + if !strings.Contains(msg, "failed to claim PR 142") || !strings.Contains(msg, "rolled back session demo-10") { + t.Fatalf("error = %v", err) + } + if sessions["demo-10"] { + t.Fatalf("spawned session still present after claim rollback: %#v", sessions) + } + want := []string{"GET /api/v1/projects/demo", "POST /api/v1/sessions", "POST /api/v1/sessions/demo-10/pr/claim", "POST /api/v1/sessions/demo-10/kill"} + if !reflect.DeepEqual(requests, want) { + t.Fatalf("requests=%#v want %#v", requests, want) + } +} + +func TestSpawnNoTakeoverRequiresClaimPR(t *testing.T) { + _, _, err := executeCLI(t, Deps{}, "spawn", "--project", "demo", "--no-takeover") + if err == nil || ExitCode(err) != 2 || !strings.Contains(err.Error(), "--no-takeover requires --claim-pr") { + t.Fatalf("err=%v exit=%d", err, ExitCode(err)) + } +} diff --git a/backend/internal/daemon/lifecycle_wiring.go b/backend/internal/daemon/lifecycle_wiring.go index a313a6db6..d978fc689 100644 --- a/backend/internal/daemon/lifecycle_wiring.go +++ b/backend/internal/daemon/lifecycle_wiring.go @@ -81,7 +81,11 @@ func startSession(cfg config.Config, runtime ports.Runtime, store *sqlite.Store, Lifecycle: lcm, DataDir: cfg.DataDir, }) - return sessionsvc.New(mgr, store), nil + scmProvider, err := newGitHubSCMProvider(log) + if err != nil { + logSCMProviderDisabled(log, err) + } + return sessionsvc.NewWithDeps(sessionsvc.Deps{Manager: mgr, Store: store, PRClaimer: store, SCM: scmProvider}), nil } // runtimeMessageSender is the narrow part of the concrete runtime needed by diff --git a/backend/internal/daemon/scm_wiring.go b/backend/internal/daemon/scm_wiring.go index 772bbd528..58f5cb769 100644 --- a/backend/internal/daemon/scm_wiring.go +++ b/backend/internal/daemon/scm_wiring.go @@ -20,27 +20,35 @@ import ( // observer performs a lazy credential check in its background goroutine, logs // one warning, and disables itself before any provider API calls. func startSCMObserver(ctx context.Context, store *sqlite.Store, lcm *lifecycle.Manager, logger *slog.Logger) <-chan struct{} { - tokens := scmgithub.FallbackTokenSource{ - scmgithub.EnvTokenSource{EnvVars: []string{"AO_GITHUB_TOKEN"}}, - &scmgithub.GHTokenSource{}, - } - // Avoid token preflight on daemon startup. GHTokenSource may shell out to `gh`, - // which is too slow/flaky for the startup readiness path (especially on - // Windows CI). The provider will resolve credentials lazily in its background - // observer goroutine before it makes any GitHub API call. - provider, err := scmgithub.NewProvider(scmgithub.ProviderOptions{Token: tokens, SkipTokenPreflight: true}) + provider, err := newGitHubSCMProvider(logger) if err != nil { - if errors.Is(err, scmgithub.ErrNoToken) || errors.Is(err, scmgithub.ErrAuthFailed) { - logger.Warn("scm observer disabled: no usable GitHub token", "err", err) - } else { - logger.Warn("scm observer disabled: GitHub provider setup failed", "err", err) - } + logSCMProviderDisabled(logger, err) return closedDone() } observer := scmobserve.New(provider, store, lcm, scmobserve.Config{Logger: logger}) return observer.Start(ctx) } +func newGitHubSCMProvider(logger *slog.Logger) (*scmgithub.Provider, error) { + tokens := scmgithub.FallbackTokenSource{ + scmgithub.EnvTokenSource{EnvVars: []string{"AO_GITHUB_TOKEN"}}, + &scmgithub.GHTokenSource{}, + } + // Avoid token preflight on daemon startup and session service construction. + // GHTokenSource may shell out to `gh`, which is too slow/flaky for the startup + // readiness path. Provider calls resolve credentials lazily when claim-pr or + // the background observer actually needs GitHub. + return scmgithub.NewProvider(scmgithub.ProviderOptions{Token: tokens, SkipTokenPreflight: true, Logger: logger}) +} + +func logSCMProviderDisabled(logger *slog.Logger, err error) { + if errors.Is(err, scmgithub.ErrNoToken) || errors.Is(err, scmgithub.ErrAuthFailed) { + logger.Warn("scm observer disabled: no usable GitHub token", "err", err) + } else { + logger.Warn("scm observer disabled: GitHub provider setup failed", "err", err) + } +} + func closedDone() <-chan struct{} { done := make(chan struct{}) close(done) diff --git a/backend/internal/domain/pr.go b/backend/internal/domain/pr.go index 6195dc3c2..704c0a7cb 100644 --- a/backend/internal/domain/pr.go +++ b/backend/internal/domain/pr.go @@ -16,6 +16,7 @@ type PRFacts struct { Review ReviewDecision Mergeability Mergeability ReviewComments bool // has unresolved review comments (any author) to address + UpdatedAt time.Time } // PullRequest is the app-level representation of one tracked pull request as diff --git a/backend/internal/httpd/apispec/openapi.yaml b/backend/internal/httpd/apispec/openapi.yaml index 7e2ed972e..2233f1cf3 100644 --- a/backend/internal/httpd/apispec/openapi.yaml +++ b/backend/internal/httpd/apispec/openapi.yaml @@ -540,6 +540,108 @@ paths: summary: Mark a session terminated and tear down runtime/workspace resources tags: - sessions + /api/v1/sessions/{sessionId}/pr: + get: + operationId: listSessionPRs + parameters: + - description: Session identifier, e.g. project-1. + in: path + name: sessionId + required: true + schema: + description: Session identifier, e.g. project-1. + type: string + responses: + "200": + content: + application/json: + schema: + $ref: '#/components/schemas/ListSessionPRsResponse' + description: OK + "404": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Not Found + "500": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Internal Server Error + "501": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Not Implemented + summary: List pull requests owned by a session + tags: + - sessions + /api/v1/sessions/{sessionId}/pr/claim: + post: + operationId: claimSessionPR + parameters: + - description: Session identifier, e.g. project-1. + in: path + name: sessionId + required: true + schema: + description: Session identifier, e.g. project-1. + type: string + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/ClaimPRRequest' + required: true + responses: + "200": + content: + application/json: + schema: + $ref: '#/components/schemas/ClaimPRResponse' + description: OK + "400": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Bad Request + "404": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Not Found + "409": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Conflict + "422": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Unprocessable Entity + "501": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Not Implemented + "503": + content: + application/json: + schema: + $ref: '#/components/schemas/APIError' + description: Service Unavailable + summary: Claim an existing pull request for a session + tags: + - sessions /api/v1/sessions/{sessionId}/restore: post: operationId: restoreSession @@ -693,6 +795,41 @@ components: required: - path type: object + ClaimPRRequest: + properties: + allowTakeover: + type: + - "null" + - boolean + pr: + minLength: 1 + type: string + required: + - pr + type: object + ClaimPRResponse: + properties: + branchChanged: + type: boolean + ok: + type: boolean + prs: + items: + $ref: '#/components/schemas/SessionPRFacts' + type: array + sessionId: + type: string + takenOverFrom: + items: + type: string + type: array + required: + - ok + - sessionId + - prs + - branchChanged + - takenOverFrom + type: object CleanupSessionsResponse: properties: cleaned: @@ -753,6 +890,18 @@ components: required: - projects type: object + ListSessionPRsResponse: + properties: + prs: + items: + $ref: '#/components/schemas/SessionPRFacts' + type: array + sessionId: + type: string + required: + - sessionId + - prs + type: object ListSessionsResponse: properties: sessions: @@ -994,6 +1143,40 @@ components: - updatedAt - status type: object + SessionPRFacts: + properties: + ci: + type: string + mergeability: + type: string + number: + type: integer + review: + type: string + reviewComments: + type: boolean + state: + enum: + - draft + - open + - merged + - closed + type: string + updatedAt: + format: date-time + type: string + url: + type: string + required: + - url + - number + - state + - ci + - review + - mergeability + - reviewComments + - updatedAt + type: object SessionResponse: properties: session: diff --git a/backend/internal/httpd/apispec/specgen/build.go b/backend/internal/httpd/apispec/specgen/build.go index e178a1532..b83cc571f 100644 --- a/backend/internal/httpd/apispec/specgen/build.go +++ b/backend/internal/httpd/apispec/specgen/build.go @@ -133,6 +133,10 @@ var schemaNames = map[string]string{ "ControllersKillSessionResponse": "KillSessionResponse", "ControllersSendSessionMessageRequest": "SendSessionMessageRequest", "ControllersSendSessionMessageResponse": "SendSessionMessageResponse", + "ControllersClaimPRResponse": "ClaimPRResponse", + "ControllersClaimPRRequest": "ClaimPRRequest", + "ControllersSessionPRFacts": "SessionPRFacts", + "ControllersListSessionPRsResponse": "ListSessionPRsResponse", "ControllersSpawnOrchestratorRequest": "SpawnOrchestratorRequest", "ControllersSpawnOrchestratorResponse": "SpawnOrchestratorResponse", "ControllersOrchestratorResponse": "OrchestratorResponse", @@ -308,6 +312,32 @@ func sessionOperations() []operation { {http.StatusInternalServerError, envelope.APIError{}}, }, }, + { + method: http.MethodGet, path: "/api/v1/sessions/{sessionId}/pr", id: "listSessionPRs", tag: "sessions", + summary: "List pull requests owned by a session", + pathParams: []any{controllers.SessionIDParam{}}, + resps: []respUnit{ + {http.StatusOK, controllers.ListSessionPRsResponse{}}, + {http.StatusNotFound, envelope.APIError{}}, + {http.StatusInternalServerError, envelope.APIError{}}, + {http.StatusNotImplemented, envelope.APIError{}}, + }, + }, + { + method: http.MethodPost, path: "/api/v1/sessions/{sessionId}/pr/claim", id: "claimSessionPR", tag: "sessions", + summary: "Claim an existing pull request for a session", + pathParams: []any{controllers.SessionIDParam{}}, + reqBody: controllers.ClaimPRRequest{}, + resps: []respUnit{ + {http.StatusOK, controllers.ClaimPRResponse{}}, + {http.StatusBadRequest, envelope.APIError{}}, + {http.StatusNotFound, envelope.APIError{}}, + {http.StatusConflict, envelope.APIError{}}, + {http.StatusUnprocessableEntity, envelope.APIError{}}, + {http.StatusServiceUnavailable, envelope.APIError{}}, + {http.StatusNotImplemented, envelope.APIError{}}, + }, + }, { method: http.MethodPatch, path: "/api/v1/sessions/{sessionId}", id: "renameSession", tag: "sessions", summary: "Rename a session display name", diff --git a/backend/internal/httpd/controllers/dto.go b/backend/internal/httpd/controllers/dto.go index 9ccef68b2..d35aaa943 100644 --- a/backend/internal/httpd/controllers/dto.go +++ b/backend/internal/httpd/controllers/dto.go @@ -3,6 +3,7 @@ package controllers import ( "encoding/json" "errors" + "time" "github.com/aoagents/agent-orchestrator/backend/internal/domain" projectsvc "github.com/aoagents/agent-orchestrator/backend/internal/service/project" @@ -175,6 +176,39 @@ type SendSessionMessageResponse struct { Message string `json:"message"` } +// SessionPRFacts is the pull-request read shape returned under session PR routes. +type SessionPRFacts struct { + URL string `json:"url"` + Number int `json:"number"` + State string `json:"state" enum:"draft,open,merged,closed"` + CI domain.CIState `json:"ci"` + Review domain.ReviewDecision `json:"review"` + Mergeability domain.Mergeability `json:"mergeability"` + ReviewComments bool `json:"reviewComments"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// ListSessionPRsResponse is the body of GET /sessions/{sessionId}/pr. +type ListSessionPRsResponse struct { + SessionID domain.SessionID `json:"sessionId"` + PRs []SessionPRFacts `json:"prs"` +} + +// ClaimPRRequest is the body of POST /sessions/{sessionId}/pr/claim. +type ClaimPRRequest struct { + PR string `json:"pr" minLength:"1"` + AllowTakeover *bool `json:"allowTakeover,omitempty"` +} + +// ClaimPRResponse is the body of POST /sessions/{sessionId}/pr/claim. +type ClaimPRResponse struct { + OK bool `json:"ok"` + SessionID domain.SessionID `json:"sessionId"` + PRs []SessionPRFacts `json:"prs"` + BranchChanged bool `json:"branchChanged"` + TakenOverFrom []domain.SessionID `json:"takenOverFrom"` +} + // OrchestratorIDParam is the {id} path parameter for orchestrator routes. type OrchestratorIDParam struct { ID string `path:"id" description:"Orchestrator session identifier, e.g. project-orchestrator."` diff --git a/backend/internal/httpd/controllers/sessions.go b/backend/internal/httpd/controllers/sessions.go index f2ef1be1c..a42e8f98c 100644 --- a/backend/internal/httpd/controllers/sessions.go +++ b/backend/internal/httpd/controllers/sessions.go @@ -33,6 +33,8 @@ type SessionService interface { Cleanup(ctx context.Context, project domain.ProjectID) ([]domain.SessionID, error) Rename(ctx context.Context, id domain.SessionID, displayName string) error Send(ctx context.Context, id domain.SessionID, message string) error + ListPRs(ctx context.Context, id domain.SessionID) ([]domain.PRFacts, error) + ClaimPR(ctx context.Context, id domain.SessionID, ref string, opts sessionsvc.ClaimPROptions) (sessionsvc.ClaimPRResult, error) } // SessionsController owns the session routes. Nil keeps routes registered but @@ -47,6 +49,8 @@ func (c *SessionsController) Register(r chi.Router) { r.Post("/sessions", c.spawn) r.Post("/sessions/cleanup", c.cleanup) r.Get("/sessions/{sessionId}", c.get) + r.Get("/sessions/{sessionId}/pr", c.listPRs) + r.Post("/sessions/{sessionId}/pr/claim", c.claimPR) r.Patch("/sessions/{sessionId}", c.rename) r.Post("/sessions/{sessionId}/restore", c.restore) r.Post("/sessions/{sessionId}/kill", c.kill) @@ -116,6 +120,45 @@ func (c *SessionsController) get(w http.ResponseWriter, r *http.Request) { envelope.WriteJSON(w, http.StatusOK, SessionResponse{Session: sess}) } +func (c *SessionsController) listPRs(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "GET", "/api/v1/sessions/{sessionId}/pr") + return + } + prs, err := c.Svc.ListPRs(r.Context(), sessionID(r)) + if err != nil { + envelope.WriteError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, ListSessionPRsResponse{SessionID: sessionID(r), PRs: sessionPRFacts(prs)}) +} + +func (c *SessionsController) claimPR(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/{sessionId}/pr/claim") + return + } + var in ClaimPRRequest + if err := decodeJSON(r, &in); err != nil { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "INVALID_JSON", "Invalid JSON body", nil) + return + } + if strings.TrimSpace(in.PR) == "" { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "PR_REQUIRED", "pr is required", nil) + return + } + allowTakeover := true + if in.AllowTakeover != nil { + allowTakeover = *in.AllowTakeover + } + res, err := c.Svc.ClaimPR(r.Context(), sessionID(r), in.PR, sessionsvc.ClaimPROptions{AllowTakeover: allowTakeover}) + if err != nil { + writeSessionPRError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, ClaimPRResponse{OK: true, SessionID: sessionID(r), PRs: sessionPRFacts(res.PRs), BranchChanged: res.BranchChanged, TakenOverFrom: nonNilSessionIDs(res.TakenOverFrom)}) +} + func (c *SessionsController) rename(w http.ResponseWriter, r *http.Request) { if c.Svc == nil { apispec.NotImplemented(w, r, "PATCH", "/api/v1/sessions/{sessionId}") @@ -300,3 +343,55 @@ func stripUnsafeControlChars(message string) string { return r }, message) } + +func writeSessionPRError(w http.ResponseWriter, r *http.Request, err error) { + var claimed ports.PRClaimedByActiveSessionError + switch { + case errors.Is(err, sessionsvc.ErrInvalidPRRef): + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "INVALID_PR_REF", "PR reference must be a github.com PR URL or a number", nil) + case errors.Is(err, sessionsvc.ErrPRNotFound): + envelope.WriteAPIError(w, r, http.StatusNotFound, "not_found", "PR_NOT_FOUND", "Unknown PR", nil) + case errors.Is(err, sessionsvc.ErrPRNotOpen): + envelope.WriteAPIError(w, r, http.StatusConflict, "conflict", "PR_NOT_OPEN", "PR is not open", nil) + case errors.As(err, &claimed): + envelope.WriteAPIError(w, r, http.StatusConflict, "conflict", "PR_CLAIMED_BY_ACTIVE_SESSION", "PR is already claimed by active session "+string(claimed.Owner)+" (omit --no-takeover to steal)", map[string]any{"ownerSessionId": string(claimed.Owner)}) + case errors.Is(err, sessionsvc.ErrSessionNotClaimable): + envelope.WriteAPIError(w, r, http.StatusUnprocessableEntity, "unprocessable", "SESSION_NOT_CLAIMABLE", "Session cannot claim PRs", nil) + case errors.Is(err, sessionsvc.ErrSessionNoWorkspace): + envelope.WriteAPIError(w, r, http.StatusUnprocessableEntity, "unprocessable", "SESSION_NO_WORKSPACE", "Session has no workspace", nil) + case errors.Is(err, sessionsvc.ErrProjectMismatch): + envelope.WriteAPIError(w, r, http.StatusUnprocessableEntity, "unprocessable", "PR_PROJECT_MISMATCH", "PR does not belong to the session project", nil) + case errors.Is(err, sessionsvc.ErrSCMUnavailable): + envelope.WriteAPIError(w, r, http.StatusServiceUnavailable, "unavailable", "SCM_UNAVAILABLE", "SCM unavailable", nil) + default: + envelope.WriteError(w, r, err) + } +} + +func sessionPRFacts(prs []domain.PRFacts) []SessionPRFacts { + out := make([]SessionPRFacts, 0, len(prs)) + for _, pr := range prs { + out = append(out, SessionPRFacts{URL: pr.URL, Number: pr.Number, State: prState(pr), CI: pr.CI, Review: pr.Review, Mergeability: pr.Mergeability, ReviewComments: pr.ReviewComments, UpdatedAt: pr.UpdatedAt}) + } + return out +} + +func prState(pr domain.PRFacts) string { + switch { + case pr.Merged: + return string(domain.PRStateMerged) + case pr.Closed: + return string(domain.PRStateClosed) + case pr.Draft: + return string(domain.PRStateDraft) + default: + return string(domain.PRStateOpen) + } +} + +func nonNilSessionIDs(ids []domain.SessionID) []domain.SessionID { + if ids == nil { + return []domain.SessionID{} + } + return ids +} diff --git a/backend/internal/httpd/controllers/sessions_test.go b/backend/internal/httpd/controllers/sessions_test.go index 706427d75..a85e071b0 100644 --- a/backend/internal/httpd/controllers/sessions_test.go +++ b/backend/internal/httpd/controllers/sessions_test.go @@ -22,6 +22,8 @@ type fakeSessionService struct { sent string cleanupProjects []domain.ProjectID cleanupResult []domain.SessionID + claimErr error + listPRErr error } func newFakeSessionService() *fakeSessionService { @@ -113,6 +115,27 @@ func (f *fakeSessionService) Send(_ context.Context, _ domain.SessionID, message return nil } +func (f *fakeSessionService) ListPRs(_ context.Context, id domain.SessionID) ([]domain.PRFacts, error) { + if f.listPRErr != nil { + return nil, f.listPRErr + } + if _, ok := f.sessions[id]; !ok { + return nil, apierr.NotFound("SESSION_NOT_FOUND", "Unknown session") + } + return []domain.PRFacts{{URL: "https://github.com/aoagents/agent-orchestrator/pull/142", Number: 142, CI: domain.CIPassing, Review: domain.ReviewRequired, Mergeability: domain.MergeMergeable, UpdatedAt: time.Date(2026, 6, 4, 12, 0, 0, 0, time.UTC)}}, nil +} + +func (f *fakeSessionService) ClaimPR(_ context.Context, id domain.SessionID, ref string, opts sessionsvc.ClaimPROptions) (sessionsvc.ClaimPRResult, error) { + if f.claimErr != nil { + return sessionsvc.ClaimPRResult{}, f.claimErr + } + if _, ok := f.sessions[id]; !ok { + return sessionsvc.ClaimPRResult{}, apierr.NotFound("SESSION_NOT_FOUND", "Unknown session") + } + prs, _ := f.ListPRs(context.Background(), id) + return sessionsvc.ClaimPRResult{PRs: prs, TakenOverFrom: []domain.SessionID{}, BranchChanged: true}, nil +} + func newSessionTestServer(t *testing.T, svc *fakeSessionService) *httptest.Server { t.Helper() log := slog.New(slog.NewTextHandler(io.Discard, nil)) @@ -336,3 +359,71 @@ type sessionBody struct { DisplayName string `json:"displayName"` Status string `json:"status"` } + +func TestSessionsAPI_PRRoutes(t *testing.T) { + srv := newSessionTestServer(t, newFakeSessionService()) + + body, status, _ := doRequest(t, srv, "GET", "/api/v1/sessions/ao-1/pr", "") + if status != http.StatusOK { + t.Fatalf("GET PRs = %d body=%s", status, body) + } + var listed struct { + SessionID string `json:"sessionId"` + PRs []struct { + URL string `json:"url"` + Number int `json:"number"` + State string `json:"state"` + UpdatedAt string `json:"updatedAt"` + } `json:"prs"` + } + mustJSON(t, body, &listed) + if listed.SessionID != "ao-1" || len(listed.PRs) != 1 || listed.PRs[0].State != "open" { + t.Fatalf("GET shape = %#v", listed) + } + + body, status, _ = doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/pr/claim", `{"pr":"142"}`) + if status != http.StatusOK { + t.Fatalf("claim = %d body=%s", status, body) + } + var claimed struct { + OK bool `json:"ok"` + SessionID string `json:"sessionId"` + PRs []any `json:"prs"` + BranchChanged bool `json:"branchChanged"` + TakenOverFrom []string `json:"takenOverFrom"` + } + mustJSON(t, body, &claimed) + if !claimed.OK || claimed.SessionID != "ao-1" || len(claimed.PRs) != 1 || !claimed.BranchChanged || len(claimed.TakenOverFrom) != 0 { + t.Fatalf("claim shape = %#v", claimed) + } +} + +func TestSessionsAPI_ClaimPRErrors(t *testing.T) { + cases := []struct { + name string + body string + err error + code int + want string + }{ + {"bad json", `{`, nil, http.StatusBadRequest, "INVALID_JSON"}, + {"missing pr", `{}`, nil, http.StatusBadRequest, "PR_REQUIRED"}, + {"invalid ref", `{"pr":"x"}`, sessionsvc.ErrInvalidPRRef, http.StatusBadRequest, "INVALID_PR_REF"}, + {"session missing", `{"pr":"142"}`, apierr.NotFound("SESSION_NOT_FOUND", "Unknown session"), http.StatusNotFound, "SESSION_NOT_FOUND"}, + {"pr missing", `{"pr":"142"}`, sessionsvc.ErrPRNotFound, http.StatusNotFound, "PR_NOT_FOUND"}, + {"not open", `{"pr":"142"}`, sessionsvc.ErrPRNotOpen, http.StatusConflict, "PR_NOT_OPEN"}, + {"claimed", `{"pr":"142","allowTakeover":false}`, ports.PRClaimedByActiveSessionError{Owner: "ao-2"}, http.StatusConflict, "PR_CLAIMED_BY_ACTIVE_SESSION"}, + {"not claimable", `{"pr":"142"}`, sessionsvc.ErrSessionNotClaimable, http.StatusUnprocessableEntity, "SESSION_NOT_CLAIMABLE"}, + {"mismatch", `{"pr":"142"}`, sessionsvc.ErrProjectMismatch, http.StatusUnprocessableEntity, "PR_PROJECT_MISMATCH"}, + {"scm", `{"pr":"142"}`, sessionsvc.ErrSCMUnavailable, http.StatusServiceUnavailable, "SCM_UNAVAILABLE"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + svc := newFakeSessionService() + svc.claimErr = tc.err + srv := newSessionTestServer(t, svc) + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/pr/claim", tc.body) + assertErrorCode(t, body, status, tc.code, tc.want) + }) + } +} diff --git a/backend/internal/ports/outbound.go b/backend/internal/ports/outbound.go index efab1ea73..f2c006b16 100644 --- a/backend/internal/ports/outbound.go +++ b/backend/internal/ports/outbound.go @@ -2,6 +2,8 @@ package ports import ( "context" + "errors" + "fmt" "github.com/aoagents/agent-orchestrator/backend/internal/domain" ) @@ -38,6 +40,33 @@ type SCMWriter interface { WriteSCMObservation(ctx context.Context, pr domain.PullRequest, checks []domain.PullRequestCheck, threads []domain.PullRequestReviewThread, comments []domain.PullRequestComment, reviewMode ReviewWriteMode) error } +// PRClaimer atomically moves (or creates) a PR row for a target session and +// persists the live SCM facts observed for that PR in the same transaction. +type PRClaimer interface { + ClaimPR(ctx context.Context, pr domain.PullRequest, checks []domain.PullRequestCheck, threads []domain.PullRequestReviewThread, comments []domain.PullRequestComment, reviewMode ReviewWriteMode, allowActiveTakeover bool) (ClaimOutcome, error) +} + +// ErrPRClaimedByActiveSession is returned by PRClaimer.ClaimPR when takeover is +// explicitly disallowed and the existing owner is still alive. +var ErrPRClaimedByActiveSession = errors.New("pr claimed by active session") + +// PRClaimedByActiveSessionError carries the active owner that blocked a claim. +type PRClaimedByActiveSessionError struct { + Owner domain.SessionID +} + +func (e PRClaimedByActiveSessionError) Error() string { + return fmt.Sprintf("%s: %s", ErrPRClaimedByActiveSession, e.Owner) +} + +func (e PRClaimedByActiveSessionError) Unwrap() error { return ErrPRClaimedByActiveSession } + +// ClaimOutcome describes what owner, if any, a successful claim replaced. +type ClaimOutcome struct { + PreviousOwner domain.SessionID + OwnerTerminated bool +} + // AgentMessenger injects a message into a running agent. type AgentMessenger interface { Send(ctx context.Context, id domain.SessionID, message string) error diff --git a/backend/internal/ports/pr_observations.go b/backend/internal/ports/pr_observations.go index 91eac64b0..eb29615f7 100644 --- a/backend/internal/ports/pr_observations.go +++ b/backend/internal/ports/pr_observations.go @@ -1,6 +1,20 @@ package ports -import "github.com/aoagents/agent-orchestrator/backend/internal/domain" +import ( + "context" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +// ErrSCMPRNotFound is the legacy PR-observation not-found sentinel. It aliases +// the provider-neutral SCM sentinel so old PRObservation callers and new SCM +// callers compose under errors.Is. +var ErrSCMPRNotFound = ErrSCMNotFound + +// PRObserver fetches one legacy PR observation by canonical PR URL. +type PRObserver interface { + Observe(ctx context.Context, prURL string) (PRObservation, error) +} // PRObservation is what the SCM poller reports for one PR. Fetched is the // failed-fetch guard: when false the rest is meaningless and lifecycle must not diff --git a/backend/internal/service/pr/manager_test.go b/backend/internal/service/pr/manager_test.go index 92acea5b5..35cd0c996 100644 --- a/backend/internal/service/pr/manager_test.go +++ b/backend/internal/service/pr/manager_test.go @@ -22,6 +22,12 @@ func (f *fakeWriter) WritePR(_ context.Context, pr domain.PullRequest, checks [] return nil } +func (f *fakeWriter) ClaimPR(_ context.Context, url string, sessionID domain.SessionID, observation ports.PRObservation, _ bool) (ports.ClaimOutcome, error) { + pr := domain.PullRequest{URL: url, SessionID: sessionID, Number: observation.Number, Draft: observation.Draft, Merged: observation.Merged, Closed: observation.Closed, CI: observation.CI, Review: observation.Review, Mergeability: observation.Mergeability} + f.pr[sessionID] = pr + return ports.ClaimOutcome{}, nil +} + type fakeLifecycle struct { observed []ports.PRObservation } diff --git a/backend/internal/service/session/claim_pr.go b/backend/internal/service/session/claim_pr.go new file mode 100644 index 000000000..6d6cd92b7 --- /dev/null +++ b/backend/internal/service/session/claim_pr.go @@ -0,0 +1,395 @@ +package session + +import ( + "context" + "errors" + "fmt" + "net/url" + "sort" + "strconv" + "strings" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/httpd/apierr" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +var ( + // ErrInvalidPRRef is returned when a claim request does not name a GitHub PR URL or positive PR number. + ErrInvalidPRRef = errors.New("session: invalid pr ref") + // ErrPRNotFound is returned when the SCM provider has no matching pull request. + ErrPRNotFound = errors.New("session: pr not found") + // ErrPRNotOpen is returned when a PR is draft, merged, or closed and therefore cannot be claimed. + ErrPRNotOpen = errors.New("session: pr not open") + // ErrSCMUnavailable is returned when live SCM facts cannot be fetched. + ErrSCMUnavailable = errors.New("session: scm unavailable") + // ErrProjectMismatch is returned when the PR repository does not match the session project repository. + ErrProjectMismatch = errors.New("session: pr project mismatch") + // ErrSessionNotClaimable is returned when an orchestrator session tries to claim a PR. + ErrSessionNotClaimable = errors.New("session: not claimable") + // ErrSessionNoWorkspace is returned when a session has no workspace path to associate with PR work. + ErrSessionNoWorkspace = errors.New("session: no workspace") +) + +// ClaimPROptions controls PR claim conflict behavior. +type ClaimPROptions struct { + AllowTakeover bool +} + +// ClaimPRResult is the session PR read model returned after a claim. +type ClaimPRResult struct { + PRs []domain.PRFacts + BranchChanged bool + TakenOverFrom []domain.SessionID + DonorWasTerminated bool +} + +// ListPRs returns all PRs currently owned by a session, ordered for display. +func (s *Service) ListPRs(ctx context.Context, id domain.SessionID) ([]domain.PRFacts, error) { + if _, ok, err := s.store.GetSession(ctx, id); err != nil { + return nil, fmt.Errorf("get %s: %w", id, err) + } else if !ok { + return nil, apierr.NotFound("SESSION_NOT_FOUND", "Unknown session") + } + return s.listPRFacts(ctx, id) +} + +// ClaimPR attaches a live GitHub PR to a worker session and persists the current SCM facts atomically. +func (s *Service) ClaimPR(ctx context.Context, id domain.SessionID, ref string, opts ClaimPROptions) (ClaimPRResult, error) { + rec, ok, err := s.store.GetSession(ctx, id) + if err != nil { + return ClaimPRResult{}, fmt.Errorf("get %s: %w", id, err) + } + if !ok { + return ClaimPRResult{}, apierr.NotFound("SESSION_NOT_FOUND", "Unknown session") + } + if rec.IsTerminated { + return ClaimPRResult{}, sessionmanagerAPIError("SESSION_TERMINATED", "Session is terminated") + } + if rec.Kind == domain.KindOrchestrator { + return ClaimPRResult{}, ErrSessionNotClaimable + } + if strings.TrimSpace(rec.Metadata.WorkspacePath) == "" { + return ClaimPRResult{}, ErrSessionNoWorkspace + } + project, ok, err := s.store.GetProject(ctx, string(rec.ProjectID)) + if err != nil { + return ClaimPRResult{}, fmt.Errorf("project %s: %w", rec.ProjectID, err) + } + if !ok { + return ClaimPRResult{}, apierr.Invalid("PROJECT_NOT_RESOLVABLE", "Project is not registered or has no repo — register it with `ao project add`", nil) + } + prURL, number, err := normalizePRRef(ref, project.RepoOriginURL) + if err != nil { + return ClaimPRResult{}, err + } + if err := requireSameGitHubRepo(prURL, project.RepoOriginURL); err != nil { + return ClaimPRResult{}, err + } + if s.scm == nil || s.prClaimer == nil { + return ClaimPRResult{}, ErrSCMUnavailable + } + repo, err := scmRepoForClaim(s.scm, project.RepoOriginURL, prURL) + if err != nil { + return ClaimPRResult{}, err + } + refSpec := ports.SCMPRRef{Repo: repo, Number: number, URL: prURL} + obs, err := s.fetchClaimObservation(ctx, refSpec) + if err != nil { + return ClaimPRResult{}, err + } + if obs.PR.Number == 0 { + obs.PR.Number = number + } + if obs.PR.URL == "" { + obs.PR.URL = prURL + } + if obs.PR.Draft || obs.PR.Merged || obs.PR.Closed { + return ClaimPRResult{}, ErrPRNotOpen + } + reviewMode, err := s.enrichClaimReviews(ctx, refSpec, &obs) + if err != nil { + return ClaimPRResult{}, err + } + now := s.clock().UTC() + pr, checks, threads, comments := claimRowsFromSCM(id, obs, now) + outcome, err := s.prClaimer.ClaimPR(ctx, pr, checks, threads, comments, reviewMode, opts.AllowTakeover) + if err != nil { + return ClaimPRResult{}, err + } + prs, err := s.listPRFacts(ctx, id) + if err != nil { + return ClaimPRResult{}, err + } + prs = claimedFirst(prs, prURL) + // TODO: implement workspace branch checkout. Until then, leave BranchChanged + // false and let CLI output omit the checkout line rather than claiming the + // session was already on the PR branch. + res := ClaimPRResult{PRs: prs, BranchChanged: false, DonorWasTerminated: outcome.OwnerTerminated} + if outcome.PreviousOwner != "" && outcome.PreviousOwner != id { + res.TakenOverFrom = []domain.SessionID{outcome.PreviousOwner} + } + return res, nil +} + +func (s *Service) fetchClaimObservation(ctx context.Context, ref ports.SCMPRRef) (ports.SCMObservation, error) { + batch, err := s.scm.FetchPullRequests(ctx, []ports.SCMPRRef{ref}) + if err != nil { + if errors.Is(err, ports.ErrSCMNotFound) { + return ports.SCMObservation{}, ErrPRNotFound + } + return ports.SCMObservation{}, fmt.Errorf("%w: %w", ErrSCMUnavailable, err) + } + if len(batch) == 0 { + return ports.SCMObservation{}, ErrPRNotFound + } + obs := batch[0] + if !obs.Fetched { + return ports.SCMObservation{}, ErrSCMUnavailable + } + return obs, nil +} + +func (s *Service) enrichClaimReviews(ctx context.Context, ref ports.SCMPRRef, obs *ports.SCMObservation) (ports.ReviewWriteMode, error) { + review, err := s.scm.FetchReviewThreads(ctx, ref) + if err != nil { + if errors.Is(err, ports.ErrSCMNotFound) { + return ports.ReviewWritePreserve, ErrPRNotFound + } + return ports.ReviewWritePreserve, fmt.Errorf("%w: %w", ErrSCMUnavailable, err) + } + if review.Decision != "" { + obs.Review.Decision = review.Decision + } + obs.Review.Threads = review.Threads + obs.Review.Partial = review.Partial + if review.Partial { + return ports.ReviewWriteMerge, nil + } + return ports.ReviewWriteReplace, nil +} + +func scmRepoForClaim(provider scmProvider, projectOrigin, prURL string) (ports.SCMRepo, error) { + if repo, ok := provider.ParseRepository(projectOrigin); ok { + return repo, nil + } + owner, name, _, err := parseGitHubPRURL(prURL) + if err != nil { + return ports.SCMRepo{}, ErrInvalidPRRef + } + return ports.SCMRepo{Provider: "github", Host: "github.com", Owner: owner, Name: name, Repo: owner + "/" + name}, nil +} + +func claimRowsFromSCM(sessionID domain.SessionID, obs ports.SCMObservation, now time.Time) (domain.PullRequest, []domain.PullRequestCheck, []domain.PullRequestReviewThread, []domain.PullRequestComment) { + observedAt := obs.ObservedAt + if observedAt.IsZero() { + observedAt = now + } + pr := domain.PullRequest{ + URL: firstNonEmpty(obs.PR.URL, obs.PR.HTMLURL), + SessionID: sessionID, + Number: obs.PR.Number, + Draft: obs.PR.Draft, + Merged: obs.PR.Merged, + Closed: obs.PR.Closed, + CI: domain.CIState(firstNonEmpty(obs.CI.Summary, string(domain.CIUnknown))), + Review: domain.ReviewDecision(firstNonEmpty(obs.Review.Decision, string(domain.ReviewNone))), + Mergeability: domain.Mergeability(firstNonEmpty(obs.Mergeability.State, string(domain.MergeUnknown))), + UpdatedAt: now, + Provider: obs.Provider, + Host: obs.Host, + Repo: obs.Repo, + SourceBranch: obs.PR.SourceBranch, + TargetBranch: obs.PR.TargetBranch, + HeadSHA: obs.PR.HeadSHA, + Title: obs.PR.Title, + Additions: obs.PR.Additions, + Deletions: obs.PR.Deletions, + ChangedFiles: obs.PR.ChangedFiles, + Author: obs.PR.Author, + BaseSHA: obs.PR.BaseSHA, + MergeCommitSHA: obs.PR.MergeCommitSHA, + ProviderState: obs.PR.ProviderState, + ProviderMergeable: obs.PR.ProviderMergeable, + ProviderMergeStateStatus: obs.PR.ProviderMergeStateStatus, + HTMLURL: obs.PR.HTMLURL, + CreatedAtProvider: obs.PR.CreatedAtProvider, + UpdatedAtProvider: obs.PR.UpdatedAtProvider, + MergedAtProvider: obs.PR.MergedAtProvider, + ClosedAtProvider: obs.PR.ClosedAtProvider, + ObservedAt: observedAt, + CIObservedAt: observedAt, + ReviewObservedAt: observedAt, + } + checks := make([]domain.PullRequestCheck, 0, len(obs.CI.Checks)) + for _, ch := range obs.CI.Checks { + checks = append(checks, domain.PullRequestCheck{Name: ch.Name, CommitHash: obs.CI.HeadSHA, Status: domain.PRCheckStatus(ch.Status), Conclusion: ch.Conclusion, URL: ch.URL, Details: ch.ProviderID, LogTail: ch.LogTail, CreatedAt: now}) + } + threads := make([]domain.PullRequestReviewThread, 0, len(obs.Review.Threads)) + commentCount := 0 + for _, th := range obs.Review.Threads { + commentCount += len(th.Comments) + } + comments := make([]domain.PullRequestComment, 0, commentCount) + for _, th := range obs.Review.Threads { + threads = append(threads, domain.PullRequestReviewThread{ThreadID: th.ID, Path: th.Path, Line: th.Line, Resolved: th.Resolved, IsBot: th.IsBot, UpdatedAt: now}) + for _, c := range th.Comments { + comments = append(comments, domain.PullRequestComment{ThreadID: th.ID, ID: c.ID, Author: c.Author, File: th.Path, Line: th.Line, Body: c.Body, URL: c.URL, Resolved: th.Resolved, IsBot: c.IsBot || th.IsBot, CreatedAt: now}) + } + } + return pr, checks, threads, comments +} + +func sessionmanagerAPIError(code, message string) error { + return apierr.Conflict(code, message, nil) +} + +func (s *Service) listPRFacts(ctx context.Context, id domain.SessionID) ([]domain.PRFacts, error) { + prs, err := s.store.ListPRsBySession(ctx, id) + if err != nil { + return nil, err + } + facts := make([]domain.PRFacts, 0, len(prs)) + for _, pr := range prs { + comments, err := s.store.ListPRComments(ctx, pr.URL) + if err != nil { + return nil, err + } + facts = append(facts, pullRequestFacts(pr, comments)) + } + sortPRFacts(facts) + return facts, nil +} + +func pullRequestFacts(pr domain.PullRequest, comments []domain.PullRequestComment) domain.PRFacts { + unresolved := false + for _, c := range comments { + if !c.Resolved { + unresolved = true + break + } + } + return domain.PRFacts{URL: pr.URL, Number: pr.Number, Draft: pr.Draft, Merged: pr.Merged, Closed: pr.Closed, CI: pr.CI, Review: pr.Review, Mergeability: pr.Mergeability, ReviewComments: unresolved, UpdatedAt: pr.UpdatedAt} +} + +func sortPRFacts(prs []domain.PRFacts) { + sort.SliceStable(prs, func(i, j int) bool { + ia, ja := prActive(prs[i]), prActive(prs[j]) + if ia != ja { + return ia + } + return prs[i].UpdatedAt.After(prs[j].UpdatedAt) + }) +} + +func prActive(pr domain.PRFacts) bool { return !pr.Merged && !pr.Closed } + +func claimedFirst(prs []domain.PRFacts, prURL string) []domain.PRFacts { + idx := -1 + for i, pr := range prs { + if pr.URL == prURL { + idx = i + break + } + } + if idx <= 0 { + return prs + } + claimed := prs[idx] + copy(prs[1:idx+1], prs[0:idx]) + prs[0] = claimed + return prs +} + +func normalizePRRef(ref, repoOrigin string) (string, int, error) { + ref = strings.TrimPrefix(strings.TrimSpace(ref), "#") + if ref == "" { + return "", 0, ErrInvalidPRRef + } + if n, err := strconv.Atoi(ref); err == nil && n > 0 { + owner, repo, err := githubRepoFromURL(repoOrigin) + if err != nil { + return "", 0, ErrInvalidPRRef + } + return fmt.Sprintf("https://github.com/%s/%s/pull/%d", owner, repo, n), n, nil + } + owner, repo, n, err := parseGitHubPRURL(ref) + if err != nil || owner == "" || repo == "" || n <= 0 { + return "", 0, ErrInvalidPRRef + } + return fmt.Sprintf("https://github.com/%s/%s/pull/%d", owner, repo, n), n, nil +} + +func requireSameGitHubRepo(prURL, repoOrigin string) error { + if strings.TrimSpace(repoOrigin) == "" { + return nil + } + po, pr, _, err := parseGitHubPRURL(prURL) + if err != nil { + return ErrInvalidPRRef + } + ro, rr, err := githubRepoFromURL(repoOrigin) + if err != nil { + return ErrInvalidPRRef + } + if !strings.EqualFold(po, ro) || !strings.EqualFold(pr, rr) { + return ErrProjectMismatch + } + return nil +} + +func parseGitHubPRURL(raw string) (string, string, int, error) { + u, err := url.Parse(raw) + if err != nil { + return "", "", 0, err + } + if !strings.EqualFold(u.Scheme, "https") || !strings.EqualFold(u.Hostname(), "github.com") { + return "", "", 0, ErrInvalidPRRef + } + parts := strings.Split(strings.Trim(u.Path, "/"), "/") + if len(parts) != 4 || parts[2] != "pull" { + return "", "", 0, ErrInvalidPRRef + } + n, err := strconv.Atoi(parts[3]) + if err != nil || n <= 0 { + return "", "", 0, ErrInvalidPRRef + } + return parts[0], strings.TrimSuffix(parts[1], ".git"), n, nil +} + +func githubRepoFromURL(raw string) (string, string, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", "", ErrInvalidPRRef + } + if strings.HasPrefix(raw, "git@github.com:") { + path := strings.TrimPrefix(raw, "git@github.com:") + parts := strings.Split(strings.TrimSuffix(path, ".git"), "/") + if len(parts) == 2 && parts[0] != "" && parts[1] != "" { + return parts[0], parts[1], nil + } + return "", "", ErrInvalidPRRef + } + u, err := url.Parse(raw) + if err != nil { + return "", "", err + } + if !strings.EqualFold(u.Hostname(), "github.com") { + return "", "", ErrInvalidPRRef + } + parts := strings.Split(strings.Trim(u.Path, "/"), "/") + if len(parts) < 2 || parts[0] == "" || parts[1] == "" { + return "", "", ErrInvalidPRRef + } + return parts[0], strings.TrimSuffix(parts[1], ".git"), nil +} + +func firstNonEmpty(values ...string) string { + for _, v := range values { + if v != "" { + return v + } + } + return "" +} diff --git a/backend/internal/service/session/service.go b/backend/internal/service/session/service.go index 8eefd991f..b5667158e 100644 --- a/backend/internal/service/session/service.go +++ b/backend/internal/service/session/service.go @@ -20,6 +20,9 @@ type Store interface { ListAllSessions(ctx context.Context) ([]domain.SessionRecord, error) RenameSession(ctx context.Context, id domain.SessionID, displayName string, updatedAt time.Time) (bool, error) GetDisplayPRFactsForSession(ctx context.Context, id domain.SessionID) (domain.PRFacts, bool, error) + ListPRsBySession(ctx context.Context, sessionID domain.SessionID) ([]domain.PullRequest, error) + ListPRComments(ctx context.Context, prURL string) ([]domain.PullRequestComment, error) + GetProject(ctx context.Context, id string) (domain.ProjectRecord, bool, error) } // ListFilter captures API-facing session list query filters. @@ -40,17 +43,51 @@ type commander interface { Cleanup(ctx context.Context, project domain.ProjectID) ([]domain.SessionID, error) } +type scmProvider interface { + ParseRepository(remote string) (ports.SCMRepo, bool) + FetchPullRequests(ctx context.Context, refs []ports.SCMPRRef) ([]ports.SCMObservation, error) + FetchReviewThreads(ctx context.Context, ref ports.SCMPRRef) (ports.SCMReviewObservation, error) +} + // Service is the controller-facing session service. It delegates command-side // session operations to the internal sessionmanager.Manager and owns read-model // assembly, including user-facing display status derivation. type Service struct { - manager commander - store Store + manager commander + store Store + prClaimer ports.PRClaimer + scm scmProvider + clock func() time.Time } // New wires a controller-facing session service over an internal session Manager. func New(manager *sessionmanager.Manager, store Store) *Service { - return &Service{manager: manager, store: store} + return NewWithDeps(Deps{Manager: manager, Store: store}) +} + +// Deps are optional collaborators for the session service. The default New +// path keeps existing tests and callers small; daemon wiring uses NewWithDeps +// to supply SCM observation for PR claiming. +type Deps struct { + Manager commander + Store Store + PRClaimer ports.PRClaimer + SCM scmProvider + Clock func() time.Time +} + +// NewWithDeps wires a session service with optional PR-claim dependencies. +func NewWithDeps(d Deps) *Service { + s := &Service{manager: d.Manager, store: d.Store, prClaimer: d.PRClaimer, scm: d.SCM, clock: d.Clock} + if s.prClaimer == nil { + if w, ok := d.Store.(ports.PRClaimer); ok { + s.prClaimer = w + } + } + if s.clock == nil { + s.clock = time.Now + } + return s } // Spawn creates a session and returns the API-facing read model. diff --git a/backend/internal/service/session/service_test.go b/backend/internal/service/session/service_test.go index 702f36731..71ed69c34 100644 --- a/backend/internal/service/session/service_test.go +++ b/backend/internal/service/session/service_test.go @@ -15,11 +15,12 @@ import ( type fakeStore struct { sessions map[domain.SessionID]domain.SessionRecord pr map[domain.SessionID]domain.PRFacts + projects map[string]domain.ProjectRecord num int } func newFakeStore() *fakeStore { - return &fakeStore{sessions: map[domain.SessionID]domain.SessionRecord{}, pr: map[domain.SessionID]domain.PRFacts{}} + return &fakeStore{sessions: map[domain.SessionID]domain.SessionRecord{}, pr: map[domain.SessionID]domain.PRFacts{}, projects: map[string]domain.ProjectRecord{}} } func (f *fakeStore) CreateSession(_ context.Context, rec domain.SessionRecord) (domain.SessionRecord, error) { @@ -68,6 +69,23 @@ func (f *fakeStore) GetDisplayPRFactsForSession(_ context.Context, id domain.Ses return pr, ok, nil } +func (f *fakeStore) ListPRsBySession(_ context.Context, id domain.SessionID) ([]domain.PullRequest, error) { + pr, ok := f.pr[id] + if !ok { + return nil, nil + } + return []domain.PullRequest{{URL: pr.URL, SessionID: id, Number: pr.Number, Draft: pr.Draft, Merged: pr.Merged, Closed: pr.Closed, CI: pr.CI, Review: pr.Review, Mergeability: pr.Mergeability, UpdatedAt: pr.UpdatedAt}}, nil +} + +func (f *fakeStore) ListPRComments(context.Context, string) ([]domain.PullRequestComment, error) { + return nil, nil +} + +func (f *fakeStore) GetProject(_ context.Context, id string) (domain.ProjectRecord, bool, error) { + p, ok := f.projects[id] + return p, ok, nil +} + func TestSessionListDerivesStatusFromPRFacts(t *testing.T) { st := newFakeStore() st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityActive}} @@ -168,3 +186,109 @@ func TestSpawnOrchestratorNoCleanSkipsKills(t *testing.T) { t.Fatalf("clean=false must spawn without kills: killed=%v spawned=%v", fc.killed, fc.spawned) } } + +type fakePRClaimer struct { + out errorFreeClaimOutcome + err error +} + +type errorFreeClaimOutcome struct { + ports.ClaimOutcome +} + +func (f fakePRClaimer) ClaimPR(context.Context, domain.PullRequest, []domain.PullRequestCheck, []domain.PullRequestReviewThread, []domain.PullRequestComment, ports.ReviewWriteMode, bool) (ports.ClaimOutcome, error) { + return f.out.ClaimOutcome, f.err +} + +type fakeSCM struct { + obs ports.SCMObservation + review ports.SCMReviewObservation + fetchErr error + reviewErr error +} + +func (f fakeSCM) ParseRepository(remote string) (ports.SCMRepo, bool) { + owner, repo, err := githubRepoFromURL(remote) + if err != nil { + return ports.SCMRepo{}, false + } + return ports.SCMRepo{Provider: "github", Host: "github.com", Owner: owner, Name: repo, Repo: owner + "/" + repo}, true +} + +func (f fakeSCM) FetchPullRequests(context.Context, []ports.SCMPRRef) ([]ports.SCMObservation, error) { + if f.fetchErr != nil { + return nil, f.fetchErr + } + if !f.obs.Fetched && f.obs.PR.URL == "" && f.obs.PR.Number == 0 { + return nil, nil + } + return []ports.SCMObservation{f.obs}, nil +} + +func (f fakeSCM) FetchReviewThreads(context.Context, ports.SCMPRRef) (ports.SCMReviewObservation, error) { + return f.review, f.reviewErr +} + +func TestClaimPRMapsObserverAndStoreErrors(t *testing.T) { + st := newFakeStore() + now := time.Date(2026, 6, 4, 12, 0, 0, 0, time.UTC) + st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Kind: domain.KindWorker, Metadata: domain.SessionMetadata{WorkspacePath: "/ws"}} + st.projects["mer"] = domain.ProjectRecord{ID: "mer", RepoOriginURL: "https://github.com/acme/repo"} + + cases := []struct { + name string + svc *Service + want error + }{ + {"missing scm", NewWithDeps(Deps{Store: st}), ErrSCMUnavailable}, + {"not found", NewWithDeps(Deps{Store: st, PRClaimer: fakePRClaimer{}, SCM: fakeSCM{fetchErr: ports.ErrSCMNotFound}}), ErrPRNotFound}, + {"closed", NewWithDeps(Deps{Store: st, PRClaimer: fakePRClaimer{}, SCM: fakeSCM{obs: ports.SCMObservation{Fetched: true, Provider: "github", Host: "github.com", Repo: "acme/repo", PR: ports.SCMPRObservation{URL: "https://github.com/acme/repo/pull/7", Number: 7, Closed: true}}}}), ErrPRNotOpen}, + {"active owner", NewWithDeps(Deps{Store: st, PRClaimer: fakePRClaimer{err: ports.PRClaimedByActiveSessionError{Owner: "mer-2"}}, SCM: fakeSCM{obs: ports.SCMObservation{Fetched: true, Provider: "github", Host: "github.com", Repo: "acme/repo", PR: ports.SCMPRObservation{URL: "https://github.com/acme/repo/pull/7", Number: 7}}}}), ports.ErrPRClaimedByActiveSession}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, err := tc.svc.ClaimPR(context.Background(), "mer-1", "7", ClaimPROptions{AllowTakeover: false}) + if !errors.Is(err, tc.want) { + t.Fatalf("err=%v, want %v", err, tc.want) + } + }) + } + + st.pr["mer-1"] = domain.PRFacts{URL: "https://github.com/acme/repo/pull/7", Number: 7, CI: domain.CIPassing, UpdatedAt: now} + svc := NewWithDeps(Deps{Store: st, PRClaimer: fakePRClaimer{out: errorFreeClaimOutcome{ports.ClaimOutcome{PreviousOwner: "mer-2"}}}, SCM: fakeSCM{obs: ports.SCMObservation{Fetched: true, Provider: "github", Host: "github.com", Repo: "acme/repo", PR: ports.SCMPRObservation{URL: "https://github.com/acme/repo/pull/7", Number: 7}}}}) + res, err := svc.ClaimPR(context.Background(), "mer-1", "7", ClaimPROptions{AllowTakeover: true}) + if err != nil { + t.Fatal(err) + } + if len(res.TakenOverFrom) != 1 || res.TakenOverFrom[0] != "mer-2" || len(res.PRs) != 1 || res.PRs[0].URL == "" { + t.Fatalf("claim result = %+v", res) + } +} + +func TestListPRsOrdersActiveBeforeClosedThenUpdatedDesc(t *testing.T) { + st := newFakeStore() + st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Kind: domain.KindWorker} + now := time.Date(2026, 6, 4, 12, 0, 0, 0, time.UTC) + st.pr = map[domain.SessionID]domain.PRFacts{} + stList := &multiPRFakeStore{fakeStore: st, prs: []domain.PullRequest{ + {URL: "closed-new", SessionID: "mer-1", Number: 1, Closed: true, UpdatedAt: now.Add(2 * time.Hour)}, + {URL: "open-old", SessionID: "mer-1", Number: 2, UpdatedAt: now}, + {URL: "open-new", SessionID: "mer-1", Number: 3, UpdatedAt: now.Add(time.Hour)}, + }} + got, err := (&Service{store: stList}).ListPRs(context.Background(), "mer-1") + if err != nil { + t.Fatal(err) + } + if len(got) != 3 || got[0].URL != "open-new" || got[1].URL != "open-old" || got[2].URL != "closed-new" { + t.Fatalf("order = %+v", got) + } +} + +type multiPRFakeStore struct { + *fakeStore + prs []domain.PullRequest +} + +func (f *multiPRFakeStore) ListPRsBySession(context.Context, domain.SessionID) ([]domain.PullRequest, error) { + return f.prs, nil +} diff --git a/backend/internal/storage/sqlite/gen/pr.sql.go b/backend/internal/storage/sqlite/gen/pr.sql.go index 6b2eb02e0..754245341 100644 --- a/backend/internal/storage/sqlite/gen/pr.sql.go +++ b/backend/internal/storage/sqlite/gen/pr.sql.go @@ -13,6 +13,40 @@ import ( "github.com/aoagents/agent-orchestrator/backend/internal/domain" ) +const claimPRForSession = `-- name: ClaimPRForSession :exec +INSERT INTO pr (url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT (url) DO UPDATE SET + session_id = excluded.session_id, + review_decision = excluded.review_decision, + updated_at = excluded.updated_at +` + +type ClaimPRForSessionParams struct { + URL string + SessionID domain.SessionID + Number int64 + PRState domain.PRState + ReviewDecision domain.ReviewDecision + CIState domain.CIState + Mergeability domain.Mergeability + UpdatedAt time.Time +} + +func (q *Queries) ClaimPRForSession(ctx context.Context, arg ClaimPRForSessionParams) error { + _, err := q.db.ExecContext(ctx, claimPRForSession, + arg.URL, + arg.SessionID, + arg.Number, + arg.PRState, + arg.ReviewDecision, + arg.CIState, + arg.Mergeability, + arg.UpdatedAt, + ) + return err +} + const getDisplayPRFactsBySession = `-- name: GetDisplayPRFactsBySession :one SELECT pr.url, @@ -21,6 +55,7 @@ SELECT pr.review_decision, pr.ci_state, pr.mergeability, + pr.updated_at, EXISTS ( SELECT 1 FROM pr_comment @@ -43,6 +78,7 @@ type GetDisplayPRFactsBySessionRow struct { ReviewDecision domain.ReviewDecision CIState domain.CIState Mergeability domain.Mergeability + UpdatedAt time.Time ReviewComments bool } @@ -56,6 +92,7 @@ func (q *Queries) GetDisplayPRFactsBySession(ctx context.Context, sessionID doma &i.ReviewDecision, &i.CIState, &i.Mergeability, + &i.UpdatedAt, &i.ReviewComments, ) return i, err @@ -120,6 +157,27 @@ func (q *Queries) GetPR(ctx context.Context, url string) (PR, error) { return i, err } +const getPRClaimAndOwner = `-- name: GetPRClaimAndOwner :one +SELECT pr.session_id, sessions.is_terminated +FROM pr +JOIN sessions ON sessions.id = pr.session_id +WHERE pr.url = ? +` + +type GetPRClaimAndOwnerRow struct { + SessionID domain.SessionID + IsTerminated bool +} + +// Returns the current owner of a PR URL plus whether that owner is +// terminated. Used by the takeover guard inside the claim tx. +func (q *Queries) GetPRClaimAndOwner(ctx context.Context, url string) (GetPRClaimAndOwnerRow, error) { + row := q.db.QueryRowContext(ctx, getPRClaimAndOwner, url) + var i GetPRClaimAndOwnerRow + err := row.Scan(&i.SessionID, &i.IsTerminated) + return i, err +} + const listPRsBySession = `-- name: ListPRsBySession :many SELECT url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at, diff --git a/backend/internal/storage/sqlite/migrations/0006_pr_session_changed_cdc.sql b/backend/internal/storage/sqlite/migrations/0006_pr_session_changed_cdc.sql new file mode 100644 index 000000000..815ca483c --- /dev/null +++ b/backend/internal/storage/sqlite/migrations/0006_pr_session_changed_cdc.sql @@ -0,0 +1,345 @@ +-- +goose Up +-- +goose StatementBegin +DROP TRIGGER IF EXISTS pr_review_threads_cdc_update; +DROP TRIGGER IF EXISTS pr_review_threads_cdc_insert; +DROP TRIGGER IF EXISTS sessions_cdc_insert; +DROP TRIGGER IF EXISTS sessions_cdc_update; +DROP TRIGGER IF EXISTS pr_cdc_insert; +DROP TRIGGER IF EXISTS pr_cdc_update; +DROP TRIGGER IF EXISTS pr_session_cdc_update; +DROP TRIGGER IF EXISTS pr_checks_cdc_insert; +DROP TRIGGER IF EXISTS pr_checks_cdc_update; + +CREATE TABLE change_log_new ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + project_id TEXT NOT NULL REFERENCES projects (id), + session_id TEXT REFERENCES sessions (id), + event_type TEXT NOT NULL + CHECK (event_type IN ( + 'session_created', + 'session_updated', + 'pr_created', + 'pr_updated', + 'pr_check_recorded', + 'pr_session_changed', + 'pr_review_thread_added', + 'pr_review_thread_resolved' + )), + payload TEXT NOT NULL CHECK (json_valid(payload)), + created_at TIMESTAMP NOT NULL DEFAULT (datetime('now')) +); + +INSERT INTO change_log_new (seq, project_id, session_id, event_type, payload, created_at) +SELECT seq, project_id, session_id, event_type, payload, created_at +FROM change_log; + +DROP INDEX IF EXISTS idx_change_log_project; +DROP TABLE change_log; +ALTER TABLE change_log_new RENAME TO change_log; +CREATE INDEX idx_change_log_project ON change_log (project_id, seq); +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_review_threads_cdc_insert +AFTER INSERT ON pr_review_threads +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_review_thread_added', + json_object( + 'pr', NEW.pr_url, + 'thread', NEW.thread_id, + 'path', NEW.path, + 'line', NEW.line, + 'resolved', json(CASE WHEN NEW.resolved THEN 'true' ELSE 'false' END), + 'isBot', json(CASE WHEN NEW.is_bot THEN 'true' ELSE 'false' END) + ), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_review_threads_cdc_update +AFTER UPDATE ON pr_review_threads +WHEN OLD.resolved <> NEW.resolved +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_review_thread_resolved', + json_object( + 'pr', NEW.pr_url, + 'thread', NEW.thread_id, + 'path', NEW.path, + 'line', NEW.line, + 'resolved', json(CASE WHEN NEW.resolved THEN 'true' ELSE 'false' END) + ), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER sessions_cdc_insert +AFTER INSERT ON sessions +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES (NEW.project_id, NEW.id, 'session_created', + json_object('id', NEW.id, 'activity', NEW.activity_state, 'isTerminated', json(CASE WHEN NEW.is_terminated THEN 'true' ELSE 'false' END)), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER sessions_cdc_update +AFTER UPDATE ON sessions +WHEN OLD.activity_state <> NEW.activity_state + OR OLD.is_terminated <> NEW.is_terminated +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES (NEW.project_id, NEW.id, 'session_updated', + json_object('id', NEW.id, 'activity', NEW.activity_state, 'isTerminated', json(CASE WHEN NEW.is_terminated THEN 'true' ELSE 'false' END)), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_cdc_insert +AFTER INSERT ON pr +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ((SELECT project_id FROM sessions WHERE id = NEW.session_id), NEW.session_id, 'pr_created', + json_object('url', NEW.url, 'session', NEW.session_id, 'state', NEW.pr_state, + 'ci', NEW.ci_state, 'review', NEW.review_decision, 'mergeability', NEW.mergeability), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_cdc_update +AFTER UPDATE ON pr +WHEN OLD.pr_state <> NEW.pr_state + OR OLD.ci_state <> NEW.ci_state + OR OLD.review_decision <> NEW.review_decision + OR OLD.mergeability <> NEW.mergeability +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ((SELECT project_id FROM sessions WHERE id = NEW.session_id), NEW.session_id, 'pr_updated', + json_object('url', NEW.url, 'session', NEW.session_id, 'state', NEW.pr_state, + 'ci', NEW.ci_state, 'review', NEW.review_decision, 'mergeability', NEW.mergeability), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_session_cdc_update +AFTER UPDATE ON pr +WHEN OLD.session_id <> NEW.session_id +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT project_id FROM sessions WHERE id = NEW.session_id), + NEW.session_id, + 'pr_session_changed', + json_object( + 'url', NEW.url, + 'fromSession', OLD.session_id, + 'toSession', NEW.session_id), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_checks_cdc_insert +AFTER INSERT ON pr_checks +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_check_recorded', + json_object('pr', NEW.pr_url, 'name', NEW.name, 'commit', NEW.commit_hash, 'status', NEW.status), + NEW.created_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_checks_cdc_update +AFTER UPDATE ON pr_checks +WHEN OLD.status <> NEW.status +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_check_recorded', + json_object('pr', NEW.pr_url, 'name', NEW.name, 'commit', NEW.commit_hash, 'status', NEW.status), + datetime('now')); +END; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +DROP TRIGGER IF EXISTS pr_review_threads_cdc_update; +DROP TRIGGER IF EXISTS pr_review_threads_cdc_insert; +DROP TRIGGER IF EXISTS sessions_cdc_insert; +DROP TRIGGER IF EXISTS sessions_cdc_update; +DROP TRIGGER IF EXISTS pr_cdc_insert; +DROP TRIGGER IF EXISTS pr_cdc_update; +DROP TRIGGER IF EXISTS pr_session_cdc_update; +DROP TRIGGER IF EXISTS pr_checks_cdc_insert; +DROP TRIGGER IF EXISTS pr_checks_cdc_update; + +CREATE TABLE change_log_old ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + project_id TEXT NOT NULL REFERENCES projects (id), + session_id TEXT REFERENCES sessions (id), + event_type TEXT NOT NULL + CHECK (event_type IN ( + 'session_created', + 'session_updated', + 'pr_created', + 'pr_updated', + 'pr_check_recorded', + 'pr_review_thread_added', + 'pr_review_thread_resolved' + )), + payload TEXT NOT NULL CHECK (json_valid(payload)), + created_at TIMESTAMP NOT NULL DEFAULT (datetime('now')) +); + +INSERT INTO change_log_old (seq, project_id, session_id, event_type, payload, created_at) +SELECT seq, project_id, session_id, event_type, payload, created_at +FROM change_log +WHERE event_type <> 'pr_session_changed'; + +DROP INDEX IF EXISTS idx_change_log_project; +DROP TABLE change_log; +ALTER TABLE change_log_old RENAME TO change_log; +CREATE INDEX idx_change_log_project ON change_log (project_id, seq); +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_review_threads_cdc_insert +AFTER INSERT ON pr_review_threads +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_review_thread_added', + json_object( + 'pr', NEW.pr_url, + 'thread', NEW.thread_id, + 'path', NEW.path, + 'line', NEW.line, + 'resolved', json(CASE WHEN NEW.resolved THEN 'true' ELSE 'false' END), + 'isBot', json(CASE WHEN NEW.is_bot THEN 'true' ELSE 'false' END) + ), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_review_threads_cdc_update +AFTER UPDATE ON pr_review_threads +WHEN OLD.resolved <> NEW.resolved +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_review_thread_resolved', + json_object( + 'pr', NEW.pr_url, + 'thread', NEW.thread_id, + 'path', NEW.path, + 'line', NEW.line, + 'resolved', json(CASE WHEN NEW.resolved THEN 'true' ELSE 'false' END) + ), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER sessions_cdc_insert +AFTER INSERT ON sessions +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES (NEW.project_id, NEW.id, 'session_created', + json_object('id', NEW.id, 'activity', NEW.activity_state, 'isTerminated', json(CASE WHEN NEW.is_terminated THEN 'true' ELSE 'false' END)), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER sessions_cdc_update +AFTER UPDATE ON sessions +WHEN OLD.activity_state <> NEW.activity_state + OR OLD.is_terminated <> NEW.is_terminated +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES (NEW.project_id, NEW.id, 'session_updated', + json_object('id', NEW.id, 'activity', NEW.activity_state, 'isTerminated', json(CASE WHEN NEW.is_terminated THEN 'true' ELSE 'false' END)), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_cdc_insert +AFTER INSERT ON pr +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ((SELECT project_id FROM sessions WHERE id = NEW.session_id), NEW.session_id, 'pr_created', + json_object('url', NEW.url, 'session', NEW.session_id, 'state', NEW.pr_state, + 'ci', NEW.ci_state, 'review', NEW.review_decision, 'mergeability', NEW.mergeability), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_cdc_update +AFTER UPDATE ON pr +WHEN OLD.pr_state <> NEW.pr_state + OR OLD.ci_state <> NEW.ci_state + OR OLD.review_decision <> NEW.review_decision + OR OLD.mergeability <> NEW.mergeability +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ((SELECT project_id FROM sessions WHERE id = NEW.session_id), NEW.session_id, 'pr_updated', + json_object('url', NEW.url, 'session', NEW.session_id, 'state', NEW.pr_state, + 'ci', NEW.ci_state, 'review', NEW.review_decision, 'mergeability', NEW.mergeability), + NEW.updated_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_checks_cdc_insert +AFTER INSERT ON pr_checks +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_check_recorded', + json_object('pr', NEW.pr_url, 'name', NEW.name, 'commit', NEW.commit_hash, 'status', NEW.status), + NEW.created_at); +END; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE TRIGGER pr_checks_cdc_update +AFTER UPDATE ON pr_checks +WHEN OLD.status <> NEW.status +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_check_recorded', + json_object('pr', NEW.pr_url, 'name', NEW.name, 'commit', NEW.commit_hash, 'status', NEW.status), + datetime('now')); +END; +-- +goose StatementEnd diff --git a/backend/internal/storage/sqlite/queries/pr.sql b/backend/internal/storage/sqlite/queries/pr.sql index 7ef7009cf..5501e4d38 100644 --- a/backend/internal/storage/sqlite/queries/pr.sql +++ b/backend/internal/storage/sqlite/queries/pr.sql @@ -103,6 +103,7 @@ SELECT pr.review_decision, pr.ci_state, pr.mergeability, + pr.updated_at, EXISTS ( SELECT 1 FROM pr_comment @@ -116,3 +117,19 @@ ORDER BY CASE WHEN pr.pr_state NOT IN ('merged', 'closed') THEN 0 ELSE 1 END, pr.updated_at DESC LIMIT 1; + +-- name: ClaimPRForSession :exec +INSERT INTO pr (url, session_id, number, pr_state, review_decision, ci_state, mergeability, updated_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?) +ON CONFLICT (url) DO UPDATE SET + session_id = excluded.session_id, + review_decision = excluded.review_decision, + updated_at = excluded.updated_at; + +-- name: GetPRClaimAndOwner :one +-- Returns the current owner of a PR URL plus whether that owner is +-- terminated. Used by the takeover guard inside the claim tx. +SELECT pr.session_id, sessions.is_terminated +FROM pr +JOIN sessions ON sessions.id = pr.session_id +WHERE pr.url = ?; diff --git a/backend/internal/storage/sqlite/store/pr_cdc_test.go b/backend/internal/storage/sqlite/store/pr_cdc_test.go index 971dc00ad..2f2b06d95 100644 --- a/backend/internal/storage/sqlite/store/pr_cdc_test.go +++ b/backend/internal/storage/sqlite/store/pr_cdc_test.go @@ -3,6 +3,7 @@ package store_test import ( "context" "encoding/json" + "errors" "strings" "testing" "time" @@ -148,3 +149,126 @@ func TestWritePR_PersistsScalarsChecksAndComments(t *testing.T) { t.Fatalf("comment not persisted: %+v", comments) } } + +func TestClaimPR_CreatesMovesAndGuardsActiveOwner(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedProject(t, s, "mer") + first, _ := s.CreateSession(ctx, sampleRecord("mer")) + second, _ := s.CreateSession(ctx, sampleRecord("mer")) + url := "https://github.com/acme/repo/pull/42" + pr := domain.PullRequest{URL: url, SessionID: first.ID, Number: 42, CI: domain.CIPassing, Mergeability: domain.MergeMergeable, UpdatedAt: time.Now().UTC()} + + out, err := s.ClaimPR(ctx, pr, nil, nil, nil, ports.ReviewWritePreserve, true) + if err != nil { + t.Fatalf("initial claim: %v", err) + } + if out.PreviousOwner != "" { + t.Fatalf("new claim previous owner = %q", out.PreviousOwner) + } + got, ok, err := s.GetPR(ctx, url) + if err != nil || !ok || got.SessionID != first.ID || got.Number != 42 { + t.Fatalf("claimed row = %+v ok=%v err=%v", got, ok, err) + } + + pr.SessionID = second.ID + if _, err := s.ClaimPR(ctx, pr, nil, nil, nil, ports.ReviewWritePreserve, false); !errors.Is(err, ports.ErrPRClaimedByActiveSession) { + t.Fatalf("no-takeover err = %v, want ErrPRClaimedByActiveSession", err) + } + got, _, _ = s.GetPR(ctx, url) + if got.SessionID != first.ID { + t.Fatalf("active-owner refusal moved row to %s", got.SessionID) + } + + out, err = s.ClaimPR(ctx, pr, nil, nil, nil, ports.ReviewWritePreserve, true) + if err != nil { + t.Fatalf("takeover: %v", err) + } + if out.PreviousOwner != first.ID || out.OwnerTerminated { + t.Fatalf("takeover outcome = %+v", out) + } + got, _, _ = s.GetPR(ctx, url) + if got.SessionID != second.ID { + t.Fatalf("takeover row owner = %s, want %s", got.SessionID, second.ID) + } +} + +func TestClaimPRCreatedCDCUsesClaimReviewDecision(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedProject(t, s, "mer") + rec, err := s.CreateSession(ctx, sampleRecord("mer")) + if err != nil { + t.Fatal(err) + } + url := "https://github.com/acme/repo/pull/123" + pr := domain.PullRequest{ + URL: url, + SessionID: rec.ID, + Number: 123, + Review: domain.ReviewChangesRequest, + UpdatedAt: time.Now().UTC(), + } + if _, err := s.ClaimPR(ctx, pr, nil, nil, nil, ports.ReviewWritePreserve, true); err != nil { + t.Fatal(err) + } + + events, err := s.EventsAfter(ctx, 0, 100) + if err != nil { + t.Fatal(err) + } + for _, ev := range events { + if ev.Type != cdc.EventPRCreated { + continue + } + if !strings.Contains(string(ev.Payload), `"review":"changes_requested"`) { + t.Fatalf("pr_created payload review not from claim: %s", ev.Payload) + } + return + } + t.Fatalf("no pr_created event found; events=%v", events) +} + +func TestClaimPR_TakesOverTerminatedOwnerAndEmitsSessionChangedCDC(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedProject(t, s, "mer") + first, _ := s.CreateSession(ctx, sampleRecord("mer")) + second, _ := s.CreateSession(ctx, sampleRecord("mer")) + url := "https://github.com/acme/repo/pull/99" + pr := domain.PullRequest{URL: url, SessionID: first.ID, Number: 99, CI: domain.CIPassing, UpdatedAt: time.Now().UTC()} + if _, err := s.ClaimPR(ctx, pr, nil, nil, nil, ports.ReviewWritePreserve, true); err != nil { + t.Fatal(err) + } + first.IsTerminated = true + first.UpdatedAt = time.Now().UTC().Truncate(time.Second) + if err := s.UpdateSession(ctx, first); err != nil { + t.Fatal(err) + } + + pr.SessionID = second.ID + out, err := s.ClaimPR(ctx, pr, nil, nil, nil, ports.ReviewWritePreserve, false) + if err != nil { + t.Fatalf("terminated takeover: %v", err) + } + if out.PreviousOwner != first.ID || !out.OwnerTerminated { + t.Fatalf("terminated outcome = %+v", out) + } + + events, err := s.EventsAfter(ctx, 0, 100) + if err != nil { + t.Fatal(err) + } + var changed []cdc.Event + for _, ev := range events { + if ev.Type == "pr_session_changed" { + changed = append(changed, ev) + } + } + if len(changed) != 1 { + t.Fatalf("pr_session_changed events = %d, want 1; all=%v", len(changed), events) + } + if changed[0].SessionID != string(second.ID) || !strings.Contains(string(changed[0].Payload), `"fromSession":"`+string(first.ID)+`"`) || !strings.Contains(string(changed[0].Payload), `"toSession":"`+string(second.ID)+`"`) { + t.Fatalf("bad change event: %+v", changed[0]) + } +} diff --git a/backend/internal/storage/sqlite/store/pr_facts.go b/backend/internal/storage/sqlite/store/pr_facts.go index 7bc9a8494..894bd7277 100644 --- a/backend/internal/storage/sqlite/store/pr_facts.go +++ b/backend/internal/storage/sqlite/store/pr_facts.go @@ -36,5 +36,6 @@ func prFactsFromGen(r gen.GetDisplayPRFactsBySessionRow) domain.PRFacts { Review: r.ReviewDecision, Mergeability: r.Mergeability, ReviewComments: r.ReviewComments, + UpdatedAt: r.UpdatedAt, } } diff --git a/backend/internal/storage/sqlite/store/pr_store.go b/backend/internal/storage/sqlite/store/pr_store.go index 17a91344f..1dc391939 100644 --- a/backend/internal/storage/sqlite/store/pr_store.go +++ b/backend/internal/storage/sqlite/store/pr_store.go @@ -25,6 +25,7 @@ import ( var ( _ ports.PRWriter = (*Store)(nil) _ ports.SCMWriter = (*Store)(nil) + _ ports.PRClaimer = (*Store)(nil) ) // WritePR persists a legacy PR observation — scalar facts, check runs, and the @@ -47,10 +48,46 @@ func (s *Store) WriteSCMObservation(ctx context.Context, pr domain.PullRequest, return s.writePR(ctx, pr, checks, threads, comments, reviewMode, false) } +// ClaimPR moves (or creates) a PR row to pr.SessionID and applies the live SCM +// observation in the same transaction. The session_id update is what fires the +// pr_session_changed CDC trigger added in migration 0005. +func (s *Store) ClaimPR(ctx context.Context, pr domain.PullRequest, checks []domain.PullRequestCheck, threads []domain.PullRequestReviewThread, comments []domain.PullRequestComment, reviewMode ports.ReviewWriteMode, allowActiveTakeover bool) (ports.ClaimOutcome, error) { + s.writeMu.Lock() + defer s.writeMu.Unlock() + var outcome ports.ClaimOutcome + err := s.inTx(ctx, "claim pr", func(q *gen.Queries) error { + owner, err := q.GetPRClaimAndOwner(ctx, pr.URL) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if err == nil { + outcome.PreviousOwner = owner.SessionID + outcome.OwnerTerminated = owner.IsTerminated + if owner.SessionID != pr.SessionID && !owner.IsTerminated && !allowActiveTakeover { + return ports.PRClaimedByActiveSessionError{Owner: owner.SessionID} + } + } + if err := q.ClaimPRForSession(ctx, gen.ClaimPRForSessionParams{ + URL: pr.URL, SessionID: pr.SessionID, Number: int64(pr.Number), PRState: prState(pr), + ReviewDecision: reviewOrDefault(pr.Review), CIState: ciOrDefault(pr.CI), Mergeability: mergeabilityOrDefault(pr.Mergeability), UpdatedAt: pr.UpdatedAt, + }); err != nil { + return err + } + return writePRRows(ctx, q, pr, checks, threads, comments, reviewMode, false, false) + }) + return outcome, err +} + func (s *Store) writePR(ctx context.Context, pr domain.PullRequest, checks []domain.PullRequestCheck, threads []domain.PullRequestReviewThread, comments []domain.PullRequestComment, reviewMode ports.ReviewWriteMode, replaceLegacyComments bool) error { s.writeMu.Lock() defer s.writeMu.Unlock() return s.inTx(ctx, "write pr observation", func(q *gen.Queries) error { + return writePRRows(ctx, q, pr, checks, threads, comments, reviewMode, replaceLegacyComments, true) + }) +} + +func writePRRows(ctx context.Context, q *gen.Queries, pr domain.PullRequest, checks []domain.PullRequestCheck, threads []domain.PullRequestReviewThread, comments []domain.PullRequestComment, reviewMode ports.ReviewWriteMode, replaceLegacyComments, rejectReassignment bool) error { + if rejectReassignment { existing, err := q.GetPR(ctx, pr.URL) if err != nil && !errors.Is(err, sql.ErrNoRows) { return err @@ -58,63 +95,63 @@ func (s *Store) writePR(ctx context.Context, pr domain.PullRequest, checks []dom if err == nil && existing.SessionID != pr.SessionID { return fmt.Errorf("pr %s already belongs to session %s", pr.URL, existing.SessionID) } - if replaceLegacyComments { - if err := q.UpsertLegacyPR(ctx, genLegacyPRParams(pr)); err != nil { - return err - } - } else { - if err := q.UpsertPR(ctx, genPRParams(pr)); err != nil { - return err + } + if replaceLegacyComments { + if err := q.UpsertLegacyPR(ctx, genLegacyPRParams(pr)); err != nil { + return err + } + } else { + if err := q.UpsertPR(ctx, genPRParams(pr)); err != nil { + return err + } + } + for _, c := range checks { + if err := q.UpsertPRCheck(ctx, genCheckParams(pr.URL, c)); err != nil { + return err + } + } + if reviewMode == ports.ReviewWriteReplace { + if err := q.DeletePRReviewThreads(ctx, pr.URL); err != nil { + return err + } + } + if reviewMode == ports.ReviewWriteReplace { + if err := q.DeletePRComments(ctx, pr.URL); err != nil { + return err + } + } else if replaceLegacyComments { + if err := q.DeleteLegacyPRComments(ctx, pr.URL); err != nil { + return err + } + } + if reviewMode == ports.ReviewWriteReplace || reviewMode == ports.ReviewWriteMerge { + for _, th := range threads { + if err := q.UpsertPRReviewThread(ctx, genReviewThreadParams(pr.URL, th)); err != nil { + return fmt.Errorf("review thread %q: %w", th.ThreadID, err) } } - for _, c := range checks { - if err := q.UpsertPRCheck(ctx, genCheckParams(pr.URL, c)); err != nil { - return err + } + if reviewMode == ports.ReviewWriteMerge { + for _, threadID := range reviewThreadIDs(threads, comments) { + if err := q.DeletePRCommentsByThread(ctx, gen.DeletePRCommentsByThreadParams{PRURL: pr.URL, ThreadID: threadID}); err != nil { + return fmt.Errorf("delete comments for review thread %q: %w", threadID, err) } } - if reviewMode == ports.ReviewWriteReplace { - if err := q.DeletePRReviewThreads(ctx, pr.URL); err != nil { - return err + } + if reviewMode == ports.ReviewWriteReplace || reviewMode == ports.ReviewWriteMerge { + for _, c := range comments { + if err := q.InsertPRComment(ctx, genCommentParams(pr.URL, c)); err != nil { + return fmt.Errorf("comment %q: %w", c.ID, err) } } - if reviewMode == ports.ReviewWriteReplace { - if err := q.DeletePRComments(ctx, pr.URL); err != nil { - return err - } - } else if replaceLegacyComments { - if err := q.DeleteLegacyPRComments(ctx, pr.URL); err != nil { - return err + } else if replaceLegacyComments { + for _, c := range comments { + if err := q.InsertLegacyPRComment(ctx, genLegacyCommentParams(pr.URL, c)); err != nil { + return fmt.Errorf("legacy comment %q: %w", c.ID, err) } } - if reviewMode == ports.ReviewWriteReplace || reviewMode == ports.ReviewWriteMerge { - for _, th := range threads { - if err := q.UpsertPRReviewThread(ctx, genReviewThreadParams(pr.URL, th)); err != nil { - return fmt.Errorf("review thread %q: %w", th.ThreadID, err) - } - } - } - if reviewMode == ports.ReviewWriteMerge { - for _, threadID := range reviewThreadIDs(threads, comments) { - if err := q.DeletePRCommentsByThread(ctx, gen.DeletePRCommentsByThreadParams{PRURL: pr.URL, ThreadID: threadID}); err != nil { - return fmt.Errorf("delete comments for review thread %q: %w", threadID, err) - } - } - } - if reviewMode == ports.ReviewWriteReplace || reviewMode == ports.ReviewWriteMerge { - for _, c := range comments { - if err := q.InsertPRComment(ctx, genCommentParams(pr.URL, c)); err != nil { - return fmt.Errorf("comment %q: %w", c.ID, err) - } - } - } else if replaceLegacyComments { - for _, c := range comments { - if err := q.InsertLegacyPRComment(ctx, genLegacyCommentParams(pr.URL, c)); err != nil { - return fmt.Errorf("legacy comment %q: %w", c.ID, err) - } - } - } - return nil - }) + } + return nil } func reviewThreadIDs(threads []domain.PullRequestReviewThread, comments []domain.PullRequestComment) []string { diff --git a/backend/internal/terminal/session_integration_test.go b/backend/internal/terminal/session_integration_test.go index 78b611e94..09c952d76 100644 --- a/backend/internal/terminal/session_integration_test.go +++ b/backend/internal/terminal/session_integration_test.go @@ -7,6 +7,7 @@ import ( "os" "os/exec" "path/filepath" + "strconv" "strings" "testing" "time" @@ -25,8 +26,8 @@ func TestSessionStreamsRealZellijPane(t *testing.T) { t.Skip("zellij unavailable") } - name := "ao-term-it-" + strings.ReplaceAll(t.Name(), "/", "-") - socketDir := filepath.Join(os.TempDir(), name+"-socket") + name := "ao-term-it-" + strconv.Itoa(os.Getpid()) + socketDir := filepath.Join("/tmp", name+"-socket") if err := os.MkdirAll(socketDir, 0o755); err != nil { t.Fatalf("mkdir socket dir: %v", err) } diff --git a/frontend/src/api/schema.ts b/frontend/src/api/schema.ts index f9c259214..ff92a2b42 100644 --- a/frontend/src/api/schema.ts +++ b/frontend/src/api/schema.ts @@ -162,6 +162,40 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/sessions/{sessionId}/pr": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** List pull requests owned by a session */ + get: operations["listSessionPRs"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/api/v1/sessions/{sessionId}/pr/claim": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Claim an existing pull request for a session */ + post: operations["claimSessionPR"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/v1/sessions/{sessionId}/restore": { parameters: { query?: never; @@ -231,6 +265,17 @@ export interface components { path: string; projectId?: null | string; }; + ClaimPRRequest: { + allowTakeover?: null | boolean; + pr: string; + }; + ClaimPRResponse: { + branchChanged: boolean; + ok: boolean; + prs: components["schemas"]["SessionPRFacts"][]; + sessionId: string; + takenOverFrom: string[]; + }; CleanupSessionsResponse: { cleaned: string[]; ok: boolean; @@ -254,6 +299,10 @@ export interface components { ListProjectsResponse: { projects: components["schemas"]["ProjectSummary"][]; }; + ListSessionPRsResponse: { + prs: components["schemas"]["SessionPRFacts"][]; + sessionId: string; + }; ListSessionsResponse: { sessions: components["schemas"]["Session"][]; }; @@ -351,6 +400,18 @@ export interface components { /** Format: date-time */ updatedAt: string; }; + SessionPRFacts: { + ci: string; + mergeability: string; + number: number; + review: string; + reviewComments: boolean; + /** @enum {string} */ + state: "draft" | "open" | "merged" | "closed"; + /** Format: date-time */ + updatedAt: string; + url: string; + }; SessionResponse: { session: components["schemas"]["Session"]; }; @@ -1066,6 +1127,137 @@ export interface operations { }; }; }; + listSessionPRs: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Session identifier, e.g. project-1. */ + sessionId: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description OK */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ListSessionPRsResponse"]; + }; + }; + /** @description Not Found */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Internal Server Error */ + 500: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Not Implemented */ + 501: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + }; + }; + claimSessionPR: { + parameters: { + query?: never; + header?: never; + path: { + /** @description Session identifier, e.g. project-1. */ + sessionId: string; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["ClaimPRRequest"]; + }; + }; + responses: { + /** @description OK */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ClaimPRResponse"]; + }; + }; + /** @description Bad Request */ + 400: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Not Found */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Conflict */ + 409: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Unprocessable Entity */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Not Implemented */ + 501: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + /** @description Service Unavailable */ + 503: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["APIError"]; + }; + }; + }; + }; restoreSession: { parameters: { query?: never; diff --git a/package-lock.json b/package-lock.json index 763032a0a..458892d11 100644 --- a/package-lock.json +++ b/package-lock.json @@ -6,7 +6,7 @@ "": { "name": "agent-orchestrator", "devDependencies": { - "openapi-typescript": "^7.4.4" + "openapi-typescript": "7.4.4" } }, "node_modules/@babel/code-frame": {