feat(FN-4409): add branch contamination assertions in worktree flows

Fusion-Task-Id: FN-4409
Fusion-Task-Lineage: c608c942-fec7-4489-9970-b8dc314f8be2

packages/engine/src/__tests__/branch-conflicts.test.ts

@@ -42,7 +42,13 @@ vi.mock("node:fs", () => ({
 
 import { execSync } from "node:child_process";
 import { existsSync } from "node:fs";
-import { inspectBranchConflict, listBranchRecoveryCandidates, BranchConflictError } from "../branch-conflicts.js";
+import {
+  BranchConflictError,
+  BranchCrossContaminationError,
+  assertCleanBranchAtBase,
+  inspectBranchConflict,
+  listBranchRecoveryCandidates,
+} from "../branch-conflicts.js";
 
 const mockedExecSync = vi.mocked(execSync);
 const mockedExistsSync = vi.mocked(existsSync);
@@ -140,6 +146,31 @@ describe("branch-conflicts", () => {
     expect(result.error.message).toContain("2 stranded commits since main");
   });
 
+  it("assertCleanBranchAtBase passes when no foreign task commits exist", async () => {
+    mockedExecSync.mockImplementation((cmd: string | string[]) => {
+      const command = typeof cmd === "string" ? cmd : cmd[0];
+      if (command.includes("git log --format=%H%x1f%s%x1f%b 'main..fusion/fn-4068'")) {
+        return Buffer.from("aaa111\u001ffeat(FN-4068): own\u001fFusion-Task-Id: FN-4068\n");
+      }
+      throw new Error(`Unexpected command: ${command}`);
+    });
+
+    await expect(assertCleanBranchAtBase("/tmp/repo", "fusion/fn-4068", "main", "FN-4068")).resolves.toBeUndefined();
+  });
+
+  it("assertCleanBranchAtBase throws BranchCrossContaminationError for foreign commits", async () => {
+    mockedExecSync.mockImplementation((cmd: string | string[]) => {
+      const command = typeof cmd === "string" ? cmd : cmd[0];
+      if (command.includes("git log --format=%H%x1f%s%x1f%b 'main..fusion/fn-4068'")) {
+        return Buffer.from("bbb222\u001ffeat(FN-4386): foreign\u001fFusion-Task-Id: FN-4386\n");
+      }
+      throw new Error(`Unexpected command: ${command}`);
+    });
+
+    await expect(assertCleanBranchAtBase("/tmp/repo", "fusion/fn-4068", "main", "FN-4068"))
+      .rejects.toBeInstanceOf(BranchCrossContaminationError);
+  });
+
   it("lists canonical and sibling recovery candidates with worktrees and stranded commits", async () => {
     mockedExecSync.mockImplementation((cmd: string | string[]) => {
       const command = typeof cmd === "string" ? cmd : cmd[0];

packages/engine/src/branch-conflicts.ts

@@ -10,6 +10,10 @@ export interface BranchConflictCommit {
   subject: string;
 }
 
+export interface BranchCrossContaminationCommit extends BranchConflictCommit {
+  foreignTaskId: string;
+}
+
 export interface BranchRecoveryCandidate {
   branchName: string;
   tipSha: string;
@@ -58,6 +62,32 @@ export function isBranchConflictError(error: unknown): error is BranchConflictEr
   return error instanceof BranchConflictError;
 }
 
+export interface BranchCrossContaminationDetails {
+  branchName: string;
+  baseSha: string;
+  taskId: string;
+  foreignCommits: BranchCrossContaminationCommit[];
+}
+
+export class BranchCrossContaminationError extends Error implements BranchCrossContaminationDetails {
+  readonly name = "BranchCrossContaminationError";
+  readonly branchName: string;
+  readonly baseSha: string;
+  readonly taskId: string;
+  readonly foreignCommits: BranchCrossContaminationCommit[];
+
+  constructor(details: BranchCrossContaminationDetails) {
+    super(
+      `Branch ${details.branchName} contains ${details.foreignCommits.length} foreign task-attributed commits ` +
+      `since base ${details.baseSha.slice(0, 12)} for ${details.taskId}`,
+    );
+    this.branchName = details.branchName;
+    this.baseSha = details.baseSha;
+    this.taskId = details.taskId;
+    this.foreignCommits = details.foreignCommits;
+  }
+}
+
 export interface InspectBranchConflictInput {
   repoDir: string;
   branchName: string;
@@ -202,6 +232,34 @@ async function countTaskAttributedCommits(repoDir: string, range: string, taskId
   return count;
 }
 
+export async function assertCleanBranchAtBase(
+  repoDir: string,
+  branchName: string,
+  baseSha: string,
+  taskId: string,
+): Promise<void> {
+  const output = await runGit(repoDir, `git log --format=%H%x1f%s%x1f%b ${quoteShellArg(`${baseSha}..${branchName}`)}`)
+    .catch(() => "");
+  if (!output) return;
+
+  const subjectPattern = /^(feat|fix|test|chore|docs|refactor|perf|build)\((FN-\d+)\):/i;
+  const trailerPattern = /(?:^|\n)Fusion-Task-Id:\s*(FN-\d+)\s*(?:\n|$)/i;
+  const foreignCommits: BranchCrossContaminationCommit[] = [];
+  for (const line of output.split("\n").map((entry) => entry.trim()).filter(Boolean)) {
+    const [sha, subject, body] = line.split("\u001f");
+    const subjectMatch = (subject ?? "").match(subjectPattern);
+    const trailerMatch = (body ?? "").match(trailerPattern);
+    const attributedTaskId = (trailerMatch?.[1] ?? subjectMatch?.[2] ?? "").toUpperCase();
+    if (attributedTaskId && attributedTaskId !== taskId.toUpperCase()) {
+      foreignCommits.push({ sha, subject: subject ?? "", foreignTaskId: attributedTaskId });
+    }
+  }
+
+  if (foreignCommits.length > 0) {
+    throw new BranchCrossContaminationError({ branchName, baseSha, taskId, foreignCommits });
+  }
+}
+
 export async function inspectBranchConflict(
   input: InspectBranchConflictInput,
 ): Promise<BranchConflictInspectionResult> {

packages/engine/src/executor.ts

@@ -38,7 +38,13 @@ import { reviewStep, type ReviewVerdict } from "./reviewer.js";
 import { ModelRegistry, SessionManager, type ToolDefinition, type AgentSession } from "@mariozechner/pi-coding-agent";
 import { PRIORITY_EXECUTE, type AgentSemaphore } from "./concurrency.js";
 import { getRegisteredWorktreePaths, isGitRepository, isInsideWorktreesDir, isRegisteredGitWorktree, isUsableTaskWorktree, type WorktreePool } from "./worktree-pool.js";
-import { BranchConflictError, isBranchConflictError, inspectBranchConflict } from "./branch-conflicts.js";
+import {
+  BranchConflictError,
+  BranchCrossContaminationError,
+  assertCleanBranchAtBase,
+  isBranchConflictError,
+  inspectBranchConflict,
+} from "./branch-conflicts.js";
 import { AgentLogger } from "./agent-logger.js";
 import { executorLog, reviewerLog, formatError } from "./logger.js";
 import { TokenCapDetector } from "./token-cap-detector.js";
@@ -2507,6 +2513,12 @@ export class TaskExecutor {
         await this.captureBaseCommitSha(task, worktreePath, audit);
       }
 
+      const latestTaskForBase = await this.store.getTask(task.id);
+      const contaminationBaseRef = await this.resolveDiffBaseRef(worktreePath, latestTaskForBase.baseCommitSha);
+      if (contaminationBaseRef) {
+        await assertCleanBranchAtBase(this.rootDir, acquisition.branch, contaminationBaseRef, task.id);
+      }
+
       const expectedRoot = canonicalizePath(this.rootDir);
       let observedWorktreeRealpath: string;
       let livenessFailure: string | null = null;
@@ -3948,6 +3960,18 @@ export class TaskExecutor {
             nextRecoveryAt: null,
           });
           // Fall through to terminal failure marking
+        } else if (err instanceof BranchCrossContaminationError) {
+          const details = err.foreignCommits
+            .map((commit) => `${commit.sha.slice(0, 12)}:${commit.foreignTaskId}`)
+            .join(", ");
+          await this.store.logEntry(task.id, `[recovery] branch cross-contamination detected on ${err.branchName} since ${err.baseSha}: ${details}`, undefined, this.currentRunContext);
+          await this.store.updateTask(task.id, {
+            status: "failed",
+            error: err.message,
+            paused: true,
+            pausedReason: "branch-cross-contamination",
+          });
+          return;
         } else if (isBranchConflictError(err)) {
           const conflictCount = (this.branchConflictErrorCount.get(task.id) ?? 0) + 1;
           this.branchConflictErrorCount.set(task.id, conflictCount);
@@ -6420,6 +6444,11 @@ and show an appropriate message to the user.\`
     count: number,
   ): Promise<void> {
     await this.store.updateTask(task.id, { worktree: livePath, branch });
+    const latestTask = await this.store.getTask(task.id);
+    const baseRef = await this.resolveDiffBaseRef(livePath, latestTask.baseCommitSha);
+    if (baseRef) {
+      await assertCleanBranchAtBase(this.rootDir, branch, baseRef, task.id);
+    }
     const message = `[recovery] reclaimed existing worktree for ${task.id} at ${livePath} (${count} commits preserved, tip ${tipSha.slice(0, 12)})`;
     await this.store.logEntry(task.id, message, undefined, this.currentRunContext);
     await this.store.appendAgentLog(task.id, "Branch conflict auto-recovery", "text", message, "executor");

packages/engine/src/index.ts

@@ -95,6 +95,8 @@ export { HEARTBEAT_PROCEDURE, HEARTBEAT_SYSTEM_PROMPT, HEARTBEAT_NO_TASK_SYSTEM_
 export { WorktreePool, scanIdleWorktrees, cleanupOrphanedWorktrees, reapOrphanWorktrees } from "./worktree-pool.js";
 export {
   BranchConflictError,
+  BranchCrossContaminationError,
+  assertCleanBranchAtBase,
   isBranchConflictError,
   inspectBranchConflict,
   listBranchRecoveryCandidates,

packages/engine/src/worktree-pool.ts

@@ -3,7 +3,7 @@ import { promisify } from "node:util";
 import { existsSync, lstatSync, readdirSync, rmSync, realpathSync } from "node:fs";
 import { join, relative, resolve, isAbsolute } from "node:path";
 import type { Column, TaskStore } from "@fusion/core";
-import { inspectBranchConflict } from "./branch-conflicts.js";
+import { assertCleanBranchAtBase, inspectBranchConflict } from "./branch-conflicts.js";
 import { worktreePoolLog } from "./logger.js";
 
 const execAsync = promisify(exec);
@@ -101,6 +101,11 @@ export function isInsideWorktreesDir(rootDir: string, worktreePath: string): boo
  * pool. When `recycleWorktrees` is false, orphaned worktrees are cleaned
  * up via {@link cleanupOrphanedWorktrees}.
  */
+function deriveTaskIdFromBranch(branchName: string): string {
+  const match = branchName.match(/^fusion\/(fn-\d+)(?:-\d+)?$/i);
+  return match ? match[1].toUpperCase() : branchName.toUpperCase();
+}
+
 export class WorktreePool {
   private idle = new Set<string>();
 
@@ -223,10 +228,13 @@ export class WorktreePool {
 
     // Create or force-reset the branch from the start point (or main)
     const checkoutCmd = `git checkout -B "${branchName}" ${base}`;
+    const resolvedBase = (await execAsync(`git rev-parse --verify "${base}^{commit}"`, { cwd: worktreePath, encoding: "utf-8" })).stdout.trim();
+    const taskId = deriveTaskIdFromBranch(branchName);
     try {
       await execAsync(checkoutCmd, {
         cwd: worktreePath,
       });
+      await assertCleanBranchAtBase(worktreePath, branchName, resolvedBase, taskId);
       return branchName;
     } catch (err: unknown) {
       const execError = err instanceof Error ? err : new Error(String(err));
@@ -253,6 +261,7 @@ export class WorktreePool {
       if (inspection.kind === "stale") {
         await execAsync("git worktree prune", { cwd: worktreePath });
         await execAsync(checkoutCmd, { cwd: worktreePath });
+        await assertCleanBranchAtBase(worktreePath, branchName, resolvedBase, taskId);
         return branchName;
       }
 
@@ -269,6 +278,7 @@ export class WorktreePool {
         const suffixedCmd = `git checkout -B "${suffixedName}" ${conflictBase}`;
         try {
           await execAsync(suffixedCmd, { cwd: worktreePath });
+          await assertCleanBranchAtBase(worktreePath, suffixedName, resolvedBase, taskId);
           return suffixedName;
         } catch (suffixErr: unknown) {
           const suffixExecError = suffixErr instanceof Error ? suffixErr : new Error(String(suffixErr));