ship-queue wave: review fixes — stamped wave re-READY, constituent double-land, drop dry-run, settle names hand-resolved merges, one settled row per PR, ready refuses an unreadable PR

Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

lib/ship-queue-train-ready.ts

@@ -194,7 +194,8 @@ export async function trainReady(ctx: QueueContext, o: ReadyOptions): Promise<Tr
   const kind = classifyTrainHead({ repoRoot: ctx.repoRoot, baseRef: ctx.base, head });
   if (kind.kind === 'invalid') return outRefused({ code: 'QUEUE_NOT_READY', message: `the Version Gate refuses ${head.slice(0, 12)}: ${kind.reason}` }, base.lines, `fix the head, push, then ${Q} ready --pr ${o.pr} --screen|--no-screen`, 'none');
   const info = prInfo(ctx, o.pr);
-  const wave = 'error' in info ? null : checkWavePin(env, o.pr, info);
+  if ('error' in info) return outRefused({ code: 'QUEUE_PROOF_MISSING', message: info.error }, base.lines, `${Q} ready --pr ${o.pr} --screen|--no-screen once gh can read the PR`, 'none');
+  const wave = checkWavePin(env, o.pr, info);
   if (wave && 'failure' in wave) return outRefused(wave.failure, base.lines, `${Q} wave status (rebuild the wave without a moved constituent: ${Q} wave build --wave <nonce> --drop <pr>)`, 'none');
   const sel = await screenSelection(env, head, baseSha);
   const lines = [...base.lines, ...(wave ? wave.lines : []), ...sel.lines];

lib/ship-queue-train-record.ts

@@ -75,7 +75,11 @@ export interface FreezeRecord {
   repair: { pr: string; head: string; train: string | null } | null; created_at: string; cleared_at: string | null; cleared_by: string | null; updated_at: string;
 }
 /** A constituent of a wave PR (lib/ship-queue-wave.ts): the PR, its pinned head and the wave-branch merge commit whose second parent is that head. */
