dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 6 Packages Projects Releases Wiki Activity

fix: clear stale PR data on task switch and add on-demand PR detection (#357)

Browse Source
This commit is contained in:
Carlos Florêncio
2026-03-03 20:14:54 +00:00
committed by GitHub
parent 8cbaf950d2
commit 730751cece
8 changed files with 418 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/internal/orchestrator/event_handlers_github.go
+59
View File
@@ -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 {
apps/backend/internal/orchestrator/event_handlers_github_test.go
+213
View File
@@ -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)
}
})
}
apps/backend/internal/orchestrator/handlers/handlers.go
+26
View File
@@ -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"`
apps/backend/pkg/websocket/actions.go
+1
View File
@@ -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
apps/web/components/task/changes-panel.tsx
+12 -3
View File
@@ -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]);
apps/web/hooks/domains/github/use-pr-commits.ts
+24 -5
View File
@@ -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<PRCommitsState>(INITIAL_STATE);
const hasParams = !!owner && !!repo && !!prNumber;
const paramsKeyRef = useRef<string>("");
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 };
}
apps/web/hooks/domains/github/use-pr-diff.ts
+24 -5
View File
@@ -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<PRDiffState>(INITIAL_STATE);
const hasParams = !!owner && !!repo && !!prNumber;
const paramsKeyRef = useRef<string>("");
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 };
}
apps/web/hooks/domains/github/use-task-pr.ts
+59 -1
View File
@@ -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]);
}