feat: add dynamic orchestration with replanning, inter-job comms, session forking, and fix-before-rollback (#68)

Author: nigel-dev

src/lib/config.ts

@@ -25,6 +25,7 @@ const DEFAULT_CONFIG: MCConfig = {
   useServeMode: true,
   portRangeStart: 14100,
   portRangeEnd: 14199,
+  fixBeforeRollbackTimeout: 120000,
   omo: {
     enabled: false,
     defaultMode: 'vanilla',

src/lib/job-comms.ts

@@ -0,0 +1,143 @@
+import type { JobSpec } from './plan-types';
+import { sendPrompt, waitForServer } from './sdk-client';
+
+export interface RelayContext {
+  finding: string;
+  filePath?: string;
+  lineNumber?: number;
+  severity?: 'info' | 'warning' | 'error';
+}
+
+export interface RelayMessage {
+  from: string;
+  to: string;
+  context: RelayContext;
+  timestamp: string;
+}
+
+export class JobComms {
+  private messageBus: Map<string, RelayMessage[]> = new Map();
+  private relayPatterns: Map<string, Bun.Glob[]> = new Map();
+  private relayPatternSources: Map<string, string[]> = new Map();
+
+  registerJob(job: JobSpec): void {
+    if (job.relayPatterns && job.relayPatterns.length > 0) {
+      const patterns: Bun.Glob[] = [];
+      const sources: string[] = [];
+      for (const pattern of job.relayPatterns) {
+        const normalized = pattern.endsWith('/') ? `${pattern}**` : pattern;
+        patterns.push(new Bun.Glob(normalized));
+        sources.push(pattern);
+      }
+      this.relayPatterns.set(job.name, patterns);
+      this.relayPatternSources.set(job.name, sources);
+    }
+    if (!this.messageBus.has(job.name)) {
+      this.messageBus.set(job.name, []);
+    }
+  }
+
+  unregisterJob(jobName: string): void {
+    this.relayPatterns.delete(jobName);
+    this.relayPatternSources.delete(jobName);
+    this.messageBus.delete(jobName);
+  }
+
+  relayFinding(from: string, to: string, context: RelayContext): void {
+    const message: RelayMessage = {
+      from,
+      to,
+      context,
+      timestamp: new Date().toISOString(),
+    };
+
+    const messages = this.messageBus.get(to) ?? [];
+    messages.push(message);
+    this.messageBus.set(to, messages);
+  }
+
+  getMessagesForJob(jobName: string): RelayMessage[] {
+    return this.messageBus.get(jobName) ?? [];
+  }
+
+  clearMessagesForJob(jobName: string): void {
+    this.messageBus.set(jobName, []);
+  }
+
+  shouldRelayForFile(jobName: string, filePath: string): boolean {
+    const patterns = this.relayPatterns.get(jobName);
+    if (!patterns || patterns.length === 0) {
+      return false;
+    }
+    return patterns.some((pattern) => pattern.match(filePath));
+  }
+
+  async deliverMessages(
+    job: JobSpec,
+    options?: { filterFrom?: string[] },
+  ): Promise<number> {
+    const messages = this.getMessagesForJob(job.name);
+    if (messages.length === 0) {
+      return 0;
+    }
+
+    const filtered = options?.filterFrom
+      ? messages.filter((m) => options.filterFrom!.includes(m.from))
+      : messages;
+
+    if (filtered.length === 0) {
+      return 0;
+    }
+
+    if (!job.port) {
+      return 0;
+    }
+
+    try {
+      const client = await waitForServer(job.port, { timeoutMs: 5000 });
+
+      for (const message of filtered) {
+        const prompt = this.formatRelayPrompt(message);
+        await sendPrompt(client, job.launchSessionID ?? '', prompt);
+      }
+
+      this.clearMessagesForJob(job.name);
+      return filtered.length;
+    } catch {
+      return 0;
+    }
+  }
+
+  private formatRelayPrompt(message: RelayMessage): string {
+    const { from, context } = message;
+    const { finding, filePath, lineNumber, severity } = context;
+
+    const parts: string[] = [`[Inter-Job Communication from ${from}]`];
+
+    if (severity) {
+      parts.push(`Severity: ${severity.toUpperCase()}`);
+    }
+
+    parts.push(`Finding: ${finding}`);
+
+    if (filePath) {
+      parts.push(`File: ${filePath}`);
+    }
+
+    if (lineNumber) {
+      parts.push(`Line: ${lineNumber}`);
+    }
+
+    parts.push('\nConsider how this finding may affect your current work.');
+
+    return parts.join('\n');
+  }
+
+  getAllRegisteredJobs(): string[] {
+    return Array.from(this.messageBus.keys());
+  }
+
+  getRelayPatternsForJob(jobName: string): string[] | undefined {
+    return this.relayPatternSources.get(jobName);
+  }
+}

src/lib/merge-train.ts

@@ -4,6 +4,7 @@ import type { JobSpec } from './plan-types';
 import { gitCommand } from './git';
 import { getIntegrationWorktree } from './integration';
 import { extractConflicts } from './utils';
+import { createJobClient, sendPrompt, waitForServer } from './sdk-client';
 
 export type MergeResult =
   | { success: true; mergedAt: string; testReport: MergeTestReport }
@@ -35,9 +36,11 @@ type MergeTrainConfig = {
   testTimeout?: number;
   mergeStrategy?: 'squash' | 'ff-only' | 'merge';
   setupCommands?: string[];
+  fixBeforeRollbackTimeout?: number;
 };
 
 const DEFAULT_TEST_TIMEOUT_MS = 600000;
+const DEFAULT_FIX_BEFORE_ROLLBACK_TIMEOUT_MS = 120000;
 
 const INSTALL_COMMAND_BY_LOCKFILE = [
   { file: 'bun.lockb', command: 'bun install --frozen-lockfile' },
@@ -564,6 +567,37 @@ export class MergeTrain {
     }
 
     if (!testResult.success) {
+      const fixTimeout = this.config?.fixBeforeRollbackTimeout ?? DEFAULT_FIX_BEFORE_ROLLBACK_TIMEOUT_MS;
+      const isServeMode = job.port !== undefined;
+
+      if (isServeMode && fixTimeout > 0) {
+        const fixPrompt = this.buildFixPrompt(job.name, testCommand, testResult.output);
+        const fixPrompted = await this.promptAgentForFix(job, fixPrompt);
+
+        if (fixPrompted) {
+          await this.sleep(fixTimeout);
+
+          const retestResult = await runTestCommand(this.integrationWorktree, testCommand, timeoutMs);
+
+          if (retestResult.success) {
+            return {
+              success: true,
+              mergedAt: new Date().toISOString(),
+              testReport: {
+                status: 'passed',
+                command: testCommand,
+                output: retestResult.output || undefined,
+                setup: {
+                  status: dependencySetupResult.status,
+                  commands: dependencySetupResult.commands,
+                  output: dependencySetupResult.output || undefined,
+                },
+              },
+            };
+          }
+        }
+      }
+
       await rollbackMergeToHead(this.integrationWorktree, headBeforeStr);
       return {
         success: false,
@@ -609,4 +643,35 @@ export class MergeTrain {
 
     return results;
   }
+
+  private buildFixPrompt(jobName: string, testCommand: string, testOutput: string): string {
+    return `[AUTO-GENERATED] Test Failure in Merge Train
+
+Job "${jobName}" failed during merge train testing.
+
+Test Command: ${testCommand}
+
+Test Output:
+${testOutput.slice(0, 2000)}
+
+Please fix the failing tests. You have a limited time window to apply fixes before the merge is rolled back. Focus on the most critical failures first.`;
+  }
+
+  private async promptAgentForFix(job: JobSpec, prompt: string): Promise<boolean> {
+    if (!job.port || !job.launchSessionID) {
+      return false;
+    }
+
+    try {
+      const client = await waitForServer(job.port, { timeoutMs: 10000 });
+      await sendPrompt(client, job.launchSessionID, prompt);
+      return true;
+    } catch {
+      return false;
+    }
+  }
+
+  private sleep(ms: number): Promise<void> {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+  }
 }