From 876d38b33ec2cc18622aecef191e4a5abec50225 Mon Sep 17 00:00:00 2001 From: Prateek Date: Mon, 16 Feb 2026 06:09:31 +0530 Subject: [PATCH] feat: auto-rebase open PR branches when main advances (INT-1364) Automatically rebases all open PR branches onto the new main after each merge, detecting conflicts early and pushing clean rebases silently. - Add rebaseAndPush() and getBranchSHA() to SCM interface - Implement rebase logic in GitHub SCM plugin with --force-with-lease - Create MainBranchTracker to detect when main branch advances - Integrate auto-rebase into lifecycle polling loop (30s cycle) - Add rebase tracking fields to SessionMetadata - Add pr.rebased, pr.rebase_conflict, pr.rebase_error event types - Add rebase-conflicts default reaction (escalates after 30m) - Skip rebasing when agent is actively processing - Silent success for clean rebases, urgent notification for conflicts Safety: uses execFile, --force-with-lease, dirty workspace checks, rollback on push rejection, aborts on conflicts. All new metadata fields optional for backwards compatibility. Co-Authored-By: Claude Sonnet 4.5 --- packages/core/src/config.ts | 7 ++ packages/core/src/index.ts | 3 + packages/core/src/lifecycle-manager.ts | 115 +++++++++++++++++++++++ packages/core/src/main-branch-tracker.ts | 98 +++++++++++++++++++ packages/core/src/types.ts | 36 +++++++ packages/plugins/scm-github/src/index.ts | 103 ++++++++++++++++++++ 6 files changed, 362 insertions(+) create mode 100644 packages/core/src/main-branch-tracker.ts diff --git a/packages/core/src/config.ts b/packages/core/src/config.ts index 4a4c34070..819aa80a7 100644 --- a/packages/core/src/config.ts +++ b/packages/core/src/config.ts @@ -241,6 +241,13 @@ function applyDefaultReactions(config: OrchestratorConfig): OrchestratorConfig { message: "Your branch has merge conflicts. Rebase on the default branch and resolve them.", escalateAfter: "15m", }, + "rebase-conflicts": { + auto: true, + action: "send-to-agent", + message: + "Your branch has rebase conflicts with main. Fetch latest, rebase, resolve conflicts, and force-push with --force-with-lease.", + escalateAfter: "30m", + }, "approved-and-green": { auto: false, action: "notify", diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 9c8d60a08..eaa7fed2e 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -51,6 +51,9 @@ export type { SessionManagerDeps } from "./session-manager.js"; export { createLifecycleManager } from "./lifecycle-manager.js"; export type { LifecycleManagerDeps } from "./lifecycle-manager.js"; +// Main branch tracker — detects when main advances +export { MainBranchTracker } from "./main-branch-tracker.js"; + // Prompt builder — layered prompt composition export { buildPrompt, BASE_AGENT_PROMPT } from "./prompt-builder.js"; export type { PromptBuildConfig } from "./prompt-builder.js"; diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts index 0ab2df839..8ad886f13 100644 --- a/packages/core/src/lifecycle-manager.ts +++ b/packages/core/src/lifecycle-manager.ts @@ -22,6 +22,7 @@ import { type EventType, type OrchestratorEvent, type OrchestratorConfig, + type ProjectConfig, type ReactionConfig, type ReactionResult, type PluginRegistry, @@ -35,6 +36,7 @@ import { } from "./types.js"; import { updateMetadata } from "./metadata.js"; import { getSessionsDir } from "./paths.js"; +import { MainBranchTracker } from "./main-branch-tracker.js"; /** Parse a duration string like "10m", "30s", "1h" to milliseconds. */ function parseDuration(str: string): number { @@ -141,6 +143,8 @@ function eventToReactionKey(eventType: EventType): string | null { return "bugbot-comments"; case "merge.conflicts": return "merge-conflicts"; + case "pr.rebase_conflict": + return "rebase-conflicts"; case "merge.ready": return "approved-and-green"; case "session.stuck": @@ -174,6 +178,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan const states = new Map(); const reactionTrackers = new Map(); // "sessionId:reactionKey" + const mainBranchTracker = new MainBranchTracker(); let pollTimer: ReturnType | null = null; let polling = false; // re-entrancy guard let allCompleteEmitted = false; // guard against repeated all_complete @@ -500,6 +505,113 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan } } + /** Check for main branch updates and trigger rebases. */ + async function checkMainBranchUpdates(sessions: Session[]): Promise { + // Group sessions by project (only those with open PRs) + const projectSessions = new Map(); + for (const session of sessions) { + if (!session.pr || session.status === "merged" || session.status === "killed") { + continue; + } + const list = projectSessions.get(session.projectId) ?? []; + list.push(session); + projectSessions.set(session.projectId, list); + } + + // Check each project's main branch + for (const [projectId, projectSessionsList] of projectSessions.entries()) { + const project = config.projects[projectId]; + if (!project) continue; + + const scm = project.scm ? registry.get("scm", project.scm.plugin) : null; + if (!scm || !scm.rebaseAndPush) continue; + + try { + // Check if main advanced + const result = await mainBranchTracker.checkMainAdvanced(project, projectId, scm); + if (!result.advanced) continue; + + // Trigger rebases for all open PRs + await rebaseAllSessions(projectSessionsList, project, scm); + } catch { + // Continue with other projects + } + } + } + + /** Rebase all sessions for a project. */ + async function rebaseAllSessions( + sessions: Session[], + project: ProjectConfig, + scm: SCM, + ): Promise { + const agent = registry.get("agent", project.agent ?? config.defaults.agent); + + for (const session of sessions) { + // Skip if agent is actively processing + if (agent && session.runtimeHandle) { + try { + const activity = await agent.getActivityState(session); + if (activity && activity.state === "active") continue; + } catch { + continue; // Can't determine - skip to be safe + } + } + + if (!session.workspacePath || !session.pr || !scm.rebaseAndPush) continue; + + const sessionsDir = getSessionsDir(config.configPath, project.path); + + // Attempt rebase + const result = await scm.rebaseAndPush({ + workspacePath: session.workspacePath, + branch: session.pr.branch, + baseBranch: project.defaultBranch, + remoteName: "origin", + }); + + if (result.success) { + // Silent success - update metadata + updateMetadata(sessionsDir, session.id, { + lastRebaseTime: new Date().toISOString(), + lastRebaseMainSHA: result.newSha ?? "", + rebaseStatus: "clean", + }); + + // Emit low-priority info event + const event = createEvent("pr.rebased", { + sessionId: session.id, + projectId: session.projectId, + message: `Rebased ${session.id} onto ${project.defaultBranch}`, + priority: "info", + }); + await notifyHuman(event, "info"); + } else if (result.conflicted) { + // Conflicts - escalate to human + updateMetadata(sessionsDir, session.id, { + rebaseStatus: "conflicted", + lastRebaseAttempt: new Date().toISOString(), + }); + + const event = createEvent("pr.rebase_conflict", { + sessionId: session.id, + projectId: session.projectId, + message: `${session.id}: Rebase conflicts with ${project.defaultBranch}`, + priority: "urgent", + data: { prUrl: session.pr.url }, + }); + await notifyHuman(event, "urgent"); + } else { + // Other errors - log to metadata, continue + updateMetadata(sessionsDir, session.id, { + rebaseStatus: "error", + rebaseError: result.error ?? "Unknown error", + lastRebaseAttempt: new Date().toISOString(), + }); + } + } + } + /** Run one polling cycle across all sessions. */ async function pollAll(): Promise { // Re-entrancy guard: skip if previous poll is still running @@ -509,6 +621,9 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan try { const sessions = await sessionManager.list(); + // Check for main branch advancement and trigger rebases + await checkMainBranchUpdates(sessions); + // Include sessions that are active OR whose status changed from what we last saw // (e.g., list() detected a dead runtime and marked it "killed" — we need to // process that transition even though the new status is terminal) diff --git a/packages/core/src/main-branch-tracker.ts b/packages/core/src/main-branch-tracker.ts new file mode 100644 index 000000000..d55d3a72f --- /dev/null +++ b/packages/core/src/main-branch-tracker.ts @@ -0,0 +1,98 @@ +/** + * Main Branch Tracker — detects when a project's main branch advances. + * + * Caches the SHA of each project's main branch and detects when it changes. + * In-memory only (acceptable for 30s poll cycle, resets on restart). + */ + +import type { ProjectConfig, SCM } from "./types.js"; + +interface MainBranchState { + projectId: string; + sha: string; + lastChecked: Date; + lastAdvanced: Date | null; +} + +export class MainBranchTracker { + private states: Map = new Map(); + + /** + * Check if a project's main branch has advanced since last check. + * Returns advanced=true only when SHA actually changes (debouncing). + */ + async checkMainAdvanced( + project: ProjectConfig, + projectId: string, + scm: SCM, + ): Promise<{ + advanced: boolean; + oldSha: string | null; + newSha: string; + }> { + if (!scm.getBranchSHA) { + return { advanced: false, oldSha: null, newSha: "" }; + } + + try { + const newSha = await scm.getBranchSHA(project.repo, project.defaultBranch); + const cached = this.states.get(projectId); + + if (!cached) { + // First time checking this project — cache and return no change + this.states.set(projectId, { + projectId, + sha: newSha, + lastChecked: new Date(), + lastAdvanced: null, + }); + return { advanced: false, oldSha: null, newSha }; + } + + const oldSha = cached.sha; + + // Update last checked + cached.lastChecked = new Date(); + + if (newSha !== oldSha) { + // Main branch advanced! + cached.sha = newSha; + cached.lastAdvanced = new Date(); + this.states.set(projectId, cached); + + return { advanced: true, oldSha, newSha }; + } + + return { advanced: false, oldSha, newSha }; + } catch { + // Failed to check — return no change + return { advanced: false, oldSha: null, newSha: "" }; + } + } + + /** + * Manually update the cached SHA for a project (used after manual rebases). + */ + updateState(projectId: string, sha: string): void { + const cached = this.states.get(projectId); + if (cached) { + cached.sha = sha; + cached.lastChecked = new Date(); + this.states.set(projectId, cached); + } else { + this.states.set(projectId, { + projectId, + sha, + lastChecked: new Date(), + lastAdvanced: null, + }); + } + } + + /** + * Get the cached state for a project (for debugging/inspection). + */ + getState(projectId: string): MainBranchState | null { + return this.states.get(projectId) ?? null; + } +} diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 6de795c8f..725ec18f1 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -540,6 +540,14 @@ export interface SCM { /** Check if PR is ready to merge */ getMergeability(pr: PRInfo): Promise; + + // --- Rebase Support --- + + /** Get the current SHA of a branch */ + getBranchSHA?(repo: string, branch: string): Promise; + + /** Rebase a branch in a workspace, push with --force-with-lease */ + rebaseAndPush?(config: RebaseConfig): Promise; } // --- PR Types --- @@ -630,6 +638,23 @@ export interface MergeReadiness { blockers: string[]; } +// --- Rebase Types --- + +export interface RebaseConfig { + workspacePath: string; + branch: string; + baseBranch: string; + remoteName: string; +} + +export interface RebaseResult { + success: boolean; + conflicted: boolean; + error?: string; + oldSha?: string; + newSha?: string; +} + // ============================================================================= // NOTIFIER — Plugin Slot 6 (PRIMARY INTERFACE) // ============================================================================= @@ -730,6 +755,10 @@ export type EventType = // Reactions | "reaction.triggered" | "reaction.escalated" + // Rebase + | "pr.rebased" + | "pr.rebase_conflict" + | "pr.rebase_error" // Summary | "summary.all_complete"; @@ -969,6 +998,13 @@ export interface SessionMetadata { dashboardPort?: number; terminalWsPort?: number; directTerminalWsPort?: number; + + // Rebase tracking (all optional for backwards compatibility) + lastRebaseTime?: string; + lastRebaseMainSHA?: string; + rebaseStatus?: "clean" | "conflicted" | "error"; + rebaseError?: string; + lastRebaseAttempt?: string; } // ============================================================================= diff --git a/packages/plugins/scm-github/src/index.ts b/packages/plugins/scm-github/src/index.ts index e47321afd..18056fde1 100644 --- a/packages/plugins/scm-github/src/index.ts +++ b/packages/plugins/scm-github/src/index.ts @@ -22,6 +22,8 @@ import { type ReviewComment, type AutomatedComment, type MergeReadiness, + type RebaseConfig, + type RebaseResult, } from "@composio/ao-core"; const execFileAsync = promisify(execFile); @@ -68,6 +70,20 @@ function parseDate(val: string | undefined | null): Date { return isNaN(d.getTime()) ? new Date(0) : d; } +async function git(args: string[], cwd: string): Promise { + try { + const { stdout } = await execFileAsync("git", args, { + cwd, + timeout: 60_000, + }); + return stdout.trim(); + } catch (err) { + throw new Error(`git ${args.slice(0, 3).join(" ")} failed: ${(err as Error).message}`, { + cause: err, + }); + } +} + // --------------------------------------------------------------------------- // SCM implementation // --------------------------------------------------------------------------- @@ -560,6 +576,93 @@ function createGitHubSCM(): SCM { blockers, }; }, + + async getBranchSHA(repo: string, branch: string): Promise { + const raw = await gh([ + "api", + `repos/${repo}/git/ref/heads/${branch}`, + "--jq", + ".object.sha", + ]); + return raw; + }, + + async rebaseAndPush(config: RebaseConfig): Promise { + const { workspacePath, branch, baseBranch, remoteName } = config; + + try { + // 1. Fetch latest base branch + await git(["fetch", remoteName, baseBranch], workspacePath); + + // 2. Check workspace is clean + const statusOut = await git(["status", "--porcelain"], workspacePath); + if (statusOut.trim()) { + return { + success: false, + conflicted: false, + error: "Workspace is dirty", + }; + } + + // 3. Get pre-rebase SHA + const oldSha = await git(["rev-parse", "HEAD"], workspacePath); + + // 4. Attempt rebase + try { + await git(["rebase", `${remoteName}/${baseBranch}`], workspacePath); + } catch (err) { + // Check for conflicts + const conflicts = await git( + ["diff", "--name-only", "--diff-filter=U"], + workspacePath, + ).catch(() => ""); + + if (conflicts.trim()) { + // Abort rebase + await git(["rebase", "--abort"], workspacePath).catch(() => {}); + return { + success: false, + conflicted: true, + oldSha, + }; + } + throw err; + } + + // 5. Get new SHA after rebase + const newSha = await git(["rev-parse", "HEAD"], workspacePath); + + // 6. Push with --force-with-lease + try { + await git( + ["push", remoteName, branch, "--force-with-lease"], + workspacePath, + ); + } catch (pushErr) { + // Rollback on force-with-lease rejection + await git(["reset", "--hard", oldSha], workspacePath).catch(() => {}); + return { + success: false, + conflicted: false, + error: `Push rejected: ${(pushErr as Error).message}`, + oldSha, + }; + } + + return { + success: true, + conflicted: false, + oldSha, + newSha, + }; + } catch (err) { + return { + success: false, + conflicted: false, + error: (err as Error).message, + }; + } + }, }; }