Bound final merge and rebase admission

Author: mchwang

runner/merge.ts

@@ -410,10 +410,18 @@ export class MergeCoordinator {
         commandStatus = authorized.status;
       }
       if (commandStatus.remote.mergeQueue) {
-        // The stored cursor is the final external read: events produced by another actor during authorization or the
-        // status checks precede this boundary and therefore cannot be adopted as this attempt's queue lifecycle.
+        // Events produced by another actor during authorization or the preceding status checks come before this
+        // boundary and cannot be adopted as this attempt's lifecycle. The following fresh status read then validates
+        // the PR identity and queue mode that this cursor will be stored with.
         queueWatermark = await (this.gateway as QueueGateway).queueWatermark(commandStatus.remote.head,
           { signal, timeoutMs: 6_000, pullRequest: commandStatus.remote.pullRequest });
+        const boundary = await this.#statusForMerge(view, signal);
+        if (boundary.status.remote.base !== commandStatus.remote.base || boundary.status.remote.head !== commandStatus.remote.head
+          || boundary.status.remote.mergeQueue !== commandStatus.remote.mergeQueue || !samePullRequest(boundary))
+          throw new Error('The pull request changed after queue correlation. Refresh before merging.');
+        if (!boundary.status.ready)
+          throw new Error(`Merge requirements changed after queue correlation. ${boundary.status.blockers[0]!.message}`);
+        commandStatus = boundary.status;
       }
       // The external authorization read completed before the final PR inspection and queue boundary. Re-read local
       // trust synchronously after those awaits so neither remote identity nor local authorization can race admission.

runner/pre-merge.ts

@@ -3,7 +3,8 @@ import { DEFAULT_PROCESS_SETTLEMENT_MS } from '../agents/process-group.ts';
 import type { PlanIdentity } from '../core/identity.ts';
 import type { RunnerCoordinator } from './coordinator.ts';
 import { GuardRefusal, MERGEABLE_STATUSES, settleWith, type ShutdownCapability } from './lifecycle.ts';
-import { RebaseResourcesUnsettled, type GitRebaser } from './rebase.ts';
+import { MAX_REBASE_TIMEOUT_MS, MIN_REBASE_CLEANUP_TIMEOUT_MS, MIN_REBASE_TIMEOUT_MS,
+  RebaseResourcesUnsettled, type GitRebaser } from './rebase.ts';
 import type { ReviewService } from './review.ts';
 import type { PreMergeReadiness, RebaseMarker } from './store.ts';
 
@@ -189,10 +190,10 @@ export class PreMergeCoordinator {
     }
     return null;
   }
-  async #cleanupRebase(identity: PlanIdentity, marker: RebaseMarker): Promise<void> {
+  async #cleanupRebase(identity: PlanIdentity, marker: RebaseMarker, timeoutMs: number): Promise<void> {
     const current = this.service.store.getTask(identity).rebaseInProgress as RebaseMarker | null;
     if (current?.attemptId !== marker.attemptId) throw new GuardRefusal('This rebase attempt no longer owns cleanup.');
-    await this.rebaser.abort(marker.attemptId, current.resultHead ?? undefined, current.resultState);
+    await this.rebaser.abort(marker.attemptId, current.resultHead ?? undefined, current.resultState, timeoutMs);
     if (!this.settle(() => this.service.store.abortRebase(this.service.store.getTask(identity).planKey, marker.attemptId)))
       throw new GuardRefusal('This rebase attempt no longer owns its durable marker.');
   }
@@ -217,6 +218,13 @@ export class PreMergeCoordinator {
         { code: 'ETIMEDOUT' });
       return value;
     };
