fix: pre-landing review fixes (train gates: dispatch bound, lease, rebase, post-merge checks, output tail, caps)

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

docs/ship-policy.md

@@ -451,7 +451,7 @@ prefix, not just its tip, so a red train names its culprit without a bisect:
 "train": {
   "gates": [
     { "kind": "workflows", "workflows": ["ci.yml"], "required_jobs": ["ci.yml/test", "ci.yml/lint"] },
-    { "kind": "command", "command": "bun test", "timeout_sec": 1800, "spend_cap_usd": 2 }
+    { "kind": "command", "command": "bun test", "timeout_sec": 1800, "spend_cap_usd": 3 }
   ],
   "land": { "merge_method": "squash", "required_checks": ["test", "lint"] },
   "max_size": 5, "fill_hours": 3
@@ -477,8 +477,9 @@ prefixes.
   JSON file reporting at least one executed test, and failed test ids become
   the red cases. A red run gets one `gh run rerun --failed`: green on the
   rerun is GREEN with a `FLAKY:` line, red again is real, and both attempts'
-  jobs stay in the record. `post_merge_checks` names checks to wait for on
-  main after each land (`verify --post-merge`). A workflow whose concurrency
+  jobs stay in the record. `post_merge_checks` names checks that must succeed
+  on main after each land before the next slot lands (`verify --post-merge
+  <sha> --checks <names>`; `train watch --land` runs it between slots). A workflow whose concurrency
   group is not keyed by `github.ref` (or the nonce) gets one prefix at a
   time.
 - **`command` gates.** Each prefix is checked out clean and detached and runs
@@ -487,8 +488,10 @@ prefixes.
   the prefix tree, and exit 0 is GREEN. Only names in `pass_env` reach the VM
   (`UBICLOUD_*` never). Each run reserves its timeout plus 10 minutes at the
   list price, and one `spend_cap_usd` covers every run of the train, reruns
-  included; the next run past the cap refuses (`QUEUE_TRAIN_SPEND_CAP`) and
-  `train build --train <nonce>` resumes once the cap is raised.
+  included (3 dollars covers six prefixes and one rerun at 1800 s; `train init`
+  prints that default and `train doctor` flags a cap below a full train). The
+  next run past the cap refuses (`QUEUE_TRAIN_SPEND_CAP`); raising the cap is a
+  policy commit that moves main, so it is followed by a new train.
 
 **The verdict.** T0 red is red on main (`QUEUE_TRAIN_RED_ON_MAIN`, fix
 forward). Otherwise the first non-green prefix decides: incomplete stops the
@@ -522,7 +525,8 @@ commit when there is one, before its first remote effect:
    lease, its checks waited for, then merged at that commit. It is forced for
    a stamped train, strict or unreadable protection, a conditional PR, a merge
    that resolved release files, and a pin whose checks are red. `rebase`
-   refuses here (`QUEUE_TRAIN_LAND_METHOD`), since it would replay commits.
+   always refuses (`QUEUE_TRAIN_LAND_METHOD`): it would replay commits, and
+   the replayed tree is not the gated one.
 
 Required checks are `changelog_fragments.required_checks`, else
 `train.land.required_checks`. Fork heads refuse. Either way main's tree must

docs/troubleshooting.md

@@ -4529,6 +4529,12 @@ GitHub accepted the dispatch). It is never dispatched again blindly.
 run carrying the nonce, or marks the dispatch absent; only then does
 `train build --train <nonce>` dispatch it again, under the same nonce.
 
+On a train with `gates`, a workflow is sent at most twice per prefix. When
+neither send shows a run carrying the nonce, `train build` (and `train watch`)
+refuses with this code instead of sending a third: run `train doctor` (a
+run-name without the nonce is the usual cause), fix the workflow on the base
+branch, then build a new train.
+
 <a id="queue-train-waiting"></a>
 ### `QUEUE_TRAIN_WAITING`
 
