test(integration): boot freight API in-process across parallel shards

The suite drove a containerized freight API, so every code change needed an
image rebuild before a test could see it, and there was no way to attach a
debugger. Files also ran strictly in sequence against one shared database,
which is the root of the warm-stack gotchas the README documents: stowaway
paid bookings climbing back aboard, a short consist on the fifth file.

The freight app now boots inside each vitest worker from dist/, and each
worker owns a whole shard of the topology - its own database, payment API,
gateway mock and broker vhost - so nothing mutable is shared and files run
in parallel. Full suite drops from roughly 20 minutes to 196s at 4 shards.

- main.ts exports createFreightApp() so the harness applies the same prefix,
  pipes, filters and interceptors as production instead of replaying them by
  hand; self-start is guarded by require.main so the Dockerfile CMD still boots
- booking-window tick cadence is env-driven (BOOKING_WINDOW_TICK_CRON), */1 in
  the suite, */10 unchanged in production
- prepare-shards.mjs seeds a template database (boot seeders, then the SQL
  fixtures that depend on them) and clones it per shard; it.mjs re-clones on
  every run, so each run is hermetic
- gateway mock and payment API are generated per shard: the mock keeps modes
  and orders process-global and 20 of 25 specs reset it in beforeAll, and the
  inbound CBE bill query has to reach one specific shard's app
- poll() samples every 250ms instead of 2000ms, keeping the caller's deadline
- authz.it.ts seeds its own invoice; it previously read another spec's leftover
  and returned early, which silently passed on a pristine database

Known: an unlocked MAX(sequence_no)+1 in train-scheduling.service.ts races
under concurrent allocation and leaves a short consist, so 1-3 specs fail
intermittently. Pre-existing and reproduces at the production tick cadence.

integration/src/client.ts

@@ -1,7 +1,11 @@
 /**
- * Plumbing for the integration suite: HTTP against the containerized freight
- * and payment APIs, SQL against their shared throwaway Postgres, and the
- * gateway mock's control plane.
+ * Plumbing for the integration suite: HTTP against this worker's IN-PROCESS
+ * freight API and its containerized payment API, SQL against the worker's own
+ * throwaway database, and its own gateway mock's control plane.
+ *
+ * Everything here is shard-scoped — see app.ts for the topology. supertest takes
+ * an `http.Server` exactly where it takes a base URL, so pointing the suite at
+ * the in-process app is a one-token change at each call site.
  *
  * This is the Cypress-free port of e2e/freight/cypress/e2e/flows/import-utils.ts —
  * same request sequences, same SQL, `pg.Pool` instead of `cy.task`.
@@ -9,13 +13,11 @@
 import request from "supertest";
 import { Pool, type QueryResultRow } from "pg";
 
-export const API = process.env.IT_API_URL ?? "http://localhost:3111";
-export const PAYMENT_API = process.env.IT_PAYMENT_URL ?? "http://localhost:3113";
-export const GATEWAY = process.env.IT_GATEWAY_URL ?? "http://localhost:4600";
-export const SERVICE_TOKEN = process.env.IT_SERVICE_TOKEN ?? "e2e-service-token";
+import { DB_URL, GATEWAY, PAYMENT_API, SHARD, freightServer, tag } from "./app";
 
-const DB_URL =
-  process.env.IT_DB_URL ?? "postgres://edr_e2e:edr_e2e@localhost:5543/edr_freight_e2e";
+export { GATEWAY, PAYMENT_API, SHARD, freightServer };
+
+export const SERVICE_TOKEN = process.env.IT_SERVICE_TOKEN ?? "e2e-service-token";
 
 // ---------------------------------------------------------------------------
 // users (e2e/freight/cypress/fixtures/seed-users.sql + users.json)
@@ -45,11 +47,23 @@ export async function db<T extends QueryResultRow = Record<string, unknown>>(
   return res.rows;
 }
 
-export async function closeDb(): Promise<void> {
-  await pool.end();
-}
+/**
+ * No-op. The pool is per-WORKER, not per-file: with `isolate: false` the module
+ * registry survives across every spec a worker runs, so the first `afterAll` to
+ * call this would break every later file on that shard. The pool dies with the
+ * worker process.
+ */
+export async function closeDb(): Promise<void> {}
+
+/**
+ * Poll a query until `check` passes. Async settlement here is broker-driven and
+ * tick-driven, so the interval is pure overshoot: at the old 2000ms every wait
+ * in the suite finished up to two seconds after the state it wanted was already
+ * in the database, several hundred times over. The ceiling
+ * (`attempts × intervalMs`) is unchanged.
+ */
+const POLL_INTERVAL_MS = Number(process.env.IT_POLL_INTERVAL_MS ?? 250);
 