+    const rebaseBudget = (cleanupOnly = false) => {
+      const value = Math.min(MAX_REBASE_TIMEOUT_MS, remaining());
+      if (value < (cleanupOnly ? MIN_REBASE_CLEANUP_TIMEOUT_MS : MIN_REBASE_TIMEOUT_MS))
+        throw Object.assign(new Error(`Pre-merge preparation deadline exceeded before rebase ${cleanupOnly ? 'cleanup' : 'work and cleanup'} could be reserved.`),
+          { code: 'ETIMEDOUT' });
+      return value;
+    };
     const authorize = async () => {
       if (!this.authorize) return;
       const authorization = await this.authorize(signal);
@@ -268,19 +276,21 @@ export class PreMergeCoordinator {
       const reviewed = { revision: view.expected.revision, snapshotId: view.expected.snapshotId,
         reviewVersion: view.expected.reviewVersion! };
       const oldBase = view.snapshot.base, oldHead = view.snapshot.head;
+      await authorize();
+      signal.throwIfAborted();
+      const rebaseTimeoutMs = rebaseBudget();
       const marker = this.service.store.beginRebase(identity, reviewed, task.stateVersion,
         { oldBase: view.snapshot.base, oldHead: view.snapshot.head, oldHistory: old, onto: initial.base });
       try {
-        await authorize();
         const result = await this.rebaser.run({ attemptId: marker.attemptId, oldBase,
           oldHead, oldHistory: old, onto: initial.base,
-          ledger: this.service.store.getLedger(identity), signal });
+          ledger: this.service.store.getLedger(identity), signal, timeoutMs: rebaseTimeoutMs });
         signal.throwIfAborted();
         this.service.store.finishRebase(identity, reviewed, task.stateVersion, marker.attemptId,
           result.base, result.head, result.mappings);
       } catch (error) {
         if (error instanceof RebaseResourcesUnsettled) throw error;
-        try { await this.#cleanupRebase(identity, marker); }
+        try { await this.#cleanupRebase(identity, marker, rebaseBudget(true)); }
         catch (cleanup) { throw new AggregateError([error, cleanup], error instanceof Error ? error.message : 'Rebase failed.', { cause: error }); }
         // beginRebase and successful cleanup both advance task state. Bind the actionable failure to that settled
         // state without rebuilding history; if a read itself fails, retaining the older binding fails closed as stale.

runner/rebase.ts

@@ -14,7 +14,9 @@ const MAX_REBASE_COMMITS = 500;
 const PROCESS_SETTLEMENT_RESERVE_MS = 13_000;
 // Cleanup receives an operation-wide 30 s: 13 s to settle a cleanup process group and 17 s for Git and follow-up calls.
 const CLEANUP_RESERVE_MS = 30_000;
-const MIN_REBASE_TIMEOUT_MS = CLEANUP_RESERVE_MS + PROCESS_SETTLEMENT_RESERVE_MS + 1;
+export const MIN_REBASE_CLEANUP_TIMEOUT_MS = PROCESS_SETTLEMENT_RESERVE_MS + 1;
+export const MIN_REBASE_TIMEOUT_MS = CLEANUP_RESERVE_MS + PROCESS_SETTLEMENT_RESERVE_MS + 1;
+export const MAX_REBASE_TIMEOUT_MS = 120_000;
 
 export interface RebaseMapping { oldSha: string; newSha: string }
 export interface RebaseConflictInput {
@@ -101,7 +103,7 @@ export class GitRebaser {
     if (!options.committer.name || !options.committer.email || /[<>\n\r\0]/.test(options.committer.name + options.committer.email))
       throw new Error('A valid rebase committer is required.');
     if (options.timeoutMs !== undefined && (!Number.isSafeInteger(options.timeoutMs) ||
-        options.timeoutMs < MIN_REBASE_TIMEOUT_MS || options.timeoutMs > 120_000))
+        options.timeoutMs < MIN_REBASE_TIMEOUT_MS || options.timeoutMs > MAX_REBASE_TIMEOUT_MS))
       throw new Error('Invalid rebase deadline.');
     const git = configuredGit();
     this.#options = options;
@@ -111,7 +113,8 @@ export class GitRebaser {
   }
 
   async run(input: { attemptId: string; oldBase: string; oldHead: string; oldHistory: readonly string[]; onto: string;
-    ledger?: readonly { sha: string; owner: string | null; origin: 'owned' | 'foreign' }[]; signal?: AbortSignal }): Promise<RebaseResult> {
+    ledger?: readonly { sha: string; owner: string | null; origin: 'owned' | 'foreign' }[]; signal?: AbortSignal;
+    timeoutMs?: number }): Promise<RebaseResult> {
     const { attemptId, oldBase, oldHead, oldHistory, onto, signal } = input;
     rebaseRef(attemptId);
     for (const value of [oldBase, oldHead, onto]) if (!COMMIT_ID.test(value)) throw new Error('A full commit ID is required for rebasing.');
@@ -123,7 +126,7 @@ export class GitRebaser {
       ledger.set(entry.sha, { owner: entry.owner, origin: entry.origin });
     }
     signal?.throwIfAborted();
-    const scope = this.#scope(attemptId, signal);
+    const scope = this.#scope(attemptId, signal, input.timeoutMs);
     const ref = rebaseRef(attemptId);
     const existing = await this.#call(this.#options.repository.path, ['show-ref', '--verify', '--quiet', ref], scope);
     this.#throwIfCancelled(existing, scope);
@@ -260,10 +263,10 @@ export class GitRebaser {
 
   /** Startup/live recovery for an attempt whose Store marker still exists. */
   async abort(attemptId: string, resultHead?: string,
-    resultState: 'none' | 'prepared' | 'uncertain' | 'refused' | 'ready' = 'ready'): Promise<void> {
+    resultState: 'none' | 'prepared' | 'uncertain' | 'refused' | 'ready' = 'ready', timeoutMs?: number): Promise<void> {
     const ref = rebaseRef(attemptId), path = this.#path(attemptId), stat = lstatSync(path, { throwIfNoEntry: false });
     if (resultHead !== undefined && !COMMIT_ID.test(resultHead)) throw new Error('A full retained result ID is required.');
-    const scope = this.#scope(attemptId);
+    const scope = this.#scope(attemptId, undefined, timeoutMs, true);
     if (stat && (!stat.isDirectory() || stat.isSymbolicLink())) throw new Error('The rebase workspace is not a plain directory.');
     await this.#remove(path, scope);
     if (resultState === 'refused') return;
@@ -548,8 +551,13 @@ export class GitRebaser {
     return outcome;
   }
 
-  #scope(attemptId: string, signal?: AbortSignal): CallScope {
-    const timeout = this.#options.timeoutMs ?? 120_000, deadline = performance.now() + timeout;
+  #scope(attemptId: string, signal?: AbortSignal, requestedTimeoutMs?: number, cleanupOnly = false): CallScope {
+    const configured = this.#options.timeoutMs ?? MAX_REBASE_TIMEOUT_MS;
+    const timeout = requestedTimeoutMs === undefined ? configured : Math.min(requestedTimeoutMs, configured);
+    const minimum = cleanupOnly ? MIN_REBASE_CLEANUP_TIMEOUT_MS : MIN_REBASE_TIMEOUT_MS;
+    if (!Number.isSafeInteger(timeout) || timeout < minimum || timeout > MAX_REBASE_TIMEOUT_MS)
+      throw new Error('Invalid rebase deadline.');
+    const deadline = performance.now() + timeout;
     return { attemptId, signal, deadline, workDeadline: deadline - CLEANUP_RESERVE_MS };
   }
 

test/merge-task-pr.test.ts

@@ -188,6 +188,22 @@ describe('the merge gate with a runner block (#121)', () => {
     expect(store.getMergeAttempt(identity)).toMatchObject({ queueWatermark: 'CURSOR_after_authorization' });
   });
 
+  it('revalidates the pull request after capturing the queue event boundary', async () => {
+    const { store } = published();
+    let remoteBase = oid(1), watermarks = 0;
+    const gh = github(store, () => ({ base: remoteBase, mergeQueue: true }));
+    gh.client.queueWatermark = vi.fn(async () => {
+      if (++watermarks === 2) remoteBase = oid(3);
+      return 'CURSOR';
+    });
+    prepare(store);
+    const merges = new MergeCoordinator(service(store), gh.client, undefined, undefined, PREPARED_PUBLISHED);
+
+    await expect(merges.merge('review-token')).rejects.toThrow(/pull request|requirements changed/i);
+    expect(gh.merged).toEqual([]);
+    expect(store.getMergeAttempt(identity)).toBeNull();
+  });
+
   it('revalidates local authorization after the final queue event boundary', async () => {
     const { store } = published(), gh = github(store, () => ({ mergeQueue: true }));
     let authorized = false, revoked = false;

test/pre-merge.test.ts

@@ -12,7 +12,7 @@ import { commandCheckDeps } from '../runner/checks.ts';
 import { RunnerCoordinator } from '../runner/coordinator.ts';
 import type { InvocationHandle, InvocationResult } from '../agents/contract.ts';
 import type { TaskWorkspace, WorkspaceRef } from '../runner/execution.ts';
-import { RebaseResourcesUnsettled } from '../runner/rebase.ts';
+import { MIN_REBASE_TIMEOUT_MS, RebaseResourcesUnsettled } from '../runner/rebase.ts';
 
 const roots: string[] = [], services: ReviewService[] = [];
 vi.setConfig({ testTimeout: 15_000 });
@@ -49,6 +49,7 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options:
   commandWaitsForCancellation?: boolean;
   releaseFails?: boolean;
   unsettledRebase?: boolean;
+  rebaseFailure?: Error;
   runnerCommitted?: boolean;
   operationTimeoutMs?: number;
   authorize?: (signal: AbortSignal) => Promise<PreMergeAuthorization>;
@@ -85,17 +86,22 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options:
   view = service.load(); view = service.act({ action: 'approve', item: 'P1', token: view.token });
   let coordinator!: PreMergeCoordinator, rebaseRuns = 0;
   let rebaseAborts = 0;
-  const rebaser = { run: async (input: { attemptId: string; signal: AbortSignal }) => {
+  const rebaseBudgets: Array<number | undefined> = [], abortBudgets: Array<number | undefined> = [];
+  const rebaser = { run: async (input: { attemptId: string; signal: AbortSignal; timeoutMs?: number }) => {
     rebaseRuns++;
+    rebaseBudgets.push(input.timeoutMs);
     options.duringRebase?.(service, coordinator);
     input.signal.throwIfAborted();
     if (options.unsettledRebase) throw new RebaseResourcesUnsettled('Conflict resources remain owned.');
+    if (options.rebaseFailure) throw options.rebaseFailure;
     const planKey = service.store.getTask(identity).planKey;
     service.store.prepareRebaseResult(planKey, input.attemptId, rebased, [rebased]);
     service.store.completeRebaseResult(planKey, input.attemptId);
     return { oldHead: head, base: onto, head: rebased,
       mappings: [{ oldSha: head, newSha: rebased }], resolvedConflicts: [] };
-  }, abort: async () => { rebaseAborts++; } } as never;
+  }, abort: async (_attemptId: string, _head?: string, _state?: string, timeoutMs?: number) => {
+    rebaseAborts++; abortBudgets.push(timeoutMs);
+  } } as never;
   const checkedHeads: string[] = [];
   const workspace: TaskWorkspace = {
     async materialize(attempt, checkedHead) {
@@ -155,7 +161,7 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options:
   const result = await coordinator.start({ stateVersion: task.stateVersion,
     reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head });
   return { service, coordinator, runner, result, rebased, checkedHeads, cancelReasons,
-    rebaseRuns: () => rebaseRuns, rebaseAborts: () => rebaseAborts };
+    rebaseRuns: () => rebaseRuns, rebaseAborts: () => rebaseAborts, rebaseBudgets, abortBudgets };
 }
 
 it('preserves an approval when a moved base leaves its attributed fingerprint unchanged', async () => {
@@ -205,7 +211,8 @@ it('revalidates authorization before starting the delayed local rebase', async (
   expect(fixture.result).toMatchObject({ state: 'failed', reason: 'Issue trust was revoked.' });
   expect(calls).toEqual(['read', 'validate']);
   expect(fixture.rebaseRuns()).toBe(0);
-  expect(fixture.rebaseAborts()).toBe(1);
+  // Refused authorization precedes the first durable rebase claim, so there is no cleanup lifecycle to start.
+  expect(fixture.rebaseAborts()).toBe(0);
   expect(fixture.service.store.getTask(fixture.service.config.identity).rebaseInProgress).toBeNull();
   expect(fixture.coordinator.last).toMatchObject({ state: 'failed', stale: false });
   await fixture.coordinator.close(); await fixture.runner.close();
@@ -365,7 +372,8 @@ it('preserves shutdown as the reason that stops an active command check', async
 
 it('stops an active command check when the operation-wide deadline expires', async () => {
   const fixture = await rebaseFixture('feature\n', 0,
-    { commandWaitsForCancellation: true, operationTimeoutMs: 3_000, reserves: { processMs: 1_000, commandMs: 200 } });
+    { commandWaitsForCancellation: true, operationTimeoutMs: MIN_REBASE_TIMEOUT_MS + 3_000,
+      reserves: { processMs: MIN_REBASE_TIMEOUT_MS + 1_000, commandMs: 200 } });
   expect(fixture.result).toMatchObject({ state: 'failed', reason: 'Pre-merge preparation deadline exceeded.' });
   const check = fixture.service.store.getAttempts(fixture.service.config.identity).find(attempt => attempt.kind === 'check');
   expect(check).toMatchObject({ state: 'failed', firstReason: null, stopReason: 'timeout', diagnostic: 'Timed out.' });
@@ -523,6 +531,20 @@ it('aborts remote work early enough to reserve bounded subprocess settlement', a
   await coordinator.close();
 });
 
+it('budgets live rebase work and its follow-up cleanup inside the preparation deadline', async () => {
+  const operationTimeoutMs = 60_000;
+  const fixture = await rebaseFixture('feature\n', undefined,
+    { operationTimeoutMs, rebaseFailure: new Error('rebase failed') });
+  expect(fixture.result).toMatchObject({ state: 'failed', reason: 'rebase failed' });
+  expect(fixture.rebaseBudgets).toHaveLength(1);
+  expect(fixture.abortBudgets).toHaveLength(1);
+  for (const budget of [...fixture.rebaseBudgets, ...fixture.abortBudgets]) {
+    expect(budget).toBeTypeOf('number');
+    expect(budget).toBeGreaterThan(0);
+    expect(budget).toBeLessThanOrEqual(operationTimeoutMs);
+  }
+});
+
 it('makes a preparation action resendable when its terminal storage write fails', async () => {
   const root = mkdtempSync(join(tmpdir(), 'codeboost-pre-merge-settlement-failure-')); roots.push(root);
   const config = createDemo(join(root, 'demo')), service = new ReviewService(config); services.push(service);
@@ -582,7 +604,7 @@ it('settles a failed preparation even when the fallback snapshot read would fail
 });
 
 it('threads the remaining operation budget through every synchronous review reload', async () => {
-  const operationTimeoutMs = 30_000, processMs = 10_000;
+  const operationTimeoutMs = MIN_REBASE_TIMEOUT_MS + 10_000, processMs = 10_000;
   const fixture = await rebaseFixture('feature\n', undefined, { operationTimeoutMs, reserves: { processMs, commandMs: 0 } });
   const view = fixture.service.load(), task = fixture.service.store.getTask(fixture.service.config.identity);
   const original = fixture.service.load.bind(fixture.service); const budgets: number[] = [];