feat(KB-004): add GitHub PR creation and comment monitoring
- Add PR fields (prInfo, prNumber, prUrl) to Task type and TaskStore with updatePrInfo method - Add PR Management API endpoints: create, status, refresh with per-repo rate limiting - Add GitHubClient for PR creation and status fetching with comment support - Add PrSection component for PR status display and actions in dashboard - Add PR comment monitoring engine with PrMonitor and PrCommentHandler - Integrate PR monitoring into scheduler for automatic PR tracking - Add comprehensive tests for PR features in all packages
This commit is contained in:
gsxdsm
37
packages/engine/src/github.ts
Normal file
37
packages/engine/src/github.ts
Normal file
@@ -0,0 +1,37 @@
|
||||
import { execFileSync } from "node:child_process";
|
||||
|
||||
/**
|
||||
* Extract owner/repo from a GitHub remote URL or return null if not a GitHub remote.
|
||||
*/
|
||||
export function parseGitHubRemote(remoteUrl: string): { owner: string; repo: string } | null {
|
||||
// Handle HTTPS: https://github.com/owner/repo.git or https://github.com/owner/repo
|
||||
const httpsMatch = remoteUrl.match(/github\.com\/([^\/]+)\/([^\/\.]+)(?:\.git)?$/);
|
||||
if (httpsMatch) {
|
||||
return { owner: httpsMatch[1], repo: httpsMatch[2] };
|
||||
}
|
||||
|
||||
// Handle SSH: git@github.com:owner/repo.git or git@github.com:owner/repo
|
||||
const sshMatch = remoteUrl.match(/github\.com:([^\/]+)\/([^\/\.]+)(?:\.git)?$/);
|
||||
if (sshMatch) {
|
||||
return { owner: sshMatch[1], repo: sshMatch[2] };
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current GitHub remote owner/repo from the git config.
|
||||
*/
|
||||
export function getCurrentGitHubRepo(cwd: string): { owner: string; repo: string } | null {
|
||||
try {
|
||||
const remoteUrl = execFileSync("git", ["remote", "get-url", "origin"], {
|
||||
cwd,
|
||||
encoding: "utf-8",
|
||||
stdio: ["pipe", "pipe", "ignore"],
|
||||
}).trim();
|
||||
|
||||
return parseGitHubRemote(remoteUrl);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -9,3 +9,5 @@ export { createKbAgent, type AgentOptions, type AgentResult } from "./pi.js";
|
||||
export { WorktreePool, scanIdleWorktrees, cleanupOrphanedWorktrees } from "./worktree-pool.js";
|
||||
export { createLogger, type Logger } from "./logger.js";
|
||||
export { isUsageLimitError, UsageLimitPauser } from "./usage-limit-detector.js";
|
||||
export { PrMonitor, type PrComment, type TrackedPr, type OnNewCommentsCallback } from "./pr-monitor.js";
|
||||
export { PrCommentHandler } from "./pr-comment-handler.js";
|
||||
|
||||
@@ -61,3 +61,6 @@ export const worktreePoolLog = createLogger("worktree-pool");
|
||||
|
||||
/** Logger for the review subsystem. */
|
||||
export const reviewerLog = createLogger("reviewer");
|
||||
|
||||
/** Logger for the PR monitor subsystem. */
|
||||
export const prMonitorLog = createLogger("pr-monitor");
|
||||
|
||||
237
packages/engine/src/pr-comment-handler.test.ts
Normal file
237
packages/engine/src/pr-comment-handler.test.ts
Normal file
@@ -0,0 +1,237 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
import { PrCommentHandler } from "./pr-comment-handler.js";
|
||||
import type { TaskStore } from "@kb/core";
|
||||
|
||||
const mockStore = {
|
||||
addSteeringComment: vi.fn(),
|
||||
createTask: vi.fn(),
|
||||
} as unknown as TaskStore;
|
||||
|
||||
describe("PrCommentHandler", () => {
|
||||
let handler: PrCommentHandler;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
handler = new PrCommentHandler(mockStore);
|
||||
});
|
||||
|
||||
const mockPrInfo = {
|
||||
url: "https://github.com/owner/repo/pull/42",
|
||||
number: 42,
|
||||
status: "open" as const,
|
||||
title: "Test PR",
|
||||
headBranch: "kb/kb-001",
|
||||
baseBranch: "main",
|
||||
commentCount: 0,
|
||||
};
|
||||
|
||||
describe("isNonActionable", () => {
|
||||
it.each([
|
||||
"LGTM",
|
||||
"lgtm",
|
||||
"Looks good",
|
||||
"Looks good to me",
|
||||
"Thanks",
|
||||
"Thank you",
|
||||
"Nice",
|
||||
"Great work",
|
||||
"👍",
|
||||
"✅",
|
||||
])("filters out non-actionable comment: %s", async (body) => {
|
||||
await handler.handleNewComments("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body,
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
]);
|
||||
|
||||
expect(mockStore.addSteeringComment).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("isActionable", () => {
|
||||
it.each([
|
||||
{ body: "Please fix the indentation", keyword: "fix" },
|
||||
{ body: "Should change the variable name", keyword: "change" },
|
||||
{ body: "Update the documentation", keyword: "update" },
|
||||
{ body: "Remove the unused import", keyword: "remove" },
|
||||
{ body: "Add error handling", keyword: "add" },
|
||||
{ body: "You should refactor this", keyword: "should" },
|
||||
{ body: "Needs to handle edge cases", keyword: "needs to" },
|
||||
{ body: "Consider using a different approach", keyword: "consider" },
|
||||
{ body: "I suggest renaming this", keyword: "suggest" },
|
||||
{ body: "Recommend adding tests", keyword: "recommend" },
|
||||
])("creates steering comment for actionable feedback containing '$keyword': $body", async ({ body }) => {
|
||||
await handler.handleNewComments("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body,
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
]);
|
||||
|
||||
expect(mockStore.addSteeringComment).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("code suggestions", () => {
|
||||
it("creates steering comment for comments with code blocks", async () => {
|
||||
await handler.handleNewComments("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body: "```typescript\nconst x = 1;\n```",
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
]);
|
||||
|
||||
expect(mockStore.addSteeringComment).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("creates steering comment for inline code suggestions", async () => {
|
||||
await handler.handleNewComments("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body: "Use `const` instead of `let`",
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
]);
|
||||
|
||||
expect(mockStore.addSteeringComment).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("steering comment content", () => {
|
||||
it("includes PR info and comment details", async () => {
|
||||
await handler.handleNewComments("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body: "Please fix the bug",
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
]);
|
||||
|
||||
const call = mockStore.addSteeringComment.mock.calls[0];
|
||||
const text = call[1] as string;
|
||||
|
||||
expect(text).toContain("PR Review Feedback");
|
||||
expect(text).toContain("@reviewer");
|
||||
expect(text).toContain("#42");
|
||||
expect(text).toContain("open");
|
||||
expect(text).toContain("Please fix the bug");
|
||||
expect(text).toContain("View on GitHub");
|
||||
});
|
||||
|
||||
it("truncates long comments", async () => {
|
||||
const longBody = "Please fix this issue: " + "a".repeat(1000);
|
||||
|
||||
await handler.handleNewComments("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body: longBody,
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
]);
|
||||
|
||||
const call = mockStore.addSteeringComment.mock.calls[0];
|
||||
const text = call[1] as string;
|
||||
|
||||
expect(text.length).toBeLessThan(longBody.length);
|
||||
expect(text).toContain("...");
|
||||
});
|
||||
|
||||
it("marks as agent-authored", async () => {
|
||||
await handler.handleNewComments("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body: "Please fix this",
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
]);
|
||||
|
||||
expect(mockStore.addSteeringComment).toHaveBeenCalledWith(
|
||||
"KB-001",
|
||||
expect.any(String),
|
||||
"agent"
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("createFollowUpTask", () => {
|
||||
it("creates follow-up task for unaddressed feedback", async () => {
|
||||
await handler.createFollowUpTask("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body: "This needs fixing",
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
]);
|
||||
|
||||
expect(mockStore.createTask).toHaveBeenCalledWith({
|
||||
title: "Follow-up: Address PR #42 feedback",
|
||||
description: expect.stringContaining("KB-001"),
|
||||
column: "triage",
|
||||
dependencies: ["KB-001"],
|
||||
});
|
||||
});
|
||||
|
||||
it("does nothing when no unaddressed comments", async () => {
|
||||
await handler.createFollowUpTask("KB-001", mockPrInfo, []);
|
||||
|
||||
expect(mockStore.createTask).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("summarizes multiple comments", async () => {
|
||||
await handler.createFollowUpTask("KB-001", mockPrInfo, [
|
||||
{
|
||||
id: 1,
|
||||
body: "First issue to fix",
|
||||
user: { login: "reviewer1" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-1",
|
||||
},
|
||||
{
|
||||
id: 2,
|
||||
body: "Second issue",
|
||||
user: { login: "reviewer2" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-2",
|
||||
},
|
||||
]);
|
||||
|
||||
const call = mockStore.createTask.mock.calls[0];
|
||||
const description = call[0].description as string;
|
||||
|
||||
expect(description).toContain("@reviewer1");
|
||||
expect(description).toContain("@reviewer2");
|
||||
expect(description).toContain("First issue");
|
||||
expect(description).toContain("Second issue");
|
||||
});
|
||||
});
|
||||
});
|
||||
187
packages/engine/src/pr-comment-handler.ts
Normal file
187
packages/engine/src/pr-comment-handler.ts
Normal file
@@ -0,0 +1,187 @@
|
||||
import type { TaskStore } from "@kb/core";
|
||||
import type { PrInfo } from "@kb/core";
|
||||
import { prMonitorLog } from "./logger.js";
|
||||
|
||||
interface PrComment {
|
||||
id: number;
|
||||
body: string;
|
||||
user: { login: string };
|
||||
created_at: string;
|
||||
updated_at: string;
|
||||
html_url: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Analyzes PR comments for actionable feedback and creates
|
||||
* steering comments or follow-up tasks.
|
||||
*/
|
||||
export class PrCommentHandler {
|
||||
// Keywords that suggest actionable feedback
|
||||
private readonly ACTION_KEYWORDS = [
|
||||
"fix",
|
||||
"change",
|
||||
"update",
|
||||
"remove",
|
||||
"add",
|
||||
"should",
|
||||
"need to",
|
||||
"needs to",
|
||||
"please",
|
||||
"consider",
|
||||
"suggest",
|
||||
"recommend",
|
||||
];
|
||||
|
||||
// Non-actionable patterns to filter out
|
||||
private readonly NON_ACTIONABLE_PATTERNS = [
|
||||
/^\s*lgtm\s*$/i,
|
||||
/^\s*looks? good\s*$/i,
|
||||
/^\s*thanks?\s*$/i,
|
||||
/^\s*thank you\s*$/i,
|
||||
/^\s*nice\s*$/i,
|
||||
/^\s*great\s*$/i,
|
||||
/^\s*awesome\s*$/i,
|
||||
/^\s*👍\s*$/,
|
||||
/^\s*✅\s*$/,
|
||||
];
|
||||
|
||||
constructor(private store: TaskStore) {}
|
||||
|
||||
/**
|
||||
* Process new PR comments for a task.
|
||||
* Called by PrMonitor when new comments are detected.
|
||||
*/
|
||||
async handleNewComments(
|
||||
taskId: string,
|
||||
prInfo: PrInfo,
|
||||
comments: PrComment[]
|
||||
): Promise<void> {
|
||||
for (const comment of comments) {
|
||||
await this.processComment(taskId, prInfo, comment);
|
||||
}
|
||||
}
|
||||
|
||||
private async processComment(
|
||||
taskId: string,
|
||||
prInfo: PrInfo,
|
||||
comment: PrComment
|
||||
): Promise<void> {
|
||||
// Skip non-actionable comments
|
||||
if (this.isNonActionable(comment.body)) {
|
||||
prMonitorLog.log(`Skipping non-actionable comment #${comment.id}`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Check if comment contains actionable feedback
|
||||
const isActionable = this.isActionable(comment.body);
|
||||
const hasCodeSuggestions = this.hasCodeBlock(comment.body);
|
||||
|
||||
if (!isActionable && !hasCodeSuggestions) {
|
||||
prMonitorLog.log(`Comment #${comment.id} does not contain actionable feedback`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Build steering comment text
|
||||
const text = this.buildSteeringText(prInfo, comment, hasCodeSuggestions);
|
||||
|
||||
try {
|
||||
await this.store.addSteeringComment(taskId, text, "agent");
|
||||
prMonitorLog.log(`Added steering comment for PR review #${comment.id}`);
|
||||
} catch (err) {
|
||||
prMonitorLog.error(`Failed to add steering comment for ${taskId}:`, err);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a comment is non-actionable (LGTM, thanks, etc.)
|
||||
*/
|
||||
private isNonActionable(body: string): boolean {
|
||||
const trimmed = body.trim();
|
||||
return this.NON_ACTIONABLE_PATTERNS.some((pattern) => pattern.test(trimmed));
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a comment contains actionable feedback keywords.
|
||||
*/
|
||||
private isActionable(body: string): boolean {
|
||||
const lowerBody = body.toLowerCase();
|
||||
return this.ACTION_KEYWORDS.some((keyword) => lowerBody.includes(keyword));
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a comment contains code blocks suggesting changes.
|
||||
*/
|
||||
private hasCodeBlock(body: string): boolean {
|
||||
// Look for code blocks (``` or `code`)
|
||||
return /```[\s\S]*?```/.test(body) || /`[^`]+`/.test(body);
|
||||
}
|
||||
|
||||
/**
|
||||
* Build steering comment text from PR review comment.
|
||||
*/
|
||||
private buildSteeringText(
|
||||
prInfo: PrInfo,
|
||||
comment: PrComment,
|
||||
hasCodeSuggestions: boolean
|
||||
): string {
|
||||
const lines: string[] = [];
|
||||
|
||||
lines.push(`**PR Review Feedback** from @${comment.user.login}`);
|
||||
lines.push(`**PR:** #${prInfo.number} (${prInfo.status})`);
|
||||
lines.push("");
|
||||
|
||||
// Truncate comment body if too long
|
||||
const maxBodyLength = 500;
|
||||
let body = comment.body.trim();
|
||||
if (body.length > maxBodyLength) {
|
||||
body = body.slice(0, maxBodyLength) + "...";
|
||||
}
|
||||
lines.push(body);
|
||||
lines.push("");
|
||||
|
||||
if (hasCodeSuggestions) {
|
||||
lines.push("💡 This comment contains code suggestions. Please review and apply if appropriate.");
|
||||
}
|
||||
|
||||
lines.push(`[View on GitHub](${comment.html_url})`);
|
||||
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a follow-up task when a PR is closed with unaddressed feedback.
|
||||
* This is called when a PR is merged or closed.
|
||||
*/
|
||||
async createFollowUpTask(
|
||||
originalTaskId: string,
|
||||
prInfo: PrInfo,
|
||||
unaddressedComments: PrComment[]
|
||||
): Promise<void> {
|
||||
if (unaddressedComments.length === 0) return;
|
||||
|
||||
const summary = unaddressedComments
|
||||
.map((c) => `- @${c.user.login}: ${c.body.slice(0, 100).trim()}${c.body.length > 100 ? "..." : ""}`)
|
||||
.join("\n");
|
||||
|
||||
const description = `Follow-up for ${originalTaskId}
|
||||
|
||||
PR #${prInfo.number} was ${prInfo.status} with unaddressed feedback:
|
||||
|
||||
${summary}
|
||||
|
||||
Please review the PR comments and address any remaining issues.`;
|
||||
|
||||
try {
|
||||
const task = await this.store.createTask({
|
||||
title: `Follow-up: Address PR #${prInfo.number} feedback`,
|
||||
description,
|
||||
column: "triage",
|
||||
dependencies: [originalTaskId],
|
||||
});
|
||||
|
||||
prMonitorLog.log(`Created follow-up task ${task.id} for PR #${prInfo.number}`);
|
||||
} catch (err) {
|
||||
prMonitorLog.error(`Failed to create follow-up task:`, err);
|
||||
}
|
||||
}
|
||||
}
|
||||
143
packages/engine/src/pr-monitor.test.ts
Normal file
143
packages/engine/src/pr-monitor.test.ts
Normal file
@@ -0,0 +1,143 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import { PrMonitor, type PrComment } from "./pr-monitor.js";
|
||||
|
||||
describe("PrMonitor", () => {
|
||||
let monitor: PrMonitor;
|
||||
const mockFetch = vi.fn();
|
||||
const originalFetch = globalThis.fetch;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
monitor = new PrMonitor({ getGitHubToken: () => "test-token" });
|
||||
globalThis.fetch = mockFetch;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
monitor.stopAll();
|
||||
globalThis.fetch = originalFetch;
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
const mockPrInfo = {
|
||||
url: "https://github.com/owner/repo/pull/42",
|
||||
number: 42,
|
||||
status: "open" as const,
|
||||
title: "Test PR",
|
||||
headBranch: "kb/kb-001",
|
||||
baseBranch: "main",
|
||||
commentCount: 0,
|
||||
};
|
||||
|
||||
const mockComment: PrComment = {
|
||||
id: 123,
|
||||
body: "Test comment",
|
||||
user: { login: "reviewer" },
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
html_url: "https://github.com/owner/repo/pull/42#issuecomment-123",
|
||||
};
|
||||
|
||||
describe("startMonitoring", () => {
|
||||
it("starts monitoring a PR", () => {
|
||||
monitor.startMonitoring("KB-001", "owner", "repo", mockPrInfo);
|
||||
|
||||
const tracked = monitor.getTrackedPrs();
|
||||
expect(tracked.has("KB-001")).toBe(true);
|
||||
expect(tracked.get("KB-001")?.prInfo.number).toBe(42);
|
||||
});
|
||||
|
||||
it("replaces existing monitoring for same task", () => {
|
||||
monitor.startMonitoring("KB-001", "owner", "repo", mockPrInfo);
|
||||
const newPrInfo = { ...mockPrInfo, number: 43 };
|
||||
monitor.startMonitoring("KB-001", "owner", "repo", newPrInfo);
|
||||
|
||||
const tracked = monitor.getTrackedPrs();
|
||||
expect(tracked.get("KB-001")?.prInfo.number).toBe(43);
|
||||
});
|
||||
});
|
||||
|
||||
describe("stopMonitoring", () => {
|
||||
it("stops monitoring a task", () => {
|
||||
monitor.startMonitoring("KB-001", "owner", "repo", mockPrInfo);
|
||||
monitor.stopMonitoring("KB-001");
|
||||
|
||||
const tracked = monitor.getTrackedPrs();
|
||||
expect(tracked.has("KB-001")).toBe(false);
|
||||
});
|
||||
|
||||
it("does nothing for untracked task", () => {
|
||||
expect(() => monitor.stopMonitoring("KB-999")).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
describe("stopAll", () => {
|
||||
it("stops all monitoring", () => {
|
||||
monitor.startMonitoring("KB-001", "owner", "repo", mockPrInfo);
|
||||
monitor.startMonitoring("KB-002", "owner", "repo", mockPrInfo);
|
||||
|
||||
monitor.stopAll();
|
||||
|
||||
const tracked = monitor.getTrackedPrs();
|
||||
expect(tracked.size).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("polling", () => {
|
||||
it("polls for comments on interval", async () => {
|
||||
mockFetch.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
json: () => Promise.resolve([]),
|
||||
});
|
||||
|
||||
monitor.startMonitoring("KB-001", "owner", "repo", mockPrInfo);
|
||||
|
||||
// Wait for initial check
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
|
||||
expect(mockFetch).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("calls onNewComments when new comments found", async () => {
|
||||
const callback = vi.fn();
|
||||
monitor.onNewComments(callback);
|
||||
|
||||
mockFetch.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
json: () => Promise.resolve([mockComment]),
|
||||
});
|
||||
|
||||
monitor.startMonitoring("KB-001", "owner", "repo", mockPrInfo);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
|
||||
expect(callback).toHaveBeenCalledWith("KB-001", mockPrInfo, [mockComment]);
|
||||
});
|
||||
|
||||
it("tracks lastCommentId to avoid duplicate notifications", async () => {
|
||||
const callback = vi.fn();
|
||||
monitor.onNewComments(callback);
|
||||
|
||||
mockFetch
|
||||
.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
json: () => Promise.resolve([mockComment]),
|
||||
})
|
||||
.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
json: () => Promise.resolve([mockComment]), // Same comment again
|
||||
});
|
||||
|
||||
monitor.startMonitoring("KB-001", "owner", "repo", mockPrInfo);
|
||||
|
||||
// First check
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(callback).toHaveBeenCalledTimes(1);
|
||||
|
||||
// Second scheduled check after 30s
|
||||
await vi.advanceTimersByTimeAsync(30 * 1000);
|
||||
|
||||
// Second poll should not trigger callback for same comment
|
||||
expect(callback).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
});
|
||||
261
packages/engine/src/pr-monitor.ts
Normal file
261
packages/engine/src/pr-monitor.ts
Normal file
@@ -0,0 +1,261 @@
|
||||
import { prMonitorLog } from "./logger.js";
|
||||
import type { PrInfo } from "@kb/core";
|
||||
|
||||
export interface TrackedPr {
|
||||
owner: string;
|
||||
repo: string;
|
||||
prInfo: PrInfo;
|
||||
lastCheckedAt: Date;
|
||||
lastCommentId?: number;
|
||||
consecutiveErrors: number;
|
||||
isActive: boolean; // true if we've seen recent activity
|
||||
}
|
||||
|
||||
export interface PrComment {
|
||||
id: number;
|
||||
body: string;
|
||||
user: { login: string };
|
||||
created_at: string;
|
||||
updated_at: string;
|
||||
html_url: string;
|
||||
}
|
||||
|
||||
export type OnNewCommentsCallback = (
|
||||
taskId: string,
|
||||
prInfo: PrInfo,
|
||||
comments: PrComment[]
|
||||
) => void | Promise<void>;
|
||||
|
||||
/**
|
||||
* Monitors GitHub PRs for new comments.
|
||||
* Uses adaptive polling: 30s when active, 5min when idle.
|
||||
* Implements exponential backoff on errors.
|
||||
*/
|
||||
export class PrMonitor {
|
||||
private trackedPrs = new Map<string, TrackedPr>();
|
||||
private intervals = new Map<string, ReturnType<typeof setInterval>>();
|
||||
private newCommentsCallback?: OnNewCommentsCallback;
|
||||
private getGitHubToken: () => string | undefined;
|
||||
|
||||
// Polling intervals in ms
|
||||
private readonly ACTIVE_INTERVAL = 30 * 1000; // 30 seconds
|
||||
private readonly IDLE_INTERVAL = 5 * 60 * 1000; // 5 minutes
|
||||
private readonly MIN_INTERVAL = 30 * 1000;
|
||||
private readonly MAX_INTERVAL = 15 * 60 * 1000; // 15 minutes max backoff
|
||||
|
||||
constructor(options: { getGitHubToken?: () => string | undefined } = {}) {
|
||||
this.getGitHubToken = options.getGitHubToken ?? (() => process.env.GITHUB_TOKEN);
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a callback to be called when new comments are found.
|
||||
*/
|
||||
onNewComments(callback: OnNewCommentsCallback): void {
|
||||
this.newCommentsCallback = callback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Start monitoring a PR for comments.
|
||||
*/
|
||||
startMonitoring(
|
||||
taskId: string,
|
||||
owner: string,
|
||||
repo: string,
|
||||
prInfo: PrInfo
|
||||
): void {
|
||||
// Stop any existing monitoring for this task
|
||||
this.stopMonitoring(taskId);
|
||||
|
||||
const tracked: TrackedPr = {
|
||||
owner,
|
||||
repo,
|
||||
prInfo,
|
||||
lastCheckedAt: new Date(),
|
||||
lastCommentId: undefined,
|
||||
consecutiveErrors: 0,
|
||||
isActive: true, // Start as active
|
||||
};
|
||||
|
||||
this.trackedPrs.set(taskId, tracked);
|
||||
|
||||
// Do an initial check immediately
|
||||
this.checkForComments(taskId, tracked);
|
||||
|
||||
// Set up polling interval
|
||||
this.scheduleNextCheck(taskId, tracked);
|
||||
|
||||
prMonitorLog.log(`Started monitoring PR #${prInfo.number} for task ${taskId}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop monitoring a PR.
|
||||
*/
|
||||
stopMonitoring(taskId: string): void {
|
||||
const interval = this.intervals.get(taskId);
|
||||
if (interval) {
|
||||
clearTimeout(interval);
|
||||
this.intervals.delete(taskId);
|
||||
}
|
||||
|
||||
if (this.trackedPrs.has(taskId)) {
|
||||
this.trackedPrs.delete(taskId);
|
||||
prMonitorLog.log(`Stopped monitoring task ${taskId}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop monitoring all PRs. Called on scheduler shutdown.
|
||||
*/
|
||||
stopAll(): void {
|
||||
for (const [taskId] of this.trackedPrs) {
|
||||
this.stopMonitoring(taskId);
|
||||
}
|
||||
prMonitorLog.log("Stopped all PR monitoring");
|
||||
}
|
||||
|
||||
/**
|
||||
* Get currently tracked PRs (for testing/debugging).
|
||||
*/
|
||||
getTrackedPrs(): Map<string, TrackedPr> {
|
||||
return new Map(this.trackedPrs);
|
||||
}
|
||||
|
||||
private scheduleNextCheck(taskId: string, tracked: TrackedPr): void {
|
||||
// Calculate interval based on activity and error count
|
||||
let interval = tracked.isActive ? this.ACTIVE_INTERVAL : this.IDLE_INTERVAL;
|
||||
|
||||
// Exponential backoff on errors: 30s * 2^errors, capped at 15min
|
||||
if (tracked.consecutiveErrors > 0) {
|
||||
const backoffMultiplier = Math.pow(2, Math.min(tracked.consecutiveErrors, 5));
|
||||
interval = Math.min(interval * backoffMultiplier, this.MAX_INTERVAL);
|
||||
}
|
||||
|
||||
const timeoutId = setTimeout(() => {
|
||||
this.checkForComments(taskId, tracked).then(() => {
|
||||
// Reschedule if still tracked
|
||||
if (this.trackedPrs.has(taskId)) {
|
||||
this.scheduleNextCheck(taskId, tracked);
|
||||
}
|
||||
});
|
||||
}, interval);
|
||||
|
||||
this.intervals.set(taskId, timeoutId);
|
||||
}
|
||||
|
||||
private async checkForComments(
|
||||
taskId: string,
|
||||
tracked: TrackedPr
|
||||
): Promise<boolean> {
|
||||
const token = this.getGitHubToken();
|
||||
if (!token) {
|
||||
prMonitorLog.warn(`No GitHub token available for task ${taskId}`);
|
||||
tracked.consecutiveErrors++;
|
||||
return false; // Don't reschedule - wait for next scheduled check
|
||||
}
|
||||
|
||||
try {
|
||||
const since = tracked.lastCheckedAt.toISOString();
|
||||
const comments = await this.fetchComments(
|
||||
tracked.owner,
|
||||
tracked.repo,
|
||||
tracked.prInfo.number,
|
||||
since,
|
||||
token
|
||||
);
|
||||
|
||||
// Filter to only new comments (by ID)
|
||||
const newComments = tracked.lastCommentId
|
||||
? comments.filter((c) => c.id > tracked.lastCommentId!)
|
||||
: comments;
|
||||
|
||||
if (newComments.length > 0) {
|
||||
prMonitorLog.log(
|
||||
`Found ${newComments.length} new comment(s) on PR #${tracked.prInfo.number}`
|
||||
);
|
||||
|
||||
// Update lastCommentId
|
||||
const maxId = Math.max(...newComments.map((c) => c.id));
|
||||
tracked.lastCommentId = maxId;
|
||||
|
||||
// Mark as active since we found new comments
|
||||
tracked.isActive = true;
|
||||
|
||||
// Notify handler
|
||||
if (this.newCommentsCallback) {
|
||||
try {
|
||||
await this.newCommentsCallback(taskId, tracked.prInfo, newComments);
|
||||
} catch (err) {
|
||||
prMonitorLog.error(`Error handling new comments for ${taskId}:`, err);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// No new comments - mark as idle after 5 minutes of no activity
|
||||
const timeSinceLastComment = Date.now() - tracked.lastCheckedAt.getTime();
|
||||
if (timeSinceLastComment > 5 * 60 * 1000) {
|
||||
tracked.isActive = false;
|
||||
}
|
||||
}
|
||||
|
||||
// Reset error count on success
|
||||
tracked.consecutiveErrors = 0;
|
||||
tracked.lastCheckedAt = new Date();
|
||||
return true;
|
||||
} catch (err: any) {
|
||||
tracked.consecutiveErrors++;
|
||||
prMonitorLog.error(
|
||||
`Error checking PR #${tracked.prInfo.number} for task ${taskId} ` +
|
||||
`(attempt ${tracked.consecutiveErrors}):`,
|
||||
err.message
|
||||
);
|
||||
|
||||
// Disable monitoring after 5 consecutive failures
|
||||
if (tracked.consecutiveErrors >= 5) {
|
||||
prMonitorLog.warn(
|
||||
`Disabling PR monitoring for task ${taskId} after 5 consecutive failures`
|
||||
);
|
||||
this.stopMonitoring(taskId);
|
||||
return false;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private async fetchComments(
|
||||
owner: string,
|
||||
repo: string,
|
||||
prNumber: number,
|
||||
since: string,
|
||||
token: string
|
||||
): Promise<PrComment[]> {
|
||||
const params = new URLSearchParams();
|
||||
params.append("per_page", "100");
|
||||
if (since) {
|
||||
params.append("since", since);
|
||||
}
|
||||
|
||||
const url = `https://api.github.com/repos/${encodeURIComponent(
|
||||
owner
|
||||
)}/${encodeURIComponent(repo)}/issues/${prNumber}/comments?${params}`;
|
||||
|
||||
const headers: Record<string, string> = {
|
||||
Accept: "application/vnd.github+json",
|
||||
"X-GitHub-Api-Version": "2022-11-28",
|
||||
"User-Agent": "kb-engine/1.0",
|
||||
Authorization: `Bearer ${token}`,
|
||||
};
|
||||
|
||||
const response = await fetch(url, { headers });
|
||||
|
||||
if (!response.ok) {
|
||||
if (response.status === 404) {
|
||||
throw new Error(`PR #${prNumber} not found in ${owner}/${repo}`);
|
||||
}
|
||||
if (response.status === 401 || response.status === 403) {
|
||||
throw new Error("Authentication failed or rate limited");
|
||||
}
|
||||
throw new Error(`GitHub API error: ${response.status} ${response.statusText}`);
|
||||
}
|
||||
|
||||
return response.json() as Promise<PrComment[]>;
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,8 @@
|
||||
import { resolveDependencyOrder, type TaskStore, type Task } from "@kb/core";
|
||||
import type { AgentSemaphore } from "./concurrency.js";
|
||||
import { schedulerLog } from "./logger.js";
|
||||
import type { PrMonitor } from "./pr-monitor.js";
|
||||
import { getCurrentGitHubRepo } from "./github.js";
|
||||
|
||||
/**
|
||||
* Check whether two sets of file scope paths overlap.
|
||||
@@ -53,6 +55,8 @@ export interface SchedulerOptions {
|
||||
onSchedule?: (task: Task) => void;
|
||||
/** Called when a task is blocked by deps */
|
||||
onBlocked?: (task: Task, blockedBy: string[]) => void;
|
||||
/** Optional PR monitor for tracking in-review PRs */
|
||||
prMonitor?: PrMonitor;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -111,6 +115,48 @@ export class Scheduler {
|
||||
this.schedule();
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* PR Monitoring: Start monitoring when a task moves to "in-review",
|
||||
* stop monitoring when it moves out.
|
||||
*/
|
||||
this.store.on("task:moved", ({ task, to }) => {
|
||||
if (!this.options.prMonitor) return;
|
||||
|
||||
if (to === "in-review" && task.prInfo) {
|
||||
// Start monitoring existing PR
|
||||
const repo = getCurrentGitHubRepo(this.store.getRootDir());
|
||||
if (repo) {
|
||||
this.options.prMonitor.startMonitoring(task.id, repo.owner, repo.repo, task.prInfo);
|
||||
}
|
||||
} else if (task.column === "in-review" && to !== "in-review") {
|
||||
// Task moved out of in-review, stop monitoring
|
||||
this.options.prMonitor.stopMonitoring(task.id);
|
||||
|
||||
// If task has a closed/merged PR, check for unaddressed feedback
|
||||
if (task.prInfo && (task.prInfo.status === "closed" || task.prInfo.status === "merged")) {
|
||||
// This would need the tracked PR data - handled by PrMonitor/PrCommentHandler
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* PR Monitoring: Start monitoring when PR is linked to an in-review task.
|
||||
*/
|
||||
this.store.on("task:updated", (task) => {
|
||||
if (!this.options.prMonitor) return;
|
||||
if (task.column !== "in-review") return;
|
||||
if (!task.prInfo) return;
|
||||
|
||||
// Check if we're already monitoring this task
|
||||
const tracked = this.options.prMonitor.getTrackedPrs();
|
||||
if (!tracked.has(task.id)) {
|
||||
const repo = getCurrentGitHubRepo(this.store.getRootDir());
|
||||
if (repo) {
|
||||
this.options.prMonitor.startMonitoring(task.id, repo.owner, repo.repo, task.prInfo);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
start(): void {
|
||||
@@ -131,6 +177,10 @@ export class Scheduler {
|
||||
this.pollInterval = null;
|
||||
this.activePollMs = null;
|
||||
}
|
||||
// Stop all PR monitoring when scheduler shuts down
|
||||
if (this.options.prMonitor) {
|
||||
this.options.prMonitor.stopAll();
|
||||
}
|
||||
schedulerLog.log("Stopped");
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user