@@ -4685,8 +4691,10 @@ again by its lane ([ship-policy.md](ship-policy.md#train-gates)).
 spend cap, or `UBI_OWNER` is not set. Nothing more ran; finished runs keep
 their outcomes.
 
-**Fix.** Set `UBI_OWNER` to your thread code, or raise `spend_cap_usd` (or
-lower `timeout_sec`) on the base branch, then `train build --train <nonce>`.
+**Fix.** Set `UBI_OWNER` to your thread code and rerun `train build --train
+<nonce>`. For the cap, raise `spend_cap_usd` (or lower `timeout_sec`) on the
+base branch, then build a new train on the new main: the policy commit moves
+main past T0.
 
 <a id="queue-train-land-method"></a>
 ### `QUEUE_TRAIN_LAND_METHOD`
@@ -4697,7 +4705,8 @@ strict protection, conditional PR, or release-file resolution), and
 landing the gated tree.
 
 **Fix.** Set `merge_method` to `squash` or `merge` on the base branch, then
-land again.
+build a new train on the new main: the policy commit moves main off this
+train's prefixes.
 
 <a id="queue-train-init-refused"></a>
 ### `QUEUE_TRAIN_INIT_REFUSED`

lib/result-codes-release.ts

@@ -145,12 +145,12 @@ export const RELEASE_RESULT_CODES = {
   QUEUE_TRAIN_SPEND_CAP: {
     anchor: 'queue-train-spend-cap',
     summary: 'a command gate\'s next Ubicloud run would pass the one spend cap that covers every run of the train, or UBI_OWNER is not set, so nothing more ran',
-    fix: 'set UBI_OWNER to your thread code, or raise the gate\'s spend_cap_usd (or lower its timeout_sec) on the base branch, then train build --train <nonce> runs what is left',
+    fix: 'set UBI_OWNER to your thread code and train build --train <nonce> runs what is left; for the cap, raise the gate\'s spend_cap_usd (or lower its timeout_sec) on the base branch, then build a new train on the new main (the policy commit moves main past T0)',
   },
   QUEUE_TRAIN_LAND_METHOD: {
     anchor: 'queue-train-land-method',
     summary: 'train land must publish this slot through a tool-made update commit (a stamped train, strict branch protection, a conditional PR, or a merge that resolved release files), and the policy\'s merge method is rebase, which would replay commits instead of landing the gated tree',
-    fix: 'set train.land.merge_method to squash or merge on the base branch (history merge-only allows merge only), then land again',
+    fix: 'set train.land.merge_method to squash or merge on the base branch (history merge-only allows merge only), then build a new train on the new main (the policy commit moves main off this train\'s prefixes)',
   },
   QUEUE_TRAIN_INIT_REFUSED: {
     anchor: 'queue-train-init-refused',

lib/ship-policy.ts

@@ -443,7 +443,7 @@ function trainErrors(value: unknown, obj: Record<string, unknown>, out: PolicyEr
   }
   if ('land' in t) landErrors(t.land, obj, out);
   const method = (t.land as { merge_method?: unknown } | undefined)?.merge_method ?? 'squash';
-  if (obj.history === 'merge-only' && method !== 'merge') out.push({ code: 'POLICY_INVALID', path: 'train.land.merge_method', message: `history merge-only allows only the merge method; train land would use ${String(method)}` });
+  if ('land' in t && obj.history === 'merge-only' && method !== 'merge') out.push({ code: 'POLICY_INVALID', path: 'train.land.merge_method', message: `history merge-only allows only the merge method; train land would use ${String(method)}` });
 }
 
 /**

lib/ship-queue-post-merge.ts

@@ -18,18 +18,24 @@ import { outFailed, outOk, outPending, type TrainOut } from './ship-queue-train'
 const Q = 'gstack-ship-queue';
 const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
 
-export interface PostMergeOptions { sha: string; timeoutSec: number; pollMs: number }
+export interface PostMergeOptions { sha: string; timeoutSec: number; pollMs: number; checks?: string[] }
 
 export async function verifyPostMerge(ctx: QueueContext, o: PostMergeOptions): Promise<TrainOut> {
   const deadline = Date.now() + o.timeoutSec * 1000;
   const at = o.sha.slice(0, 12);
-  const again = `${Q} verify --post-merge ${o.sha} --timeout ${o.timeoutSec}`;
+  const named = o.checks ?? [];
+  const again = `${Q} verify --post-merge ${o.sha} --timeout ${o.timeoutSec}${named.length ? ` --checks ${JSON.stringify(named.join(','))}` : ''}`;
   for (;;) {
     const ci = ciAtHead(ctx, o.sha);
     const lines = [`POST_MERGE: ${ci.line.replace(/^(\S+) at ([0-9a-f]+)/, '$2 $1')}`, ...ci.checks.map(c => `CHECK: ${c.name} ${c.status === 'completed' ? c.conclusion ?? 'unknown' : c.status}`)];
     const settled = !ci.none && !/^(pending|unknown) /.test(ci.line) && !ci.checks.some(c => c.status !== 'completed');
-    if (settled && ci.green) return outOk(lines, 'none', 'none');
-    if (settled) {
+    const runs = (n: string) => ci.checks.filter(c => c.name === n || c.name.startsWith(`${n} (`));
+    const absent = named.filter(n => !runs(n).length);
+    const unsuccessful = named.filter(n => runs(n).some(c => c.status === 'completed' && c.conclusion !== 'success'));
+    if (settled && ci.green && unsuccessful.length) return outFailed({ code: 'QUEUE_POST_MERGE_RED', message: `the post-merge checks the train gate names did not succeed at ${at}: ${unsuccessful.join(', ')}` }, lines,
+      `read the run, repair on a new PR or roll back; ${Q} train status names the freeze if the backstop filed one`, 'none');
+    if (settled && ci.green && !absent.length) return outOk(lines, 'none', 'none');
+    if (settled && !ci.green) {
       const red = ci.checks.filter(c => c.status === 'completed' && !['success', 'skipped', 'neutral'].includes(c.conclusion ?? '')).map(c => c.name);
       return outFailed({ code: 'QUEUE_POST_MERGE_RED', message: `main's push workflows at ${at} are red: ${red.join(', ') || 'a commit status (no check run failed)'}` }, lines,
         `read the red run, repair on a new PR or roll back; ${Q} train status names the freeze if the backstop filed one`, 'none');
@@ -37,7 +43,7 @@ export async function verifyPostMerge(ctx: QueueContext, o: PostMergeOptions): P
     if (Date.now() >= deadline) {
       const pending = ci.checks.filter(c => c.status !== 'completed').map(c => c.name);
       const message = ci.line.startsWith('unknown') ? `the check runs at ${at} are unreadable (${ci.line.replace(/^unknown at [0-9a-f]+ /, '')})`
-        : ci.none ? `no check run has attached to ${at} yet` : `main's push workflows at ${at} have not settled at the timeout: ${pending.join(', ') || 'a commit status'} still running`;
+        : ci.none ? `no check run has attached to ${at} yet` : settled ? `the post-merge checks the train gate names never ran at ${at}: ${absent.join(', ')}` : `main's push workflows at ${at} have not settled at the timeout: ${pending.join(', ') || 'a commit status'} still running`;
       return outPending({ code: 'QUEUE_TRAIN_PENDING', message }, lines, again, 'none');
     }
     await sleep(o.pollMs);

lib/ship-queue-train-cli.ts

@@ -67,7 +67,7 @@ export const TRAIN_HELP = `  train build --prs <n>@<sha>[,...] [--order <n>,...]
                                            (train policy) the merge check against a green train record
   verify --pr <n> --screen                 (train policy) the PR head's screen verdict read from its run: SCREEN_VERDICT and
                                            FLAKY lines, recorded in the READY receipt
-  verify --post-merge <sha> [--timeout <s>] after a merge: wait for every check run at the merged commit to settle;
+  verify --post-merge <sha> [--timeout <s>] [--checks <a,b>] after a merge: wait for every check run at the merged commit to settle;
                                            POST_MERGE: green|red|pending plus one CHECK: line per workflow (resumable)
   Merge trains need a "train" block in the base revision's ship policy (docs/ship-policy.md#run-a-merge-train).`;
 
@@ -128,8 +128,8 @@ export async function trainMain(ctx: QueueContext, args: string[]): Promise<numb
   }
   if (sub === 'init') {
     const timeoutSec = positive(flag(rest, '--timeout'), 1800);
-    const spendCap = positive(flag(rest, '--spend-cap'), 2);
-    if (Number.isNaN(timeoutSec) || Number.isNaN(spendCap)) return usage('--timeout needs seconds and --spend-cap dollars');
+    const spendCap = flag(rest, '--spend-cap') === undefined ? undefined : positive(flag(rest, '--spend-cap'), NaN);
+    if (Number.isNaN(timeoutSec) || (spendCap !== undefined && Number.isNaN(spendCap))) return usage('--timeout needs seconds and --spend-cap dollars');
     const command = flag(rest, '--command');
     if (rest.includes('--command') && !command?.trim()) return usage('--command needs the test command a command gate runs');
     return guarded('train init', async () => (await import('./ship-queue-train-init')).trainInit(ctx, { command, timeoutSec, spendCap }));
@@ -194,7 +194,8 @@ export async function postMergeMain(ctx: QueueContext, args: string[]): Promise<
   if (!sha || !/^[0-9a-f]{40}$/.test(sha)) return usage('verify --post-merge needs the merged commit\'s full 40-hex SHA');
   const timeoutSec = positive(flag(args, '--timeout'), 2700);
   if (!Number.isFinite(timeoutSec)) return usage('--timeout needs a positive number of seconds');
-  return guarded('verify', () => verifyPostMerge(ctx, { sha, timeoutSec, pollMs: Number(process.env.GSTACK_SHIP_QUEUE_POLL_MS) || 30_000 }));
+  const checks = flag(args, '--checks')?.split(',').map(s => s.trim()).filter(Boolean);
+  return guarded('verify', () => verifyPostMerge(ctx, { sha, timeoutSec, pollMs: Number(process.env.GSTACK_SHIP_QUEUE_POLL_MS) || 30_000, checks }));
 }
 
 export async function verifyMain(ctx: QueueContext, args: string[], pr: string): Promise<number> {

lib/ship-queue-train-gate.ts

@@ -20,6 +20,7 @@
 import { spawn } from 'node:child_process';
 import { randomBytes } from 'node:crypto';
 import * as fs from 'node:fs';
+import * as os from 'node:os';
 import * as path from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { awaitNonceRun, listDispatchRuns, makeNonce, runForNonce } from '../scripts/eval-bg';
@@ -31,7 +32,7 @@ import type { CommandGate, TrainGate, WorkflowsGate } from './ship-policy';
 import { git, type QueueFailure } from './ship-queue-gate';
 import { execFor, repoSlug } from './ship-queue-train-github';
 import {
-  attemptsOf, listTrains, nowIso, queueRoot, trainDir, trainFile,
+  attemptsOf, listTrains, loadTrain, nowIso, queueRoot, trainDir, trainFile,
   type AttemptRecord, type GateOutcome, type GatedPrefix, type GenericVerdict, type TrainPrefix, type TrainRecord, type TrainSlot,
 } from './ship-queue-train-record';
 import { effect, outFailed, outOk, outPending, outRefused, prsArgs, save, type TrainEnv, type TrainOut } from './ship-queue-train';
@@ -91,7 +92,7 @@ export function baseRedAdmission(slots: TrainSlot[], generic: boolean, explicit:
   for (const s of conditional) {
     if (s === fixer) return { code: 'QUEUE_TRAIN_BASE_RED', message: `#${s.pr} is conditional and declares FIXES_BASE_RED; a fixer's own head must be green` };
     if (!fixer || s.base_red!.fixer !== fixer.pr) return { code: 'QUEUE_TRAIN_BASE_RED', message: `#${s.pr} is conditional on fixer #${s.base_red!.fixer}, which is not in this train at slot 1 with FIXES_BASE_RED` };
-    if (Date.parse(s.base_red!.expires_at) < Date.parse(nowIso())) return { code: 'QUEUE_TRAIN_BASE_RED', message: `#${s.pr}'s base-red condition expired at ${s.base_red!.expires_at}; its lane classifies again` };
+    if (!(Date.parse(s.base_red!.expires_at) >= Date.parse(nowIso()))) return { code: 'QUEUE_TRAIN_BASE_RED', message: `#${s.pr}'s base-red condition expired at ${s.base_red!.expires_at}; its lane classifies again` };
   }
   return { slots: order, lines: [`BASE_RED: fixer #${fixer!.pr} at slot 1 (FIXES_BASE_RED: ${fixer!.fixes_base_red!.join(', ')}); conditional ${conditional.map(s => '#' + s.pr).join(', ') || '(none)'}; every prefix is gated with nothing excluded`] };
 }
@@ -172,7 +173,7 @@ export function parseTestRecord(files: Array<{ name: string; text: string }>): R
   return read ? { executed, failed: failed.sort() } : null;
 }
 
-/** Whether a workflow's concurrency groups key on the ref or the nonce; a global group serializes the prefix dispatches. */
+/** Whether a workflow's concurrency groups key on the ref, the run or the nonce; a global group (or one keyed only on github.head_ref, empty on a dispatch) serializes the prefix dispatches. */
 export function concurrencyKeyed(yaml: string): boolean {
   const groups = [...yaml.matchAll(/^\s*concurrency:[ \t]*(.*)$/gm)].map(m => {
     const inline = m[1]!.trim();
@@ -180,7 +181,7 @@ export function concurrencyKeyed(yaml: string): boolean {
     const after = yaml.slice(m.index! + m[0].length).split('\n').slice(1, 4).join('\n');
     return /^\s*group:\s*(.+)$/m.exec(after)?.[1] ?? '';
   });
-  return groups.every(g => /github\.(ref|head_ref|run_id)|inputs\.nonce/.test(g));
+  return groups.every(g => /github\.(ref|run_id)\b|inputs\.nonce/.test(g));
 }
 
 const json = <T>(r: { ok: boolean; out: string }): T | null => { if (!r.ok) return null; try { return JSON.parse(r.out) as T; } catch { return null; } };
@@ -244,8 +245,8 @@ function pushPrefixes(env: TrainEnv, t: TrainRecord): string | null {
   return null;
 }
 
-/** Dispatch every workflow unit that has no live attempt on a prefix; a workflow with a global concurrency group gets one prefix at a time. */
-async function dispatchWorkflows(env: TrainEnv, slug: string, t: TrainRecord, units: GateUnit[], lines: string[]): Promise<void> {
+/** Dispatch every workflow unit that has no live attempt on a prefix; a workflow with a global concurrency group gets one prefix at a time. A unit is sent at most twice per prefix: two sends no run adopted refuse. */
+async function dispatchWorkflows(env: TrainEnv, slug: string, t: TrainRecord, units: GateUnit[], lines: string[]): Promise<QueueFailure | null> {
   const exec = execFor(env.ctx);
   for (const u of units.filter(x => x.workflow !== 'command')) {
     const yaml = git(env.ctx.repoRoot, ['show', `${env.remote}/${env.ctx.base}:.github/workflows/${u.workflow}`]);
@@ -255,6 +256,7 @@ async function dispatchWorkflows(env: TrainEnv, slug: string, t: TrainRecord, un
       const all = attemptsOf(t);
       const prior = unitAttempt(all, u, p.index);
       if (prior && prior.state !== 'absent') continue;
+      if (all.filter(a => a.gate === u.gate && a.workflow === u.workflow && a.prefixes[0] === p.index && a.state === 'absent').length >= 2) return { code: 'QUEUE_TRAIN_DISPATCH_UNKNOWN', message: `${u.workflow} on T${p.index} was sent twice under nonce ${prior!.nonce} and no run carrying it appeared; it is not sent a third time (check that its run-name names the nonce: train doctor)` };
       if (serial && t.gated!.some(q => { const a = unitAttempt(all, u, q.index); return !!a && a.state !== 'absent' && a.state !== 'completed'; })) break;
       const reuse = p.index === 0 ? reusedT0(env, t, u, env.train.gates![u.gate]!) : null;
       const nonce = prior?.nonce ?? makeNonce('gate', p.commit, t.base.sha, randomBytes(3).toString('hex'), { purpose: 'train' });
@@ -272,11 +274,43 @@ async function dispatchWorkflows(env: TrainEnv, slug: string, t: TrainRecord, un
       if (serial) break;
     }
   }
+  return null;
 }
 
-/** Dollars reserved by every command run of this train (an interrupted run stays counted). */
-function commandSpend(t: TrainRecord): number {
-  return Math.round(attemptsOf(t).filter(a => a.workflow === 'command' && !a.reused).reduce((s, a) => s + (a.spend_usd ?? 0), 0) * 100) / 100;
+/** Dollars reserved by one command gate's runs in this train (an interrupted run stays counted); each gate has its own cap. */
+function commandSpend(t: TrainRecord, gate: number): number {
+  return Math.round(attemptsOf(t).filter(a => a.workflow === 'command' && a.gate === gate && !a.reused).reduce((s, a) => s + (a.spend_usd ?? 0), 0) * 100) / 100;
+}
+
+const pidAlive = (pid: unknown): boolean => {
+  if (typeof pid !== 'number') return false;
+  try { process.kill(pid, 0); return true; } catch (e) { return (e as NodeJS.ErrnoException).code === 'EPERM'; }
+};
+
+/** Whether a command attempt's process may still be running: its owner is alive on this host, or on another host within the run's time. */
+function commandLive(a: AttemptRecord, g: TrainGate | undefined): boolean {
+  if (a.workflow !== 'command' || a.state !== 'dispatching' || !a.owner) return false;
+  if (a.owner.host !== os.hostname()) return g?.kind === 'command' && Date.now() - Date.parse(a.updated_at) < (g.timeout_sec + COMMAND_OVERHEAD_MIN * 60) * 2000;
+  return pidAlive(a.owner.pid);
+}
+
+/** One build or verdict at a time writes a generic train's attempts: a lock file naming its owner, taken over once that owner is gone (another host's after six hours). */
+function trainLease(t: TrainRecord): { release: () => void } | QueueFailure {
+  const file = path.join(trainDir(t.nonce), 'gate.lock');
+  fs.mkdirSync(path.dirname(file), { recursive: true });
+  for (let i = 0; i < 2; i++) {
+    try {
+      fs.writeFileSync(file, JSON.stringify({ pid: process.pid, host: os.hostname() }), { flag: 'wx' });
+      return { release: () => fs.rmSync(file, { force: true }) };
+    } catch (e) { if ((e as NodeJS.ErrnoException).code !== 'EEXIST') throw e; }
+    let o: { pid?: unknown; host?: unknown } | null = null;
+    try { o = JSON.parse(fs.readFileSync(file, 'utf8')); } catch { o = null; }
+    const age = Date.now() - (fs.statSync(file, { throwIfNoEntry: false })?.mtimeMs ?? 0);
+    const held = !o ? age < 10_000 : o.host === os.hostname() ? pidAlive(o.pid) : age < 6 * 3600_000;
+    if (held) return { code: 'QUEUE_TRAIN_PENDING', message: `another train build or verdict (process ${String(o?.pid ?? '?')} on ${String(o?.host ?? '?')}) holds ${file}; nothing read or sent` };
+    fs.rmSync(file, { force: true });
+  }
+  return { code: 'QUEUE_TRAIN_PENDING', message: `${file} changed hands twice while this process waited; nothing read or sent` };
 }
 
 /** Run one command gate on a prefix through ubi-runner; resolves with the outcome. */
@@ -294,15 +328,15 @@ async function runCommand(env: TrainEnv, t: TrainRecord, a: AttemptRecord, g: Co
     const childEnv = { ...process.env, UBI_SIZE: COMMAND_SIZE, UBI_RUN_TIMEOUT: String(g.timeout_sec), UBI_TTL_HOURS: String(Math.ceil(g.timeout_sec / 3600) + 1) };
     const child = spawn(inv.cmd, inv.argv, { cwd: wt, env: childEnv, timeout: (g.timeout_sec + COMMAND_OVERHEAD_MIN * 60) * 1000 });
     let out = '';
-    const keep = (b: Buffer) => { out = (out + b.toString('utf8')).slice(-1_000_000); };
+    let tree = null as string | null;
+    const keep = (b: Buffer) => { const all = out + b.toString('utf8'); tree ??= /^GSTACK_GATE_TREE=([0-9a-f]{40,64})$/m.exec(all)?.[1] ?? null; out = all.slice(-1_000_000); };
     child.stdout.on('data', keep);
     child.stderr.on('data', keep);
     const code = await new Promise<number | null>(resolve => { child.on('close', c => resolve(c)); child.on('error', () => resolve(null)); });
     const log = path.join(trainDir(t.nonce), 'attempts', `${a.id}.log`);
     fs.mkdirSync(path.dirname(log), { recursive: true });
     fs.writeFileSync(log, out, { mode: 0o600 });
-    const tree = /^GSTACK_GATE_TREE=([0-9a-f]{40,64})$/m.exec(out)?.[1] ?? null;
-    const left = /UBICLOUD: left=(\S+)/.exec(out)?.[1];
+    const left = [...out.matchAll(/UBICLOUD: left=(\S+)/g)].at(-1)?.[1];
     if (!tree) return { outcome: 'INCOMPLETE', detail: `the VM reported no tree (exit ${code ?? 'none'}; log ${log})`, tree };
     if (tree !== p.tree) return { outcome: 'INCOMPLETE', detail: `the VM tested tree ${tree.slice(0, 12)}, not the prefix tree ${p.tree.slice(0, 12)}`, tree };
     if (left !== undefined && left !== '0') return { outcome: 'INCOMPLETE', detail: `ubi-runner reports left=${left}; the run is not counted`, tree };
@@ -319,7 +353,7 @@ async function runCommand(env: TrainEnv, t: TrainRecord, a: AttemptRecord, g: Co
 async function settleCommand(env: TrainEnv, t: TrainRecord, a: AttemptRecord, g: CommandGate, p: GatedPrefix, lines: string[]): Promise<void> {
   const price = commandReservation(g);
   let r = await runCommand(env, t, a, g, p);
-  if (r.outcome === 'RED' && commandSpend(t) + price <= g.spend_cap_usd) {
+  if (r.outcome === 'RED' && commandSpend(t, a.gate!) + price <= g.spend_cap_usd) {
     a.spend_usd = (a.spend_usd ?? 0) + price;
     save(env, t, a);
     const again = await runCommand(env, t, a, g, p);
@@ -345,12 +379,13 @@ async function runCommands(env: TrainEnv, t: TrainRecord, units: GateUnit[], lin
       if (unitAttempt(attemptsOf(t), u, p.index)?.state === 'completed') continue;
       const reuse = p.index === 0 ? reusedT0(env, t, u, g) : null;
       if (reuse) { t.attempts.push(reuse.id); save(env, t, reuse); lines.push(`REUSED: T0 command ${reuse.detail}`); continue; }
-      if (commandSpend(t) + price > g.spend_cap_usd) {
+      if (commandSpend(t, u.gate) + price > g.spend_cap_usd) {
         await Promise.all(jobs);
-        return { code: 'QUEUE_TRAIN_SPEND_CAP', message: `gate ${u.gate}'s run on T${p.index} reserves $${price.toFixed(2)}; with $${commandSpend(t).toFixed(2)} reserved, this train would pass the cap $${g.spend_cap_usd.toFixed(2)}` };
+        return { code: 'QUEUE_TRAIN_SPEND_CAP', message: `gate ${u.gate}'s run on T${p.index} reserves $${price.toFixed(2)}; with $${commandSpend(t, u.gate).toFixed(2)} reserved, this train would pass the gate's cap $${g.spend_cap_usd.toFixed(2)}` };
       }
       const a = newAttempt(t, u, p, makeNonce('gate', p.commit, t.base.sha, randomBytes(3).toString('hex'), { purpose: 'train' }), { command: g.command });
       a.spend_usd = price;
+      a.owner = { pid: process.pid, host: os.hostname() };
       t.attempts.push(a.id);
       save(env, t, a);
       jobs.push(settleCommand(env, t, a, g, p, lines));
@@ -365,23 +400,35 @@ export async function advanceGeneric(env: TrainEnv, t: TrainRecord, lines: strin
   const run = trainFile(t.nonce);
   const slug = repoSlug(env.ctx);
   if (!slug) return outRefused({ code: 'QUEUE_PROOF_MISSING', message: 'gh repo view could not name the repository' }, lines, `${Q} train build --train ${t.nonce}`, run);
-  const pushed = pushPrefixes(env, t);
-  if (pushed) return outRefused({ code: 'QUEUE_PROOF_MISSING', message: pushed }, lines, `${Q} train build --train ${t.nonce}`, run);
-  const units = gateUnits(env.train.gates!);
-  reconcile(env, slug, t);
-  await dispatchWorkflows(env, slug, t, units, lines);
-  const capped = await runCommands(env, t, units, lines);
-  t.state = 'gate-pending';
-  save(env, t);
-  if (capped) return outRefused(capped, [...lines, ...describe(t)], `raise the gate's spend_cap_usd on the base branch or lower its timeout_sec, then ${Q} train build --train ${t.nonce}`, run);
-  return outOk([...lines, ...describe(t), `GATES: ${units.length} unit(s) on ${t.gated!.length} prefix(es); record: ${run}`], `${Q} train verdict --train ${t.nonce}`, run);
+  const lease = trainLease(t);
+  if ('code' in lease) return outPending(lease, lines, `${Q} train build --train ${t.nonce} (once that process finishes)`, run);
+  try {
+    Object.assign(t, loadTrain(t.nonce) ?? {});
+    const pushed = pushPrefixes(env, t);
+    if (pushed) return outRefused({ code: 'QUEUE_PROOF_MISSING', message: pushed }, lines, `${Q} train build --train ${t.nonce}`, run);
+    const units = gateUnits(env.train.gates!);
+    if (t.verdict) return outOk([...lines, ...describe(t), `VERDICT: ${t.verdict.kind.toUpperCase()} (settled; nothing dispatched)`], genericNext(t), run);
+    const live = attemptsOf(t).find(a => commandLive(a, env.train.gates![a.gate ?? -1]));
+    if (live) return outPending({ code: 'QUEUE_TRAIN_PENDING', message: `command run ${live.id} is still running under process ${live.owner!.pid} on ${live.owner!.host}` }, lines, `${Q} train verdict --train ${t.nonce} (once that run finishes)`, run);
+    reconcile(env, slug, t);
+    const unsent = await dispatchWorkflows(env, slug, t, units, lines);
+    const capped = unsent ? null : await runCommands(env, t, units, lines);
+    t.state = 'gate-pending';
+    save(env, t);
+    if (unsent) return outRefused(unsent, [...lines, ...describe(t)], `run ${Q} train doctor and fix the workflow on the base branch, then build a new train on the new main: ${Q} train build ${prsArgs(t.slots, t.explicit_order)}`, run);
+    if (capped) return outRefused(capped, [...lines, ...describe(t)], `raise the gate's spend_cap_usd (or lower its timeout_sec) on the base branch, then build a new train on the new main: ${Q} train build ${prsArgs(t.slots, t.explicit_order)}`, run);
+    return outOk([...lines, ...describe(t), `GATES: ${units.length} unit(s) on ${t.gated!.length} prefix(es); record: ${run}`], `${Q} train verdict --train ${t.nonce}`, run);
+  } finally { lease.release(); }
 }
 
 /** Settle unconfirmed workflow dispatches by nonce (found, or absent so the next build re-sends the same nonce); an interrupted command run is absent. */
 function reconcile(env: TrainEnv, slug: string, t: TrainRecord): void {
   for (const a of attemptsOf(t)) {
     if (a.state !== 'dispatching' && a.state !== 'unknown') continue;
-    if (a.workflow === 'command') { a.state = 'absent'; effect(t, `command run ${a.id} was interrupted; it runs again`); save(env, t, a); continue; }
+    if (a.workflow === 'command') {
+      if (commandLive(a, env.train.gates?.[a.gate ?? -1])) continue;
+      a.state = 'absent'; effect(t, `command run ${a.id} was interrupted; it runs again`); save(env, t, a); continue;
+    }
     const runs = listDispatchRuns(a.workflow, execFor(env.ctx), slug, 100);
     if (!runs) continue;
     const hit = runForNonce(runs, a.nonce);
@@ -459,16 +506,25 @@ export function genericVerdict(env: TrainEnv, t: TrainRecord): TrainOut {
   const slug = repoSlug(env.ctx);
   if (!slug) return outPending({ code: 'QUEUE_PROOF_MISSING', message: 'gh repo view could not name the repository' }, [], `${Q} train verdict --train ${t.nonce}`, run);
   if (!t.verdict) {
-    reconcile(env, slug, t);
-    for (const a of attemptsOf(t)) {
-      if (a.workflow === 'command' || a.state !== 'dispatched' || !a.run) continue;
-      readWorkflowAttempt(env, slug, t, a);
-      save(env, t, a);
-    }
-    aggregate(env, t);
-    const c = classifyPrefixes(t.gated!, t.slots);
-    if (c.kind !== 'pending') { t.verdict = { ...c, at: nowIso() }; t.state = VERDICT_STATE[c.kind]; effect(t, `verdict: ${c.kind}${c.culprit ? ` (culprit #${c.culprit})` : ''}`); }
-    save(env, t);
+    const lease = trainLease(t);
+    if ('code' in lease) return outPending(lease, [], `${Q} train verdict --train ${t.nonce} (once that process finishes)`, run);
+    try {
+      Object.assign(t, loadTrain(t.nonce) ?? {});
+      const live = t.verdict ? null : attemptsOf(t).find(a => commandLive(a, env.train.gates![a.gate ?? -1]));
+      if (live) return outPending({ code: 'QUEUE_TRAIN_PENDING', message: `command run ${live.id} is still running under process ${live.owner!.pid} on ${live.owner!.host}; nothing read or saved` }, [], `${Q} train verdict --train ${t.nonce} (once that run finishes)`, run);
+      if (!t.verdict) {
+        reconcile(env, slug, t);
+        for (const a of attemptsOf(t)) {
+          if (a.workflow === 'command' || a.state !== 'dispatched' || !a.run) continue;
+          readWorkflowAttempt(env, slug, t, a);
+          save(env, t, a);
+        }
+        aggregate(env, t);
+        const c = classifyPrefixes(t.gated!, t.slots);
+        if (c.kind !== 'pending') { t.verdict = { ...c, at: nowIso() }; t.state = VERDICT_STATE[c.kind]; effect(t, `verdict: ${c.kind}${c.culprit ? ` (culprit #${c.culprit})` : ''}`); }
+        save(env, t);
+      }
+    } finally { lease.release(); }
   }
   const lines = [
     ...t.gated!.map(p => `PREFIX: T${p.index} ${p.outcome ?? 'PENDING'} ${p.commit.slice(0, 12)} ${p.ref}${p.red.length ? ` red=${p.red.join(',')}` : ''}${p.detail ? ` (${p.detail})` : ''}`),

lib/ship-queue-train-init.ts

@@ -13,11 +13,13 @@ import { loadPolicyFromBase, isUnavailable, TRAIN_GATES_KEY_VERSION, globToRegEx
 import type { QueueContext } from './ship-queue';
 import { git } from './ship-queue-gate';
 import { versionAtRef } from './restamp';
-import { concurrencyKeyed } from './ship-queue-train-gate';
+import { commandReservation, concurrencyKeyed } from './ship-queue-train-gate';
 import { repoSlug } from './ship-queue-train-github';
+import { UNPROTECTED } from './ship-queue-train-land';
 import { fetchBase, outFailed, outOk, outRefused, type TrainOut } from './ship-queue-train';
 
 const Q = 'gstack-ship-queue';
+const MAX_SIZE = 5;
 const unquote = (s: string) => s.trim().replace(/\s+#.*$/, '').replace(/^(['"])(.*)\1$/, '$2');
 
 export interface WorkflowFacts {
@@ -118,7 +120,7 @@ function readRepo(ctx: QueueContext): Repo | TrainOut {
       strict = p.strict === true;
       required = [...new Set([...(p.contexts ?? []), ...(p.checks ?? []).map(c => c.context ?? '')].filter(Boolean))];
     } catch { strict = null; }
-  } else if (/not protected|required status checks not enabled|HTTP 404/i.test(`${prot.err ?? ''} ${prot.out}`)) strict = false;
+  } else if (UNPROTECTED.test(`${prot.err ?? ''} ${prot.out}`)) strict = false;
   return { ctx, policy: loaded.policy, base: ctx.base, slug: repoSlug(ctx), workflows, strict, required };
 }
 
@@ -175,11 +177,14 @@ export function trainInit(ctx: QueueContext, o: { command?: string; timeoutSec?:
   for (const m of mapped.filter(x => x.w)) byWorkflow.set(m.w!.file, [...(byWorkflow.get(m.w!.file) ?? []), `${m.w!.file}/${m.check}`]);
   const gates: unknown[] = [];
   if (byWorkflow.size) gates.push({ kind: 'workflows', workflows: [...byWorkflow.keys()], required_jobs: [...byWorkflow.values()].flat() });
-  if (o.command) gates.push({ kind: 'command', command: o.command, timeout_sec: o.timeoutSec ?? 1800, spend_cap_usd: o.spendCap ?? 2 });
+  const command = { kind: 'command' as const, command: o.command ?? '', timeout_sec: o.timeoutSec ?? 1800, spend_cap_usd: 0 };
+  command.spend_cap_usd = o.spendCap ?? Math.min(100, Math.ceil((MAX_SIZE + 2) * commandReservation(command)));
+  if (o.command) gates.push(command);
   if (!gates.length) lines.push(`GATES: no required check maps to a workflow job; name the gate yourself (--command "<test command>", or a workflows gate listing <workflow>/<job>)`);
   const method = r.policy.history === 'merge-only' ? 'merge' : 'squash';
   const land = { merge_method: method, ...(!r.policy.changelog_fragments && r.required.length ? { required_checks: r.required } : {}) };
-  const block = { train: { gates, land, max_size: 5, fill_hours: 3 }, min_gstack_version: TRAIN_GATES_KEY_VERSION };
+  const block = { train: { gates, land, max_size: MAX_SIZE, fill_hours: 3 }, min_gstack_version: TRAIN_GATES_KEY_VERSION };
+  if (o.command && o.spendCap !== undefined && o.spendCap < (MAX_SIZE + 2) * commandReservation(command)) lines.push(`WARN: --spend-cap $${o.spendCap} is below $${((MAX_SIZE + 2) * commandReservation(command)).toFixed(2)}, what ${MAX_SIZE + 1} prefixes and one rerun reserve; a full train would stop at QUEUE_TRAIN_SPEND_CAP`);
   lines.push('POLICY: add to .gstack/ship-policy.json on the base branch (merge with the keys it has):', ...JSON.stringify(block, null, 2).split('\n'));
   for (const file of byWorkflow.keys()) {
     lines.push(`WORKFLOW: ${file} needs, beside its own triggers:`, ...NONCE_SNIPPET.split('\n').map(l => '  ' + l));
@@ -203,7 +208,12 @@ export function trainDoctor(ctx: QueueContext): TrainOut {
   if (refused) return refused;
   const findings: string[] = [];
   for (const g of train.gates) {
-    if (g.kind !== 'workflows') { lines.push(`GATE: command ${JSON.stringify(g.command)} timeout ${g.timeout_sec}s cap $${g.spend_cap_usd}`); continue; }
+    if (g.kind !== 'workflows') {
+      lines.push(`GATE: command ${JSON.stringify(g.command)} timeout ${g.timeout_sec}s cap $${g.spend_cap_usd}`);
+      const full = (train.max_size + 1) * commandReservation(g);
+      if (g.spend_cap_usd < full) findings.push(`command gate cap $${g.spend_cap_usd} is below $${full.toFixed(2)}, what ${train.max_size + 1} prefixes reserve at timeout ${g.timeout_sec}s; a full train stops at QUEUE_TRAIN_SPEND_CAP`);
+      continue;
+    }
     for (const file of g.workflows) {
       const w = r.workflows.find(x => x.file === file);
       if (!w) { findings.push(`${file}: not in .github/workflows on ${r.base}`); continue; }

lib/ship-queue-train-land.ts

@@ -49,10 +49,13 @@ function missingChecks(ci: ReturnType<typeof ciAtHead>, required: string[]): str
   });
 }
 
+/** GitHub's answers that mean no required-checks protection; a bare 404 or 403 (a token that cannot read protection) is unreadable, never unprotected. */
+export const UNPROTECTED = /Branch not protected|Required status checks not enabled/i;
+
 /** Branch protection's "require branches to be up to date": true, false (unprotected or not strict), or null when unreadable. */
 function strictProtection(ctx: QueueContext): boolean | null {
   const r = ctx.gh(['api', `repos/{owner}/{repo}/branches/${ctx.base}/protection/required_status_checks`]);
-  if (!r.ok) return /not protected|required status checks not enabled|HTTP 404/i.test(`${r.err ?? ''} ${r.out}`) ? false : null;
+  if (!r.ok) return UNPROTECTED.test(`${r.err ?? ''} ${r.out}`) ? false : null;
   try { return (JSON.parse(r.out) as { strict?: unknown }).strict === true; } catch { return null; }
 }
 
@@ -105,6 +108,7 @@ function recordLanding(s: Slot, mode: LandingRecord['mode'], update: string | nu
 function chooseMode(ctx: QueueContext, s: Slot, mainSha: string, conditional: boolean): { mode: 'direct' | 'update' | 'wait'; why: string } {
   if (s.t.stamped) return { mode: 'update', why: `the slot publishes the stamped ${label(s)}` };
   if (conditional) return { mode: 'update', why: 'a conditional (base-red) PR lands only through a commit whose own checks run on the gated tree' };
+  if ((s.env.train.land?.merge_method ?? 'squash') === 'rebase') return { mode: 'update', why: 'a rebase replays the pin\'s commits, and the replayed tree is not the one the train gated' };
   const strict = strictProtection(ctx);
   if (strict !== false) return { mode: 'update', why: strict ? 'branch protection requires branches to be up to date' : 'branch protection could not be read, so the up-to-date path is taken' };
   const m = git(ctx.repoRoot, ['merge-tree', '--write-tree', mainSha, s.pin]);
@@ -123,6 +127,9 @@ async function mergeAt(ctx: QueueContext, s: Slot, head: string, base: string):
   let main = mainNow(ctx, s.env);
   if ('error' in main) return outPending({ code: 'QUEUE_PROOF_MISSING', message: main.error }, s.lines, s.rerun, s.run);
   if (main.sha !== base) return stopped(s.env, s.t, 'QUEUE_TRAIN_TREE_MISMATCH', `main moved to ${main.sha.slice(0, 12)} before the merge (the slot was prepared on ${base.slice(0, 12)})`, main.sha, s.lines, true);
+  const target = prInfo(ctx, pr);
+  if ('error' in target) return outPending({ code: 'QUEUE_PROOF_MISSING', message: target.error }, s.lines, s.rerun, s.run);
+  if (target.baseRefName && target.baseRefName !== ctx.base) return retargeted(s, target.baseRefName, ctx.base);
   const m = ctx.gh(['pr', 'merge', pr, `--${method}`, '--match-head-commit', head]);
   if (!m.ok) return outFailed({ code: 'QUEUE_NOT_READY', message: `gh pr merge refused: ${(m.err ?? m.out).trim().slice(0, 200)}` }, s.lines, s.rerun, s.run);
   main = mainNow(ctx, s.env);
@@ -132,12 +139,15 @@ async function mergeAt(ctx: QueueContext, s: Slot, head: string, base: string):
   return confirmed(s, main.sha, `MERGED: #${pr} as ${main.sha} (${method}); main's tree is ${label(s)} exactly`);
 }
 
-/** confirmMerged, with the post-merge checks a workflows gate declares named in the NEXT. */
+/** The check names every workflows gate asks to see green on main after each land. */
+export const postMergeChecks = (env: TrainEnv): string[] => (env.train.gates ?? []).flatMap(g => (g.kind === 'workflows' ? g.post_merge_checks ?? [] : []));
+
+/** confirmMerged, with the post-merge checks a workflows gate declares in the NEXT: the next slot lands only once they are green. */
 function confirmed(s: Slot, mainSha: string, line: string): TrainOut {
   const out = confirmMerged(s.env, s.t, s.t.slots[s.i]!.pr, mainSha, s.target.index, [...s.lines, line]);
-  const post = (s.env.train.gates ?? []).flatMap(g => (g.kind === 'workflows' ? g.post_merge_checks ?? [] : []));
+  const post = postMergeChecks(s.env);
   if (!post.length || out.exit !== 0) return out;
-  return { ...out, lines: [...out.lines, `POST_MERGE_CHECKS: ${post.join(', ')} at ${mainSha.slice(0, 12)}`], next: `${Q} verify --post-merge ${mainSha} && ${out.next}` };
+  return { ...out, lines: [...out.lines, `POST_MERGE_CHECKS: ${post.join(', ')} at ${mainSha.slice(0, 12)}`], next: `${Q} verify --post-merge ${mainSha} --checks ${JSON.stringify(post.join(','))} && ${out.next}` };
 }
 
 /** `direct`: merge the pin itself once its required checks are green there. */

lib/ship-queue-train-ready.ts

@@ -192,7 +192,7 @@ function conditionalReady(env: TrainEnv, o: ReadyOptions, lines: string[]): Trai
   const again = `gstack-base-red classify on the current base, /ship, then ${Q} ready --pr ${o.pr} --no-screen --base-red --fixer ${o.baseRed!.fixer}`;
   if (!red) return outRefused({ code: 'SHIP_BASE_RED_FIXER', message: `#${o.pr}'s receipt carries no base_red condition; ready --base-red is only for a paused /ship` }, lines, `${Q} ready --pr ${o.pr} --no-screen`, 'none');
   if (red.base !== baseSha) return outRefused({ code: 'SHIP_BASE_RED_FIXER', message: `the condition was classified at ${red.base.slice(0, 12)} and ${ctx.base} is now ${baseSha.slice(0, 12)}; a base move reclassifies, never trusts` }, lines, again, 'none');
-  if (Date.parse(red.expires_at) < Date.parse(nowIso())) return outRefused({ code: 'BASE_RED_PAUSE_EXPIRED', message: `the condition expired at ${red.expires_at}` }, lines, again, 'none');
+  if (!(Date.parse(red.expires_at) >= Date.parse(nowIso()))) return outRefused({ code: 'BASE_RED_PAUSE_EXPIRED', message: `the condition expired at ${red.expires_at}` }, lines, again, 'none');
   const shipped = bindTrainReceipt(ctx.repoRoot, `refs/remotes/${env.remote}/${ctx.base}`, info.body, o.pr, [head], true);
   lines.push(shipped.line, ...(shipped.depthLine ? [shipped.depthLine] : []));
   if (!shipped.ok) return outRefused({ code: shipped.code, message: shipped.message }, lines, `run /ship on ${head.slice(0, 12)}, then ready again`, 'none');

lib/ship-queue-train-record.ts

@@ -79,6 +79,8 @@ export interface AttemptRecord {
   /** Generic trains: the policy gate index, the run attempt read, both attempts' jobs, the rerun and its FLAKY cases, a reused T0 run. */
   gate?: number; run_attempt?: number; rerun?: 'requested' | 'refused'; jobs?: Array<{ attempt: number; name: string; conclusion: string | null }>;
   flaky?: string[]; reused?: string; tree?: string | null; spend_usd?: number;
+  /** A command gate's running process: reconcile marks the attempt absent only once this owner is gone. */
+  owner?: { pid: number; host: string };
 }
 export type PrefixResult = 'red' | 'green';
 export interface BisectRecord {

lib/ship-queue-train-watch.ts

@@ -19,6 +19,7 @@ import { repoSlug, runView } from './ship-queue-train-github';
 import { attemptsOf, trainFile, type TrainRecord } from './ship-queue-train-record';
 import type { TrainEnv } from './ship-queue-train';
 import { trainVerdict } from './ship-queue-train-verdict';
+import { verifyPostMerge } from './ship-queue-post-merge';
 
 const Q = 'gstack-ship-queue';
 const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
@@ -74,7 +75,8 @@ async function watchGeneric(ctx: QueueContext, env: TrainEnv, start: TrainRecord
   verdict ??= trainVerdict(ctx, { train: t.nonce });
   lines.push(...verdict.lines);
   if (!o.land || !t.verdict.landable) return { ...verdict, lines };
-  const { trainLand } = await import('./ship-queue-train-land');
+  const { postMergeChecks, trainLand } = await import('./ship-queue-train-land');
+  const post = postMergeChecks(env);
   for (const s of t.slots.slice(0, t.verdict.landable)) {
     if (t.merged.some(m => m.pr === s.pr)) continue;
     const remaining = Math.max(1, Math.floor((deadline - Date.now()) / 60_000));
@@ -82,6 +84,11 @@ async function watchGeneric(ctx: QueueContext, env: TrainEnv, start: TrainRecord
     lines.push(`WATCH: train land --train ${t.nonce} --pr ${s.pr} (${landed.failure?.code ?? landed.status})`, ...landed.lines);
     if (landed.status !== 'complete') return { ...landed, lines };
     reload();
+    const sha = t.merged.find(m => m.pr === s.pr)?.sha;
+    if (!post.length || !sha) continue;
+    const checked = await verifyPostMerge(ctx, { sha, checks: post, timeoutSec: Math.max(1, Math.floor((deadline - Date.now()) / 1000)), pollMs: o.pollMs });
+    lines.push(`WATCH: verify --post-merge ${sha} (${checked.failure?.code ?? checked.status})`, ...checked.lines);
+    if (checked.status !== 'complete') return { ...checked, lines, next: `${checked.next === 'none' ? '' : `${checked.next} && `}${Q} train watch --train ${t.nonce} --land` };
   }
   return outOk(lines, nextFor(t), run);
 }

lib/ship-queue-train.ts

@@ -427,7 +427,7 @@ async function newTrain(env: TrainEnv, o: BuildOptions): Promise<{ t: TrainRecor
   lines.push(`MERGE_CHECK: T1..T${order.train.length} merge clean onto ${env.ctx.base}@${baseSha.slice(0, 12)} (dry run in object storage; nothing pushed)`);
   const due = fillDue(order.train, env.train, nowIso(), o.urgent || o.repair);
   if (due && !o.dryRun) return outPending({ code: 'QUEUE_TRAIN_WAITING', message: `${order.train.length} of ${env.train.max_size} PRs; the oldest READY dispatches at ${due}` }, lines, `${Q} train build ${prsArgs(order.train, !!explicit)} (after ${due}, or with more PRs, or --urgent for one)`, 'none');
-  const live = listTrains(env.repo.keys).find(t => t.key === key && !TERMINAL_STATES.includes(t.state) && t.state !== 'red' && t.state !== 'unattributed');
+  const live = listTrains(env.repo.keys).find(t => t.key === key && !TERMINAL_STATES.includes(t.state) && t.state !== 'red' && t.state !== 'unattributed' && !(t.gated && t.state === 'incomplete'));
   if (live) return { t: Object.assign(live, { held: order.held.map(s => s.pr) }), lines };
   const policy = env.policy.policy;
   const stamps = !!policy.changelog_fragments || (!!env.train.gates && policy.queue_mode === 'stamp-at-merge');
@@ -683,6 +683,8 @@ export function pruneTrains(ctx: QueueContext, o: { dryRun: boolean; train?: str
       continue;
     }
     if (!TERMINAL_STATES.includes(t.state)) { lines.push(`KEEP: ${t.branch} (${t.state})`); continue; }
+    const waiting = t.gated && t.verdict ? t.slots.slice(0, t.verdict.landable).filter(s => !t.merged.some(m => m.pr === s.pr)) : [];
+    if (waiting.length) { lines.push(`KEEP: ${t.branch} (${t.state}; ${waiting.map(s => '#' + s.pr).join(', ')} landable, not merged)`); continue; }
     for (const p of (t.gated ?? []).filter(g => g.ref !== t.branch)) remove(p.ref, t.state);
     remove(t.branch, t.state, () => { effect(t, `branch deleted: ${t.branch}`); save(env, t); });
   }

test/ship-queue-train-generic.test.ts

@@ -7,7 +7,9 @@
  * UBI_REAP=0, so nothing here reaches a cloud API.
  */
 import { afterEach, describe, expect, test } from 'bun:test';
+import { spawnSync } from 'node:child_process';
 import * as fs from 'node:fs';
+import * as os from 'node:os';
 import * as path from 'node:path';
 import { EXIT } from '../lib/headless-artifacts';
 import { TRAIN_GATES_KEY_VERSION, validatePolicy } from '../lib/ship-policy';
@@ -15,7 +17,7 @@ import { baseRedAdmission, classifyPrefixes, classifyWorkflowRun, commandReserva
 import { pushMatchesPrefix, workflowFacts } from '../lib/ship-queue-train-init';
 import type { TrainSlot } from '../lib/ship-queue-train-record';
 import { BASE_POLICY, cleanup } from './helpers/restamp-fixture';
-import { line, nonceOf, openPr, queue, readyReceipt, repoKey, shippedBody, trainFixture, trainJson, type TrainFixture } from './helpers/ship-queue-train-fixture';
+import { line, nonceOf, openPr, queue, readyReceipt, repoKey, shippedBody, trainFixture, trainJson, TRAIN_POLICY, type TrainFixture } from './helpers/ship-queue-train-fixture';
 import { runBin } from './helpers/restamp-fixture';
 
 const WORKFLOWS_GATE = { kind: 'workflows', workflows: ['ci.yml'], required_jobs: ['ci.yml/test', 'ci.yml/lint'] };
@@ -44,7 +46,7 @@ function wrap(fx: TrainFixture): void {
     "if(a[0]==='api'&&jobs){const r=d.runs.find(x=>String(x.databaseId)===jobs[1]);const n=Number(jobs[2]);const names=d.jobNames||['test','lint'];",
     "  const list=names.map(name=>({name,conclusion:name==='test'&&red(r,n)?'failure':'success'}));out({total_count:list.length,jobs:n>1?list.filter(j=>j.name==='test'):list})}",
     "if(a[0]==='run'&&a[1]==='rerun'){const r=d.runs.find(x=>String(x.databaseId)===a[2]);r.run_attempt=(r.run_attempt||1)+1;d.log.push({cmd:'run rerun',args:a});save();process.exit(0)}",
-    "if(a[0]==='api'&&/branches\\/main\\/protection\\/required_status_checks/.test(a[1]||'')){d.log.push({cmd:'api protection',args:a});save();if(!d.protection){console.error('gh: Branch not protected (HTTP 404)');process.exit(1)}console.log(JSON.stringify(d.protection));process.exit(0)}",
+    "if(a[0]==='api'&&/branches\\/main\\/protection\\/required_status_checks/.test(a[1]||'')){d.log.push({cmd:'api protection',args:a});save();if(!d.protection||d.protection==='404'){console.error(d.protection==='404'?'gh: Not Found (HTTP 404)':'gh: Branch not protected (HTTP 404)');process.exit(1)}console.log(JSON.stringify(d.protection));process.exit(0)}",
     "if(a[0]==='pr'&&a[1]==='list'){out(Object.entries(d.prs).filter(([,p])=>p.state==='OPEN').map(([n,p])=>({number:Number(n),headRefOid:p.headRefOid})))}",
     "if(a[0]==='api'&&/^repos\\/[^/{]+\\/[^/{]+$/.test(a[1]||'')){out(d.repoSettings||{allow_squash_merge:true,allow_merge_commit:true,allow_rebase_merge:false})}",
     "if(a[0]==='pr'&&a[1]==='view'&&d.prs[a[2]]){const p=d.prs[a[2]];const h=rev('refs/heads/'+p.headRefName);if(h)p.headRefOid=h;out(p)}",
@@ -103,6 +105,10 @@ describe('generic train policy', () => {
     expect(own.train.gate_workflow).toBe('evals.yml');
     expect(own.train.gates).toBeUndefined();
   });
+  test('history merge-only constrains only a train that declares land: a gstack-evals train without it still validates', () => {
+    expect(validatePolicy({ ...TRAIN_POLICY, history: 'merge-only' })).toEqual([]);
+    expect(errs(policyWith({ ...GENERIC, land: { merge_method: 'squash' } }, { history: 'merge-only' })).join('\n')).toContain('history merge-only allows only the merge method');
+  });
   test('a train with gates validates, and needs min_gstack_version at the gates key version', () => {
     expect(errs(policyWith(GENERIC))).toEqual([]);
     expect(errs(policyWith(GENERIC, { min_gstack_version: '1.91.336.0' })).join('\n')).toContain('gates and land were introduced in gstack ' + TRAIN_GATES_KEY_VERSION);
@@ -183,6 +189,7 @@ describe('base-red admission', () => {
     expect(code(baseRedAdmission([slot('1'), slot('5', cond('7'))], true, false))).toMatch(/conditional on fixer #7, which is not in this train/);
     expect(code(baseRedAdmission([slot('7', { fixes_base_red: ['x'], ...cond('7') })], true, false))).toMatch(/a fixer's own head must be green/);
     expect(code(baseRedAdmission([slot('7', { fixes_base_red: ['x'] }), slot('5', cond('7', '2026-01-01T00:00:00Z'))], true, false))).toMatch(/expired/);
+    expect(code(baseRedAdmission([slot('7', { fixes_base_red: ['x'] }), slot('5', cond('7', 'not a date'))], true, false))).toMatch(/expired/);
     expect(code(baseRedAdmission([slot('1'), slot('2')], false, false))).toBe('admitted');
   });
 });
@@ -197,6 +204,8 @@ describe('train init: workflow facts', () => {
     expect(pushMatchesPrefix('on:\n  pull_request:\n  workflow_dispatch:\n')).toBe(false);
     expect(concurrencyKeyed('concurrency:\n  group: ci-${{ github.ref }}\n')).toBe(true);
     expect(concurrencyKeyed('concurrency: ci\n')).toBe(false);
+    expect(concurrencyKeyed('concurrency:\n  group: ci-${{ github.head_ref }}\n')).toBe(false);
+    expect(concurrencyKeyed('concurrency:\n  group: ci-${{ github.head_ref || github.run_id }}\n')).toBe(true);
   });
   test('jobs (with matrix names), nonce input and run-name', () => {
     const yaml = [
@@ -255,6 +264,13 @@ describe('generic trains end to end (fixture repos)', () => {
     expect(merged.status, merged.err + merged.out).toBe(EXIT.ok);
     expect(fx.repo.git('--git-dir', fx.repo.origin, 'rev-parse', 'main^{tree}')).toBe(rec.gated[1].tree);
 
+    const hidden = genericTrain();
+    hidden.fx.write(d => { (d as any).protection = '404'; });
+    const n4 = nonceOf(queue(hidden.fx, ['train', 'build', '--prs', hidden.pins]).out);
+    queue(hidden.fx, ['train', 'verdict', '--train', n4]);
+    const unreadable = queue(hidden.fx, ['train', 'land', '--train', n4, '--pr', '1', '--no-wait']);
+    expect(unreadable.out, unreadable.err).toMatch(/^MODE: update \(branch protection could not be read/m);
+
     const rebase = genericTrain({ ...GENERIC, land: { merge_method: 'rebase', required_checks: ['test'] } });
     rebase.fx.write(d => { (d as any).protection = { strict: true, contexts: ['test'] }; });
     const n2 = nonceOf(queue(rebase.fx, ['train', 'build', '--prs', rebase.pins]).out);
@@ -262,6 +278,13 @@ describe('generic trains end to end (fixture repos)', () => {
     const refused = queue(rebase.fx, ['train', 'land', '--train', n2, '--pr', '1', '--no-wait']);
     expect(refused.err).toContain('(QUEUE_TRAIN_LAND_METHOD)');
     expect(rebase.fx.read().log.some(l => l.cmd === 'pr merge')).toBe(false);
+    const loose = genericTrain({ ...GENERIC, land: { merge_method: 'rebase', required_checks: ['test'] } });
+    const n5 = nonceOf(queue(loose.fx, ['train', 'build', '--prs', loose.pins]).out);
+    queue(loose.fx, ['train', 'verdict', '--train', n5]);
+    const replay = queue(loose.fx, ['train', 'land', '--train', n5, '--pr', '1', '--no-wait']);
+    expect(replay.err, replay.out).toContain('(QUEUE_TRAIN_LAND_METHOD)');
+    expect(replay.err).toContain('a rebase replays the pin');
+    expect(loose.fx.read().log.some(l => l.cmd === 'pr merge')).toBe(false);
 
     const fork = genericTrain();
     const n3 = nonceOf(queue(fork.fx, ['train', 'build', '--prs', fork.pins]).out);
@@ -282,6 +305,8 @@ describe('generic trains end to end (fixture repos)', () => {
     expect(v.err).toContain('(QUEUE_TRAIN_RED)');
     expect(line(v.out, 'VERDICT:')).toBe('VERDICT: ATTRIBUTED train ' + nonce + ' landable=1/2 culprit=#2');
     expect(line(v.out, 'PREFIX: T2')).toContain('red=ci.yml/test');
+    expect(queue(fx, ['train', 'prune', '--train', nonce]).out).toContain('#1 landable, not merged');
+    expect(fx.repo.git('--git-dir', fx.repo.origin, 'rev-parse', '--verify', '--quiet', 'refs/heads/' + trainJson(fx, nonce).branch)).toMatch(/^[0-9a-f]{40}$/);
     expect(queue(fx, ['train', 'land', '--train', nonce, '--pr', '2', '--no-wait']).err).toContain('(QUEUE_GATE_MISSING)');
     expect(queue(fx, ['train', 'land', '--train', nonce, '--pr', '1', '--no-wait']).status).toBe(EXIT.ok);
     expect(queue(fx, ['train', 'bisect', '--train', nonce, '--spend-cap', '5']).err).toContain('(QUEUE_TRAIN_REBUILD_REFUSED)');
@@ -437,6 +462,17 @@ describe('train init and doctor', () => {
     expect(line(queue(fx, ['train', 'init']).out, 'LAND_MODE:')).toMatch(/^LAND_MODE: update for every slot \(branch protection requires/);
   }, 300_000);
 
+  test('init prices a command gate for a full train and one rerun; doctor flags a cap below a full train', () => {
+    const fx = initFixture(PLAIN, CI());
+    const r = queue(fx, ['train', 'init', '--command', 'bun test']);
+    expect(r.status, r.err + r.out).toBe(EXIT.ok);
+    expect(r.out).toContain('"spend_cap_usd": 3');
+    expect(r.out).not.toContain('WARN: --spend-cap');
+    expect(queue(fx, ['train', 'init', '--command', 'bun test', '--spend-cap', '1']).out).toContain('WARN: --spend-cap $1 is below $2.38');
+    const low = initFixture(policyWith({ ...GENERIC, gates: [{ kind: 'command', command: 'bun test', timeout_sec: 1800, spend_cap_usd: 1 }] }), CI());
+    expect(queue(low, ['train', 'doctor']).out).toContain('command gate cap $1 is below $2.04');
+  }, 300_000);
+
   test('init refuses version-bumping PRs without changelog_fragments or stamp-at-merge, and recommends fragments either way', () => {
     const fx = initFixture(PLAIN, CI());
     openPr(fx, '2', { file: 'VERSION', text: '1.2.4.0\n', changelog: false });
@@ -511,6 +547,85 @@ describe('a workflows gate and a command gate together (round-2 High 3)', () =>
     expect(line(v.out, 'VERDICT:')).toBe('VERDICT: INCOMPLETE train ' + nonce + ' landable=0/2');
     expect(line(v.out, 'PREFIX: T0')).toMatch(/^PREFIX: T0 INCOMPLETE .*command: the command timed out/);
     expect(queue(fx, ['train', 'land', '--train', nonce, '--pr', '1', '--no-wait']).err).toContain('(QUEUE_GATE_MISSING)');
+    const again = queue(fx, ['train', 'build', '--prs', trainJson(fx, nonce).slots.map((s: any) => s.pr + '@' + s.head).join(',')]);
+    expect(nonceOf(again.out), again.err + again.out).not.toBe('');
+    expect(nonceOf(again.out)).not.toBe(nonce);
+  }, 300_000);
+});
+
+describe('review pass 1 voices: dispatch bound, lease, output tail, post-merge checks', () => {
+  test('a workflow dispatch no run ever adopts is sent twice per prefix, then watch stops on QUEUE_TRAIN_DISPATCH_UNKNOWN', () => {
+    const { fx, pins } = genericTrain();
+    fx.write(d => { d.rules['ci.yml'] = { silent: true } as never; });
+    const nonce = nonceOf(queue(fx, ['train', 'build', '--prs', pins]).out);
+    const watched = queue(fx, ['train', 'watch', '--train', nonce, '--timeout', '120']);
+    expect(watched.err, watched.out).toContain('(QUEUE_TRAIN_DISPATCH_UNKNOWN)');
+    expect(watched.err).toContain('was sent twice under nonce');
+    const sent = fx.read().log.filter(l => l.cmd === 'workflow run').map(l => (l.inputs as any).nonce);
+    expect(sent.length).toBeLessThanOrEqual(6);
+    for (const n of new Set(sent)) expect(sent.filter(x => x === n).length).toBeLessThanOrEqual(2);
+  }, 300_000);
+
+  test('a held lease refuses build and verdict with nothing sent; a dead holder\'s lease is taken over', () => {
+    const { fx, pins } = genericTrain();
+    const nonce = nonceOf(queue(fx, ['train', 'build', '--prs', pins]).out);
+    const lock = path.join(fx.state, 'ship-queue', 'trains', nonce, 'gate.lock');
+    fs.writeFileSync(lock, JSON.stringify({ pid: process.pid, host: os.hostname() }));
+    const before = fx.read().log.length;
+    for (const args of [['train', 'verdict', '--train', nonce], ['train', 'build', '--train', nonce]]) {
+      const held = queue(fx, args);
+      expect(held.err, held.out).toContain('(QUEUE_TRAIN_PENDING)');
+      expect(held.err).toContain('holds ' + lock);
+    }
+    expect(fx.read().log.slice(before).some(l => l.cmd === 'workflow run' || l.cmd === 'run rerun')).toBe(false);
+    fs.writeFileSync(lock, JSON.stringify({ pid: spawnSync('true').pid, host: os.hostname() }));
+    const v = queue(fx, ['train', 'verdict', '--train', nonce]);
+    expect(v.status, v.err + v.out).toBe(EXIT.ok);
+    expect(fs.existsSync(lock)).toBe(false);
+  }, 300_000);
+
+  test('a command that prints more than the 1 MB log tail keeps its tree line and is GREEN', () => {
+    const gate = { kind: 'command', command: "yes x | head -c 1200000; echo; echo 'UBICLOUD: left=3'", timeout_sec: 60, spend_cap_usd: 1 };
+    const { fx, pins } = genericTrain({ ...GENERIC, gates: [gate] });
+    fakeUbi(fx);
+    const nonce = nonceOf(queue(fx, ['train', 'build', '--prs', pins]).out);
+    const v = queue(fx, ['train', 'verdict', '--train', nonce]);
+    expect(line(v.out, 'VERDICT:'), v.err + v.out).toBe('VERDICT: GREEN train ' + nonce + ' landable=2/2');
+  }, 300_000);
+
+  test('post_merge_checks: verify --post-merge --checks waits for each named check; watch --land stops after the slot whose named check never ran', () => {
+    const { fx, pins } = genericTrain({ ...GENERIC, gates: [{ ...WORKFLOWS_GATE, post_merge_checks: ['deploy'] }] });
+    const sha = fx.repo.git('--git-dir', fx.repo.origin, 'rev-parse', 'main');
+    const missing = queue(fx, ['verify', '--post-merge', sha, '--checks', 'deploy', '--timeout', '1']);
+    expect(missing.err, missing.out).toContain('the post-merge checks the train gate names never ran');
+    expect(queue(fx, ['verify', '--post-merge', sha, '--checks', 'test', '--timeout', '1']).status).toBe(EXIT.ok);
+    const nonce = nonceOf(queue(fx, ['train', 'build', '--prs', pins]).out);
+    const watched = queue(fx, ['train', 'watch', '--train', nonce, '--land', '--timeout', '20']);
+    expect(watched.out, watched.err).toMatch(/^WATCH: verify --post-merge [0-9a-f]{40} \(QUEUE_TRAIN_PENDING\)/m);
+    expect(fx.read().log.filter(l => l.cmd === 'pr merge')).toHaveLength(1);
+    expect(trainJson(fx, nonce).merged.map((m: any) => m.pr)).toEqual(['1']);
+  }, 300_000);
+});
+
+describe('a command run another process still owns (red-team #4)', () => {
+  test('verdict and build wait while the owner is alive and read nothing; a dead owner\'s run is reconciled as interrupted', () => {
+    const { fx, pins } = genericTrain({ ...GENERIC, gates: [{ kind: 'command', command: 'true', timeout_sec: 60, spend_cap_usd: 1 }] });
+    fakeUbi(fx);
+    const nonce = nonceOf(queue(fx, ['train', 'build', '--prs', pins]).out);
+    const file = path.join(fx.state, 'ship-queue', 'trains', nonce, 'attempts', trainJson(fx, nonce).attempts[0] + '.json');
+    const own = (pid: number) => fs.writeFileSync(file, JSON.stringify({ ...JSON.parse(fs.readFileSync(file, 'utf8')), state: 'dispatching', owner: { pid, host: os.hostname() } }));
+    own(process.pid);
+    for (const args of [['train', 'verdict', '--train', nonce], ['train', 'build', '--train', nonce]]) {
+      const held = queue(fx, args);
+      expect(held.err, held.out).toContain('(QUEUE_TRAIN_PENDING)');
+      expect(held.err).toContain('still running under process ' + process.pid);
+    }
+    expect(JSON.parse(fs.readFileSync(file, 'utf8')).state).toBe('dispatching');
+    own(spawnSync('true').pid!);
+    const freed = queue(fx, ['train', 'verdict', '--train', nonce]);
+    expect(freed.err + freed.out).not.toContain('still running under process');
+    expect(line(freed.out, 'PREFIX: T0')).toContain('not dispatched');
+    expect(JSON.parse(fs.readFileSync(file, 'utf8')).state).not.toBe('dispatching');
   }, 300_000);
 });