-export interface WaveConstituent { pr: string; head: string; merge_commit: string | null }
+export interface WaveConstituent {
+  pr: string; head: string; merge_commit: string | null;
+  /** Files the integrator resolved by hand in this constituent's merge (wave build --continue); absent for a clean merge. */
+  resolved?: string[];
+}
 export type WaveState = 'conflict' | 'built' | 'pushed' | 'superseded' | 'settled';
 export interface WaveRecord {
   schema: 1; kind: 'wave'; nonce: string; repo_key: string; area: string; base: { ref: string; sha: string }; branch: string;

lib/ship-queue-wave.ts

@@ -121,9 +121,10 @@ export interface WaveChain { integrator: string[]; merges: string[]; stamp: stri
  * Walk the wave branch's first parents from tip back to the recorded base. Every constituent
  * must appear, in order and exactly once, as the second parent of a two-parent commit; a
  * one-parent commit is an integrator commit; any other merge (main merged in, an unknown head)
- * is refused. With allowStamp the tip may be train land's stamp commit (parents [pin, main]).
+ * is refused. With stampBase the tip may be train land's stamp commit: parents [wave head, a
+ * commit already on stampBase], after every constituent.
  */
-export function waveChain(repoRoot: string, w: WaveRecord, tip: string, allowStamp = false): WaveChain | { error: string } {
+export function waveChain(repoRoot: string, w: WaveRecord, tip: string, stampBase: string | null = null): WaveChain | { error: string } {
   const r = git(repoRoot, ['rev-list', '--first-parent', '--parents', `${w.base.sha}..${tip}`]);
   if (!r.ok) return { error: `cannot walk ${short(tip)} back to the wave base ${short(w.base.sha)} (${r.err.slice(0, 120)})` };
   const rows = r.out.split('\n').filter(Boolean).map(l => l.split(' ')).reverse();
@@ -134,7 +135,7 @@ export function waveChain(repoRoot: string, w: WaveRecord, tip: string, allowSta
     if (!second) { out.integrator.push(commit!); continue; }
     const want = w.constituents[next];
     if (!more.length && want && second === want.head) { out.merges.push(commit!); next++; continue; }
-    if (!more.length && allowStamp && i === rows.length - 1 && next === w.constituents.length) { out.stamp = commit!; continue; }
+    if (!more.length && stampBase && i === rows.length - 1 && next === w.constituents.length && git(repoRoot, ['merge-base', '--is-ancestor', second, stampBase]).ok) { out.stamp = commit!; continue; }
     const expected = want ? `#${want.pr}'s pin ${short(want.head)}` : 'a constituent';
     return { error: `${short(commit!)} merges ${short(second)}, which is not ${expected} (only constituent merges and integrator commits belong on a wave branch)` };
   }
@@ -163,16 +164,20 @@ function constituentProblem(ctx: QueueContext, pr: string, pin: string): { failu
 /**
  * ready and train build: a PR whose head branch is wave/<nonce> must have a readable wave
  * record (refused QUEUE_WAVE_UNKNOWN otherwise, never skipped), a branch chain that merges every
- * pin unchanged, and constituents still open at their pins. Null for any other PR.
+ * pin unchanged, and constituents still open at their pins. Any other PR is null, unless a live
+ * wave lists it as a constituent: then it is refused QUEUE_WAVE_DUPLICATE so it cannot land twice.
  */
 export function checkWavePin(env: TrainEnv, pr: string, info: PrInfo): { wave: WaveRecord; lines: string[] } | { failure: QueueFailure } | null {
-  if (!info.headRefName.startsWith('wave/')) return null;
+  if (!info.headRefName.startsWith('wave/')) {
+    const holder = listWaves(env.repo.keys).find(x => LIVE.has(x.state) && x.constituents.some(c => c.pr === pr));
+    return holder ? { failure: fail('QUEUE_WAVE_DUPLICATE', `#${pr} is a constituent of live wave ${holder.nonce}; it lands through the wave PR, not on its own`) } : null;
+  }
   let w = waveByBranch(env, info.headRefName);
   if (!w) { rebuildFromNotes(env.ctx.repoRoot, env.remote); w = waveByBranch(env, info.headRefName); }
   if (!w) return { failure: fail('QUEUE_WAVE_UNKNOWN', `#${pr}'s branch ${info.headRefName} has no wave record in local state or refs/notes/gstack-trains`) };
   if (w.state === 'superseded') return { failure: fail('QUEUE_WAVE_MOVED', `wave ${w.nonce} was rebuilt as ${w.superseded_by}; READY the new wave PR`) };
   if (!ensureCommit(env, info.headRefOid, pr)) return { failure: fail('QUEUE_PROOF_MISSING', `cannot fetch #${pr}'s head ${short(info.headRefOid)}`) };
-  const chain = waveChain(env.ctx.repoRoot, w, info.headRefOid);
+  const chain = waveChain(env.ctx.repoRoot, w, info.headRefOid, `${env.remote}/${env.ctx.base}`);
   if ('error' in chain) return { failure: fail('QUEUE_WAVE_MOVED', `wave ${w.nonce}: ${chain.error}`) };
   for (const c of w.constituents) {
     const p = constituentProblem(env.ctx, c.pr, c.head);
@@ -401,6 +406,7 @@ function continueWave(env: TrainEnv, w: WaveRecord, dryRun: boolean): TrainOut {
   const problem = w.constituents.map(x => constituentProblem(env.ctx, x.pr, x.head)).find(Boolean);
   if (problem) return outRefused(problem.failure, [], `${Q} wave build --wave ${w.nonce} --drop <pr>`, wavePath(w.nonce));
   c.merge_commit = parents[0]!;
+  c.resolved = w.conflict.files;
   const lines = [`RESOLVED: #${pr} -> ${short(parents[0]!)} (integrator resolution of ${w.conflict.files.join(', ')})`];
   if (!dryRun) { git(env.ctx.repoRoot, ['worktree', 'remove', '--force', worktree]); w.conflict = null; }
   return chainFrom(env, w, idx + 1, parents[0]!, lines, dryRun, null);
@@ -413,7 +419,7 @@ function dropFrom(env: TrainEnv, w: WaveRecord, o: WaveBuildOptions): TrainOut {
   const rest = w.constituents.filter(c => c !== dropped);
   const carrier = rest.find(c => git(env.ctx.repoRoot, ['merge-base', '--is-ancestor', dropped.head, c.head]).ok);
   if (carrier) return outRefused(fail('QUEUE_WAVE_DEPENDENT', `#${carrier.pr}'s pin contains #${dropped.pr}'s pin, so dropping #${dropped.pr} would still land its code`), [], `drop #${carrier.pr} too, or keep #${dropped.pr}`, wavePath(w.nonce));
-  const out = newWave(env, { ...o, area: w.area }, o.dryRun ? null : w, rest.map(c => ({ pr: c.pr, sha: c.head })));
+  const out = newWave(env, { ...o, area: w.area }, w, rest.map(c => ({ pr: c.pr, sha: c.head })));
   out.lines.unshift(`DROP: #${dropped.pr} from ${w.nonce}; the rebuilt wave gets a new branch and PR, and the integrator closes ${w.branch}'s PR`);
   return out;
 }
@@ -466,7 +472,7 @@ export function waveSettle(ctx: QueueContext, o: { wave: string; dryRun: boolean
   if (git(repo, ['rev-parse', `${mergeSha}^{tree}`]).out !== git(repo, ['rev-parse', `${merged.headRefOid}^{tree}`]).out) {
     return outRefused(fail('QUEUE_WAVE_NOT_MERGED', `main's merge ${short(mergeSha)} does not carry #${merged.number}'s head tree; what landed is not the wave`), [], 'none', wavePath(w.nonce));
   }
-  const chain = waveChain(repo, w, merged.headRefOid, true);
+  const chain = waveChain(repo, w, merged.headRefOid, mergeSha);
   if ('error' in chain) return outRefused(fail('QUEUE_WAVE_MOVED', `the merged head ${short(merged.headRefOid)}: ${chain.error}`), [], 'none', wavePath(w.nonce));
   const at = versionAtRef(repo, mergeSha);
   const version = 'version' in at ? `v${at.version}` : 'its version';
@@ -478,18 +484,24 @@ export function waveSettle(ctx: QueueContext, o: { wave: string; dryRun: boolean
     if ('error' in info) { incomplete++; lines.push(`SETTLED: #${c.pr} unreadable (${info.error}); rerun settle`); continue; }
     if (info.state !== 'OPEN') { lines.push(`SETTLED: #${c.pr} skipped (${info.state.toLowerCase()})`); continue; }
     const own = new Set(changed(env, c.head)?.files ?? []);
-    const touched = chain.integrator.filter(x => integratorFiles.get(x)!.some(f => own.has(f))).map(short);
+    const touched = [
+      ...w.constituents.flatMap((x, i) => {
+        const why = chain.merges[i] !== x.merge_commit ? 'not the recorded merge' : x.resolved ? `resolved by hand: ${x.resolved.join(', ')}` : null;
+        return why && (x === c || x.resolved?.some(f => own.has(f))) ? [`${short(chain.merges[i]!)} (merge of #${x.pr}, ${why})`] : [];
+      }),
+      ...chain.integrator.filter(x => integratorFiles.get(x)!.some(f => own.has(f))).map(short),
+    ];
     const moved = info.headRefOid !== c.head;
     const body = moved
       ? `Wave #${merged.number} landed this PR's pin ${short(c.head)} as ${version} (${short(mergeSha)}). The head moved to ${short(info.headRefOid)} after the wave froze, so that newer work did not land. Left open for its owner.`
-      : `Landed in wave #${merged.number} as ${version} (${short(mergeSha)}), merged at this PR's pin ${short(c.head)}. Integrator commits touching this PR's files: ${touched.length ? touched.join(', ') : 'none'}.`;
+      : `Landed in wave #${merged.number} as ${version} (${short(mergeSha)}), merged from this PR's pin ${short(c.head)}. Integrator changes touching this PR's files: ${touched.length ? touched.join(', ') : 'none'}.`;
     if (o.dryRun) { lines.push(`SETTLED: #${c.pr} would be ${moved ? 'commented, left open' : 'commented and closed'} (dry run)`); continue; }
     const posted = commentOnce(ctx, c.pr, marker(w.nonce, moved ? 'moved' : 'landed'), body);
     const closed = posted !== 'failed' && !moved ? ctx.gh(['pr', 'close', c.pr]).ok : false;
     const outcome = posted === 'failed' || (!moved && !closed) ? 'incomplete' : moved ? 'moved-left-open' : 'landed-closed';
     if (outcome === 'incomplete') incomplete++;
     lines.push(`SETTLED: #${c.pr} ${outcome} (comment ${posted})`);
-    w.settled.push({ pr: c.pr, outcome, at: nowIso() });
+    w.settled = [...w.settled.filter(s => s.pr !== c.pr), { pr: c.pr, outcome, at: nowIso() }];
   }
   if (o.dryRun) return outOk([...lines, 'DRY-RUN: nothing commented or closed'], `${Q} wave settle --wave ${w.nonce}`, wavePath(w.nonce));
   if (!incomplete) w.state = 'settled';

test/ship-queue-wave.test.ts

@@ -29,8 +29,8 @@ function threePrs(fx: TrainFixture) {
 }
 
 /** A wave PR #10 whose head branch is the wave branch, with a /ship receipt and a READY receipt at its head. */
-function waveAsPr(fx: TrainFixture, nonce: string, pr = '10'): string {
-  const head = fx.repo.git('ls-remote', 'origin', `refs/heads/wave/${nonce}`).split(/\s/)[0]!;
+function waveAsPr(fx: TrainFixture, nonce: string, pr = '10', tip?: string): string {
+  const head = tip ?? fx.repo.git('ls-remote', 'origin', `refs/heads/wave/${nonce}`).split(/\s/)[0]!;
   fx.repo.git('push', '-q', '-f', 'origin', `${head}:refs/pull/${pr}/head`);
   fx.write(d => { d.prs[pr] = { headRefOid: head, isCrossRepository: false, state: 'OPEN', headRefName: `wave/${nonce}`, url: `https://github.com/fixture/repo/pull/${pr}`, body: shippedBody(pr, head) }; });
   readyReceipt(fx, pr, head);
@@ -125,7 +125,16 @@ describe('wave build', () => {
     const status = queue(fx, ['wave', 'status', '--wave', nonce]);
     expect(line(status.out, 'CHAIN:')).toContain('ok at');
     expect(line(status.out, 'CHAIN:')).toContain('3 constituent merges');
-  }, 120_000);
+    const head = waveAsPr(fx, nonce);
+    const merge = fx.repo.git('commit-tree', `${head}^{tree}`, '-p', 'origin/main', '-m', 'wave (#10)');
+    fx.repo.git('push', '-q', 'origin', `${merge}:refs/heads/main`);
+    fx.write(d => { Object.assign(d.prs['10'] as any, { state: 'MERGED', mergeCommit: { oid: merge } }); });
+    expect(queue(fx, ['wave', 'settle', '--wave', nonce]).status).toBe(EXIT.ok);
+    const prs = fx.read().prs as Record<string, any>;
+    expect(prs['3'].comments.at(-1).body).toContain('(merge of #3, resolved by hand: lib/ship-shared.ts)');
+    expect(prs['2'].comments.at(-1).body).toContain('(merge of #3, resolved by hand: lib/ship-shared.ts)');
+    expect(prs['1'].comments.at(-1).body).toContain("Integrator changes touching this PR's files: none.");
+  }, 180_000);
 
   test('--drop builds a new wave on a new branch and supersedes the old one; a dropped pin another pin contains is refused', () => {
     const fx = make();
@@ -134,6 +143,9 @@ describe('wave build', () => {
     const first = queue(fx, ['wave', 'build', '--prs', [...heads, h5].map((h, i) => `${i + 1}@${h}`).join(',')]);
     expect(first.status).toBe(EXIT.ok);
     const old = nonceOf(first.out);
+    const dry = queue(fx, ['wave', 'build', '--wave', old, '--drop', '5', '--dry-run']);
+    expect([dry.status, dry.out]).toEqual([EXIT.ok, expect.stringContaining('DRY-RUN: no record written')]);
+    expect(waveJson(fx, old).state).toBe('pushed');
     const dependent = queue(fx, ['wave', 'build', '--wave', old, '--drop', '4']);
     expect([dependent.status, dependent.err]).toEqual([EXIT.refused, expect.stringContaining('(QUEUE_WAVE_DEPENDENT')]);
     const dropped = queue(fx, ['wave', 'build', '--wave', old, '--drop', '5']);
@@ -160,6 +172,15 @@ describe('ready and train build read the wave', () => {
     readyReceipt(fx, '1', heads[0]!);
     const dup = queue(fx, ['train', 'build', '--prs', `10@${head},1@${heads[0]}`, '--dry-run']);
     expect([dup.status, dup.err]).toEqual([EXIT.refused, expect.stringContaining('(QUEUE_WAVE_DUPLICATE')]);
+    const alone = queue(fx, ['train', 'build', '--prs', `1@${heads[0]}`, '--dry-run']);
+    expect([alone.status, alone.err]).toEqual([EXIT.refused, expect.stringContaining(`#1 is a constituent of live wave ${nonce}`)]);
+    const stamp = fx.repo.git('commit-tree', `${head}^{tree}`, '-p', head, '-p', 'origin/main', '-m', 'stamp');
+    const stamped = queue(fx, ['train', 'build', '--prs', `10@${waveAsPr(fx, nonce, '10', stamp)}`, '--dry-run']);
+    expect([stamped.status, line(stamped.out, 'WAVE:')]).toEqual([EXIT.ok, expect.stringContaining(`WAVE: ${nonce}`)]);
+    const offBase = fx.repo.git('commit-tree', `${head}^{tree}`, '-p', head, '-p', heads[0]!, '-m', 'not a stamp');
+    const foreign = queue(fx, ['train', 'build', '--prs', `10@${waveAsPr(fx, nonce, '10', offBase)}`, '--dry-run']);
+    expect([foreign.status, foreign.err]).toEqual([EXIT.refused, expect.stringContaining('(QUEUE_WAVE_MOVED')]);
+    waveAsPr(fx, nonce);
     fx.write(d => { (d.prs['2'] as any).headRefOid = heads[0]; });
     const moved = queue(fx, ['train', 'build', '--prs', `10@${head}`, '--dry-run']);
     expect([moved.status, moved.err]).toEqual([EXIT.refused, expect.stringContaining('#2 moved')]);
@@ -233,6 +254,7 @@ describe('wave settle', () => {
     expect(again.out).toContain('SETTLED: #1 skipped (closed)');
     expect((fx.read().prs as any)['2'].comments.length).toBe(count);
     expect(waveJson(fx, nonce).state).toBe('settled');
+    expect(waveJson(fx, nonce).settled.map((s: any) => s.pr).sort()).toEqual(['1', '2', '3']);
   }, 180_000);
 });