-/** Poll a query until `check` passes. Async settlement here is broker-driven. */
 export async function poll<T extends QueryResultRow = Record<string, unknown>>(
   label: string,
   sql: string,
@@ -57,14 +71,24 @@ export async function poll<T extends QueryResultRow = Record<string, unknown>>(
   check: (row: T | undefined) => boolean,
   { attempts = 40, intervalMs = 2000 } = {},
 ): Promise<T> {
+  // Callers express patience as attempts × their own interval; keep that
+  // deadline and just sample it finely.
+  const deadlineMs = attempts * intervalMs;
+  const tries = Math.ceil(deadlineMs / POLL_INTERVAL_MS);
+
   let last: T | undefined;
-  for (let i = 0; i < attempts; i++) {
+  for (let i = 0; i < tries; i++) {
     last = (await db<T>(sql, params))[0];
     if (check(last)) return last as T;
-    await sleep(intervalMs);
+    await sleep(POLL_INTERVAL_MS);
   }
   throw new Error(
-    `timed out waiting for ${label} after ${attempts} attempts — last row: ${JSON.stringify(last)}`,
+    tag(
+      `timed out waiting for ${label} after ${Math.round(deadlineMs / 1000)}s\n` +
+        `  sql: ${sql.replace(/\s+/g, " ").trim()}\n` +
+        `  params: ${JSON.stringify(params)}\n` +
+        `  last row: ${JSON.stringify(last)}`,
+    ),
   );
 }
 
@@ -86,7 +110,7 @@ export async function tokenFor(email: string): Promise<string> {
   if (cached) return cached;
 
   const portal = email.endsWith("@gmail.com");
-  const res = await request(API)
+  const res = await request(await freightServer())
     .post("/api/auth/login")
     .set("x-client-app", portal ? "portal" : "backoffice")
     .send({ email, password: portal ? CUSTOMER_PASSWORD : STAFF_PASSWORD });
@@ -101,8 +125,11 @@ export async function tokenFor(email: string): Promise<string> {
 }
 
 /** Login without caching, so audience-rejection can be asserted. */
-export function login(email: string, password: string, app: "portal" | "backoffice") {
-  return request(API).post("/api/auth/login").set("x-client-app", app).send({ email, password });
+export async function login(email: string, password: string, app: "portal" | "backoffice") {
+  return request(await freightServer())
+    .post("/api/auth/login")
+    .set("x-client-app", app)
+    .send({ email, password });
 }
 
 // ---------------------------------------------------------------------------
@@ -119,7 +146,9 @@ export async function api(
   body?: unknown,
 ): Promise<request.Response> {
   const token = await tokenFor(email);
-  const req = request(API)[method](path).set("Authorization", `Bearer ${token}`);
+  const req = request(await freightServer())
+    [method](path)
+    .set("Authorization", `Bearer ${token}`);
   return method === "get" || method === "delete" ? req.send() : req.send(body ?? {});
 }
 
@@ -148,7 +177,9 @@ export async function upload(
   fields: Record<string, string> = {},
 ): Promise<request.Response> {
   const token = await tokenFor(email);
-  const req = request(API).post(path).set("Authorization", `Bearer ${token}`);
+  const req = request(await freightServer())
+    .post(path)
+    .set("Authorization", `Bearer ${token}`);
   for (const [k, v] of Object.entries(fields)) req.field(k, v);
   return req.attach(field, filePath);
 }