diff --git a/apps/backend/internal/orchestrator/event_handlers_github.go b/apps/backend/internal/orchestrator/event_handlers_github.go index b7b7f215cd..d877300bf3 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github.go +++ b/apps/backend/internal/orchestrator/event_handlers_github.go @@ -23,6 +23,7 @@ type GitHubService interface { EnsurePRWatch(ctx context.Context, sessionID, taskID, owner, repo, branch string) (*github.PRWatch, error) GetPRWatchBySession(ctx context.Context, sessionID string) (*github.PRWatch, error) AssociatePRWithTask(ctx context.Context, taskID string, pr *github.PR) (*github.TaskPR, error) + GetTaskPR(ctx context.Context, taskID string) (*github.TaskPR, error) RecordReviewPRTask(ctx context.Context, watchID, repoOwner, repoName string, prNumber int, prURL, taskID string) error } @@ -391,6 +392,64 @@ func (s *Service) associatePRFromPush( zap.String("branch", branch)) } +// CheckSessionPR checks if a PR exists for a session's branch and associates it +// if found. This provides an on-demand alternative to the background poller, +// allowing the frontend to trigger immediate PR detection. +func (s *Service) CheckSessionPR(ctx context.Context, taskID, sessionID string) (bool, error) { + if s.githubService == nil { + return false, nil + } + + // Check if a PR is already associated with this task + existing, err := s.githubService.GetTaskPR(ctx, taskID) + if err == nil && existing != nil { + return true, nil + } + + // Resolve the GitHub owner/repo from the task's repository + owner, repoName := s.resolveTaskRepo(ctx, taskID) + if owner == "" || repoName == "" { + return false, nil + } + + // Get the session's branch from the first worktree with a non-empty branch + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil || session == nil { + return false, nil + } + branch := "" + for _, wt := range session.Worktrees { + if wt.WorktreeBranch != "" { + branch = wt.WorktreeBranch + break + } + } + if branch == "" { + return false, nil + } + + // Ensure a PR watch exists so the background poller will keep checking + if _, watchErr := s.githubService.EnsurePRWatch(ctx, sessionID, taskID, owner, repoName, branch); watchErr != nil { + s.logger.Warn("failed to ensure PR watch during check", + zap.String("session_id", sessionID), + zap.Error(watchErr)) + } + + // Try to find the PR immediately + client := s.githubService.Client() + if client == nil { + return false, nil + } + pr, findErr := client.FindPRByBranch(ctx, owner, repoName, branch) + if findErr != nil || pr == nil { + return false, nil + } + + // Found a PR — associate it with the task + s.associatePRFromPush(ctx, sessionID, taskID, owner, repoName, branch, pr) + return true, nil +} + // subscribeGitHubEvents subscribes to GitHub-related events on the event bus. func (s *Service) subscribeGitHubEvents() { if s.eventBus == nil { diff --git a/apps/backend/internal/orchestrator/event_handlers_github_test.go b/apps/backend/internal/orchestrator/event_handlers_github_test.go index d0a282c01e..f7c2adbb2a 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_github_test.go @@ -1,11 +1,47 @@ package orchestrator import ( + "context" "testing" + "time" "github.com/kandev/kandev/internal/github" + "github.com/kandev/kandev/internal/task/models" ) +// mockGitHubService implements GitHubService for testing CheckSessionPR. +type mockGitHubService struct { + client github.Client + taskPR *github.TaskPR + taskPRErr error + ensureWatchCalls int + createWatchCalls int + associateCalls int +} + +func (m *mockGitHubService) Client() github.Client { return m.client } +func (m *mockGitHubService) GetTaskPR(_ context.Context, _ string) (*github.TaskPR, error) { + return m.taskPR, m.taskPRErr +} +func (m *mockGitHubService) EnsurePRWatch(_ context.Context, _, _, _, _, _ string) (*github.PRWatch, error) { + m.ensureWatchCalls++ + return &github.PRWatch{}, nil +} +func (m *mockGitHubService) GetPRWatchBySession(_ context.Context, _ string) (*github.PRWatch, error) { + return nil, nil +} +func (m *mockGitHubService) CreatePRWatch(_ context.Context, _, _, _, _ string, _ int, _ string) (*github.PRWatch, error) { + m.createWatchCalls++ + return &github.PRWatch{}, nil +} +func (m *mockGitHubService) AssociatePRWithTask(_ context.Context, _ string, _ *github.PR) (*github.TaskPR, error) { + m.associateCalls++ + return &github.TaskPR{}, nil +} +func (m *mockGitHubService) RecordReviewPRTask(context.Context, string, string, string, int, string, string) error { + return nil +} + func TestInterpolateReviewPrompt(t *testing.T) { pr := &github.PR{ Number: 42, @@ -54,3 +90,180 @@ func TestInterpolateReviewPrompt(t *testing.T) { }) } } + +func TestCheckSessionPR(t *testing.T) { + ctx := context.Background() + now := time.Now().UTC() + + testPR := &github.PR{ + Number: 10, + Title: "Test PR", + HTMLURL: "https://github.com/myorg/myrepo/pull/10", + AuthorLogin: "alice", + RepoOwner: "myorg", + RepoName: "myrepo", + HeadBranch: "feature-branch", + BaseBranch: "main", + } + + // seedWithRepo creates task + session + repository + task-repository + worktree + // so that resolveTaskRepo and GetTaskSession succeed. + seedWithRepo := func(t *testing.T, branch string) *Service { + t.Helper() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + // Create a GitHub-backed repository record + repoObj := &models.Repository{ + ID: "repo1", + WorkspaceID: "ws1", + Name: "myrepo", + SourceType: "provider", + Provider: "github", + ProviderOwner: "myorg", + ProviderName: "myrepo", + CreatedAt: now, + UpdatedAt: now, + } + if err := repo.CreateRepository(ctx, repoObj); err != nil { + t.Fatalf("failed to create repository: %v", err) + } + + // Link task to repository + taskRepo := &models.TaskRepository{ + ID: "tr1", + TaskID: "t1", + RepositoryID: "repo1", + Position: 0, + CreatedAt: now, + UpdatedAt: now, + } + if err := repo.CreateTaskRepository(ctx, taskRepo); err != nil { + t.Fatalf("failed to create task repository: %v", err) + } + + // Add worktree with branch to the session + wt := &models.TaskSessionWorktree{ + ID: "wt1", + SessionID: "s1", + WorktreeID: "wtree1", + RepositoryID: "repo1", + WorktreeBranch: branch, + CreatedAt: now, + } + if err := repo.CreateTaskSessionWorktree(ctx, wt); err != nil { + t.Fatalf("failed to create worktree: %v", err) + } + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + return svc + } + + t.Run("returns false when github service is nil", func(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + // githubService is nil by default + + found, err := svc.CheckSessionPR(ctx, "t1", "s1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if found { + t.Error("expected found=false when githubService is nil") + } + }) + + t.Run("returns true when PR already associated", func(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + ghSvc := &mockGitHubService{ + taskPR: &github.TaskPR{ID: "tpr1", TaskID: "t1", PRNumber: 10}, + } + svc.SetGitHubService(ghSvc) + + found, err := svc.CheckSessionPR(ctx, "t1", "s1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !found { + t.Error("expected found=true when PR already exists") + } + if ghSvc.ensureWatchCalls != 0 { + t.Errorf("expected no EnsurePRWatch calls, got %d", ghSvc.ensureWatchCalls) + } + }) + + t.Run("returns false when task has no repository", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + ghSvc := &mockGitHubService{taskPRErr: nil, taskPR: nil} + svc.SetGitHubService(ghSvc) + + found, err := svc.CheckSessionPR(ctx, "t1", "s1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if found { + t.Error("expected found=false when task has no repository") + } + }) + + t.Run("returns false when session has no worktree branch", func(t *testing.T) { + svc := seedWithRepo(t, "") // empty branch + ghSvc := &mockGitHubService{taskPRErr: nil, taskPR: nil} + svc.SetGitHubService(ghSvc) + + found, err := svc.CheckSessionPR(ctx, "t1", "s1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if found { + t.Error("expected found=false when no branch on worktree") + } + }) + + t.Run("returns false when no PR exists on branch", func(t *testing.T) { + svc := seedWithRepo(t, "feature-branch") + mockClient := github.NewMockClient() + // No PR added to mock client + ghSvc := &mockGitHubService{client: mockClient} + svc.SetGitHubService(ghSvc) + + found, err := svc.CheckSessionPR(ctx, "t1", "s1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if found { + t.Error("expected found=false when no PR on branch") + } + if ghSvc.ensureWatchCalls != 1 { + t.Errorf("expected 1 EnsurePRWatch call, got %d", ghSvc.ensureWatchCalls) + } + }) + + t.Run("finds PR and associates it", func(t *testing.T) { + svc := seedWithRepo(t, "feature-branch") + mockClient := github.NewMockClient() + mockClient.AddPR(testPR) + ghSvc := &mockGitHubService{client: mockClient} + svc.SetGitHubService(ghSvc) + + found, err := svc.CheckSessionPR(ctx, "t1", "s1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !found { + t.Error("expected found=true when PR exists on branch") + } + if ghSvc.ensureWatchCalls != 1 { + t.Errorf("expected 1 EnsurePRWatch call, got %d", ghSvc.ensureWatchCalls) + } + if ghSvc.createWatchCalls != 1 { + t.Errorf("expected 1 CreatePRWatch call (from associatePRFromPush), got %d", ghSvc.createWatchCalls) + } + if ghSvc.associateCalls != 1 { + t.Errorf("expected 1 AssociatePRWithTask call, got %d", ghSvc.associateCalls) + } + }) +} diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go index 4472522578..14f5120cb8 100644 --- a/apps/backend/internal/orchestrator/handlers/handlers.go +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -40,6 +40,7 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionTaskSessionPrepare, h.wsPrepareTaskSession) d.RegisterFunc(ws.ActionAgentCancel, h.wsCancelAgent) d.RegisterFunc(ws.ActionSessionLaunch, h.wsLaunchSession) + d.RegisterFunc(ws.ActionGitHubCheckSessionPR, h.wsCheckSessionPR) } // WS handlers @@ -371,6 +372,31 @@ func (h *Handlers) wsRespondToPermission(ctx context.Context, msg *ws.Message) ( return ws.NewResponse(msg.ID, msg.Action, resp) } +type wsCheckSessionPRRequest struct { + TaskID string `json:"task_id"` + SessionID string `json:"session_id"` +} + +func (h *Handlers) wsCheckSessionPR(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsCheckSessionPRRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" || req.SessionID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id and session_id are required", nil) + } + + found, err := h.service.CheckSessionPR(ctx, req.TaskID, req.SessionID) + if err != nil { + h.logger.Error("failed to check session PR", + zap.String("task_id", req.TaskID), + zap.String("session_id", req.SessionID), + zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to check session PR", nil) + } + return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"found": found}) +} + type wsGetTaskSessionStatusRequest struct { TaskID string `json:"task_id"` TaskSessionID string `json:"session_id"` diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 996c77f569..9f159e5449 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -306,6 +306,7 @@ const ( ActionGitHubPRFeedbackNotify = "github.pr_feedback.notify" // Notification ActionGitHubNewReviewPRNotify = "github.new_review_pr.notify" // Notification ActionGitHubStats = "github.stats" + ActionGitHubCheckSessionPR = "github.check_session_pr" ) // Error codes diff --git a/apps/web/components/task/changes-panel.tsx b/apps/web/components/task/changes-panel.tsx index 5f468d24ed..72bec7822f 100644 --- a/apps/web/components/task/changes-panel.tsx +++ b/apps/web/components/task/changes-panel.tsx @@ -27,7 +27,7 @@ import { } from "./changes-panel-timeline"; import type { PRChangedFile } from "./changes-panel-timeline"; import { useChangesGitHandlers, useChangesDialogHandlers } from "./changes-panel-hooks"; -import { useActiveTaskPR } from "@/hooks/domains/github/use-task-pr"; +import { useActiveTaskPR, useTaskPRDetection } from "@/hooks/domains/github/use-task-pr"; import { usePRDiff } from "@/hooks/domains/github/use-pr-diff"; import { usePRCommits } from "@/hooks/domains/github/use-pr-commits"; import type { PRDiffFile } from "@/lib/types/github"; @@ -177,6 +177,7 @@ function getBaseBranchDisplay(baseBranch: string | undefined): string { } function useChangesPanelStoreData() { + const activeTaskId = useAppStore((state) => state.tasks.activeTaskId); const activeSessionId = useAppStore((state) => state.tasks.activeSessionId); const taskTitle = useAppStore((state) => { if (!state.tasks.activeTaskId) return undefined; @@ -190,7 +191,7 @@ function useChangesPanelStoreData() { if (!taskId) return undefined; return state.taskPRs.byTaskId[taskId]?.pr_url ?? undefined; }); - return { activeSessionId, taskTitle, baseBranch, existingPrUrl }; + return { activeTaskId, activeSessionId, taskTitle, baseBranch, existingPrUrl }; } type ChangesPanelBodyProps = { @@ -484,13 +485,21 @@ const ChangesPanel = memo(function ChangesPanel({ onOpenReview, }: ChangesPanelProps) { const isArchived = useIsTaskArchived(); - const { activeSessionId, taskTitle, baseBranch, existingPrUrl } = useChangesPanelStoreData(); + const { activeTaskId, activeSessionId, taskTitle, baseBranch, existingPrUrl } = + useChangesPanelStoreData(); const git = useSessionGit(activeSessionId); const { toast } = useToast(); const { reviews } = useSessionFileReviews(activeSessionId); const { prDiffFiles, prCommitsList, hasPRFiles, hasPRCommits, prFiles } = useChangesPanelPRData(); + // Periodically check for a PR when the session has a branch but no PR yet + useTaskPRDetection( + isArchived ? null : (activeTaskId ?? null), + isArchived ? null : (activeSessionId ?? null), + isArchived ? null : (git.branch ?? null), + ); + const baseBranchDisplay = useMemo(() => getBaseBranchDisplay(baseBranch), [baseBranch]); const unstagedFiles = useMemo(() => mapToChangedFiles(git.unstagedFiles), [git.unstagedFiles]); const stagedFiles = useMemo(() => mapToChangedFiles(git.stagedFiles), [git.stagedFiles]); diff --git a/apps/web/hooks/domains/github/use-pr-commits.ts b/apps/web/hooks/domains/github/use-pr-commits.ts index 2b22a08fcb..e996f381b7 100644 --- a/apps/web/hooks/domains/github/use-pr-commits.ts +++ b/apps/web/hooks/domains/github/use-pr-commits.ts @@ -1,6 +1,6 @@ "use client"; -import { useEffect, useCallback, useState } from "react"; +import { useEffect, useCallback, useState, useRef } from "react"; import { getWebSocketClient } from "@/lib/ws/connection"; import type { PRCommitInfo } from "@/lib/types/github"; @@ -48,16 +48,35 @@ async function fetchPRCommits( */ export function usePRCommits(owner: string | null, repo: string | null, prNumber: number | null) { const [state, setState] = useState(INITIAL_STATE); + const hasParams = !!owner && !!repo && !!prNumber; + const paramsKeyRef = useRef(""); + const requestIdRef = useRef(0); const refresh = useCallback(() => { if (!owner || !repo || !prNumber) return; - void fetchPRCommits(owner, repo, prNumber, setState); + const requestId = ++requestIdRef.current; + void fetchPRCommits(owner, repo, prNumber, (next) => { + if (requestId !== requestIdRef.current) return; + setState(next); + }); }, [owner, repo, prNumber]); useEffect(() => { - if (!owner || !repo || !prNumber) return; - void fetchPRCommits(owner, repo, prNumber, setState); - }, [owner, repo, prNumber]); + const key = hasParams ? `${owner}/${repo}/${prNumber}` : ""; + if (key === paramsKeyRef.current) return; + paramsKeyRef.current = key; + if (!owner || !repo || !prNumber) { + requestIdRef.current++; // invalidate in-flight responses + return; + } + const requestId = ++requestIdRef.current; + void fetchPRCommits(owner, repo, prNumber, (next) => { + if (requestId !== requestIdRef.current) return; + setState(next); + }); + }, [owner, repo, prNumber, hasParams]); + // Return initial state when params are null to clear stale data + if (!hasParams) return { ...INITIAL_STATE, refresh }; return { ...state, refresh }; } diff --git a/apps/web/hooks/domains/github/use-pr-diff.ts b/apps/web/hooks/domains/github/use-pr-diff.ts index 9bd4f9679f..33c223a246 100644 --- a/apps/web/hooks/domains/github/use-pr-diff.ts +++ b/apps/web/hooks/domains/github/use-pr-diff.ts @@ -1,6 +1,6 @@ "use client"; -import { useEffect, useCallback, useState } from "react"; +import { useEffect, useCallback, useState, useRef } from "react"; import { getWebSocketClient } from "@/lib/ws/connection"; import type { PRDiffFile } from "@/lib/types/github"; @@ -48,16 +48,35 @@ async function fetchPRFiles( */ export function usePRDiff(owner: string | null, repo: string | null, prNumber: number | null) { const [state, setState] = useState(INITIAL_STATE); + const hasParams = !!owner && !!repo && !!prNumber; + const paramsKeyRef = useRef(""); + const requestIdRef = useRef(0); const refresh = useCallback(() => { if (!owner || !repo || !prNumber) return; - void fetchPRFiles(owner, repo, prNumber, setState); + const requestId = ++requestIdRef.current; + void fetchPRFiles(owner, repo, prNumber, (next) => { + if (requestId !== requestIdRef.current) return; + setState(next); + }); }, [owner, repo, prNumber]); useEffect(() => { - if (!owner || !repo || !prNumber) return; - void fetchPRFiles(owner, repo, prNumber, setState); - }, [owner, repo, prNumber]); + const key = hasParams ? `${owner}/${repo}/${prNumber}` : ""; + if (key === paramsKeyRef.current) return; + paramsKeyRef.current = key; + if (!owner || !repo || !prNumber) { + requestIdRef.current++; // invalidate in-flight responses + return; + } + const requestId = ++requestIdRef.current; + void fetchPRFiles(owner, repo, prNumber, (next) => { + if (requestId !== requestIdRef.current) return; + setState(next); + }); + }, [owner, repo, prNumber, hasParams]); + // Return initial state when params are null to clear stale data + if (!hasParams) return { ...INITIAL_STATE, refresh }; return { ...state, refresh }; } diff --git a/apps/web/hooks/domains/github/use-task-pr.ts b/apps/web/hooks/domains/github/use-task-pr.ts index a67802dfd9..1ca384e0ca 100644 --- a/apps/web/hooks/domains/github/use-task-pr.ts +++ b/apps/web/hooks/domains/github/use-task-pr.ts @@ -1,7 +1,8 @@ "use client"; -import { useEffect, useCallback } from "react"; +import { useEffect, useCallback, useRef } from "react"; import { listTaskPRs, getTaskPR } from "@/lib/api/domains/github-api"; +import { getWebSocketClient } from "@/lib/ws/connection"; import { useAppStore } from "@/components/state-provider"; import type { TaskPR } from "@/lib/types/github"; @@ -62,3 +63,60 @@ export function useActiveTaskPR(): TaskPR | null { return taskId ? (s.taskPRs.byTaskId[taskId] ?? null) : null; }); } + +const PR_DETECTION_INTERVAL = 30_000; // 30 seconds + +/** + * Periodically checks for a PR on the session's branch when no PR is + * associated with the task yet. Triggers a backend check that ensures + * a PR watch exists and tries to find the PR immediately. + */ +export function useTaskPRDetection( + taskId: string | null, + sessionId: string | null, + branch: string | null, +) { + const pr = useAppStore((s) => (taskId ? (s.taskPRs.byTaskId[taskId] ?? null) : null)); + const setTaskPR = useAppStore((s) => s.setTaskPR); + const checkingRef = useRef(false); + + useEffect(() => { + if (!taskId || !sessionId || !branch || pr) return; + + const check = () => { + if (checkingRef.current) return; + checkingRef.current = true; + + const client = getWebSocketClient(); + if (!client) { + checkingRef.current = false; + return; + } + + client + .request<{ found?: boolean }>("github.check_session_pr", { + task_id: taskId, + session_id: sessionId, + }) + .then((resp) => { + if (resp?.found) { + // PR was found and associated — fetch the full TaskPR record + getTaskPR(taskId, { cache: "no-store" }) + .then((taskPr) => { + if (taskPr) setTaskPR(taskId, taskPr); + }) + .catch(() => {}); + } + }) + .catch(() => {}) + .finally(() => { + checkingRef.current = false; + }); + }; + + // Check immediately, then on interval + check(); + const interval = setInterval(check, PR_DETECTION_INTERVAL); + return () => clearInterval(interval); + }, [taskId, sessionId, branch, pr, setTaskPR]); +}