fix issue and add consolidation

apps/edr-freight-api/src/modules/train-scheduling/booking-batch.service.ts

@@ -10,6 +10,7 @@ import {
   Optional,
 } from '@nestjs/common';
 import { InjectDataSource } from '@nestjs/typeorm';
+import { EventEmitter2 } from '@nestjs/event-emitter';
 import { SchedulerRegistry } from '@nestjs/schedule';
 import {
   Between,
@@ -92,6 +93,7 @@ import { BookingWindowGateway } from './booking-window.gateway';
 import {
   MAX_TEU_SLOTS_PER_WAGON,
   containerWagonsForLines,
+  roundTons,
 } from './utils/wagon-plan.util';
 import {
   Capacity,
@@ -398,6 +400,8 @@ export class BookingBatchService implements OnModuleInit {
 
     @Optional() private readonly milestoneService?: ClearanceMilestoneService,
     @Optional() private readonly splitService?: BookingSplitService,
+    // Optional so hand-constructed spec instances keep compiling.
+    @Optional() private readonly eventEmitter?: EventEmitter2,
     @Optional()
     @Inject(forwardRef(() => RemainderPlacementService))
     private readonly remainderPlacement?: RemainderPlacementService,
@@ -1484,6 +1488,43 @@ export class BookingBatchService implements OnModuleInit {
           "Train is full — no export capacity left for this day",
         );
       }
+      // Physical wagon gate — a pay window must never open for wagons that do
+      // not exist in a type this cargo can ride. PER_TON bulk is seated
+      // type-by-type at its per-wagon caps (the count allocation will really
+      // need); everything else checks the summed free stock of its types.
+      const stock = await this.stockLedgerFor(
+        schedule,
+        budget,
+        bookings.map((b) => b.id),
+      );
+      const allowedWagonTypes = await this.loadAllowedWagonTypeIds();
+      const primary = bookings[0];
+      const wagonTypeIds = this.allowedWagonTypeIdsFor(primary, allowedWagonTypes);
+      const perItemBulk =
+        Number(primary.bulkTotalWeightTons ?? 0) > 0 &&
+        Number(primary.cargoTotalWeightVgm ?? 0) > 0;
+      const useSmart =
+        bookings.length === 1 &&
+        primary.freightType === "BULK" &&
+        !perItemBulk &&
+        wagonTypeIds.length > 0;
+      const smart = useSmart
+        ? this.smartBulkNeed(
+            primary,
+            wagonDims,
+            stock,
+            leg,
+            this.scarcityRankForPool([primary], allowedWagonTypes),
+          )
+        : null;
+      const seated = useSmart
+        ? smart != null && budget.fits(smart.need, leg)
+        : this.hasWagonStock(stock, wagonTypeIds, need.wagons, leg);
+      if (!seated) {
+        throw new ConflictException(
+          "Train has no free wagons of a type this cargo can ride — payment was not opened",
+        );
+      }
 
       for (const b of bookings) await this.reserve(b, scheduleId);
     });
@@ -2275,6 +2316,7 @@ export class BookingBatchService implements OnModuleInit {
     await this.recomputeBulkPriorities(pool, wagonDims);
     this.resortPoolByPriority(pool, await this.windowCycleIndexer(schedule));
     const units = this.groupConsolidatedPool(pool);
+    const scarcityRank = this.scarcityRankForPool(pool, allowedWagonTypes);
     let armed = false;
     let preempted = false;
     let reservedThisPass = 0;
@@ -2301,17 +2343,34 @@ export class BookingBatchService implements OnModuleInit {
       const leg = budget.legForYards(booking.originYardId, booking.destinationYardId);
       const wagonTypeIds = this.allowedWagonTypeIdsFor(booking, allowedWagonTypes);
       // Abstract room AND real wagons of a type this booking can ride — see
-      // fillRouteDayInternal for why both gates are needed.
-      const stocked = this.hasWagonStock(stock, wagonTypeIds, need.wagons, leg);
+      // fillRouteDayInternal for why both gates are needed. PER_TON bulk
+      // singles get the smart gate (exact per-type seating at the cargo's
+      // caps); a booking is only reserved — and only ever invoiced — when
+      // that seating is proven against the train's actual free wagons.
+      const perItemBulk =
+        Number(booking.bulkTotalWeightTons ?? 0) > 0 &&
+        Number(booking.cargoTotalWeightVgm ?? 0) > 0;
+      const useSmart =
+        !isPair &&
+        booking.freightType === "BULK" &&
+        !perItemBulk &&
+        wagonTypeIds.length > 0;
+      const smart = useSmart
+        ? this.smartBulkNeed(booking, wagonDims, stock, leg, scarcityRank)
+        : null;
+      const admitted = useSmart
+        ? smart != null && budget.fits(smart.need, leg)
+        : budget.fits(need, leg) &&
+          this.hasWagonStock(stock, wagonTypeIds, need.wagons, leg);
 
       // Per-unit fit trace: which axis (wagons/weight/length/stock) admits or rejects.
       this.logger.debug(
-        `[fillSchedule ${scheduleId}] unit ${booking.reference}: need=${JSON.stringify(need)} ` +
-          `roomOnLeg=${JSON.stringify(budget.remainingFor(leg))} fits=${budget.fits(need, leg)} ` +
-          `stocked=${stocked}`,
+        `[fillSchedule ${scheduleId}] unit ${booking.reference}: need=${JSON.stringify(
+          smart?.need ?? need,
+        )} roomOnLeg=${JSON.stringify(budget.remainingFor(leg))} admitted=${admitted}`,
       );
 
-      if (!budget.fits(need, leg) || !stocked) {
+      if (!admitted) {
         if (isGov) {
           const freed = await this.preemptForGovernment(
             scheduleId,
@@ -2355,9 +2414,16 @@ export class BookingBatchService implements OnModuleInit {
           armed = true;
           commercialReserved += 1;
         }
-        budget.subtract(need, leg);
+        budget.subtract(smart?.need ?? need, leg);
         // Hold the physical wagons too — the next unit must not re-count them.
-        stock.consume(wagonTypeIds, need.wagons, leg);
+        // The smart gate holds the exact per-type counts it seated.
+        if (smart) {
+          for (const part of smart.perType) {
+            stock.consume([part.wagonTypeId], part.wagons, leg);
+          }
+        } else {
+          stock.consume(wagonTypeIds, need.wagons, leg);
+        }
         reservedThisPass += 1;
       } catch (err) {
         this.logger.error(
@@ -2532,6 +2598,10 @@ export class BookingBatchService implements OnModuleInit {
     // Consolidated partners collapse into one atomic unit (both-or-neither); a
     // consolidated booking whose partner isn't ready this cycle is skipped.
     const units = this.groupConsolidatedPool(pool);
+    // Least-shareable-type-first seating for bulk (see smartBulkNeed): ranked
+    // once against the whole pool, so what containers will need is known
+    // before any bulk booking picks its wagons.
+    const scarcityRank = this.scarcityRankForPool(pool, allowedWagonTypes);
 
     // Batch fill trace: each train's caps + the day pool size at entry.
     this.logger.debug(
@@ -2555,18 +2625,47 @@ export class BookingBatchService implements OnModuleInit {
       // Consolidated pairs share one wagon set; the primary's types stand for both.
       const wagonTypeIds = this.allowedWagonTypeIdsFor(booking, allowedWagonTypes);
 
+      // PER_TON bulk singles get the smart gate: seated type-by-type at the
+      // cargo's per-wagon caps, scarcest type first — the count the allocator
+      // will actually need, not a one-type estimate. Pairs, PER_ITEM and
+      // unconfigured cargo keep the generic gate (gov preemption and partial
+      // offers below also still size on the generic `need`).
+      const perItemBulk =
+        Number(booking.bulkTotalWeightTons ?? 0) > 0 &&
+        Number(booking.cargoTotalWeightVgm ?? 0) > 0;
+      const useSmart =
+        !isPair &&
+        booking.freightType === "BULK" &&
+        !perItemBulk &&
+        wagonTypeIds.length > 0;
+      let smart: {
+        need: Capacity;
+        perType: Array<{ wagonTypeId: string; wagons: number }>;
+      } | null = null;
+
       // First train (earliest departure) whose corridor carries this booking's
       // leg, still fits it as-is AND physically holds enough wagons of a type the
       // booking can ride. Both gates matter: abstract room without the right
       // wagon type is space the allocator can never turn into a loaded consist.
-      let target = trains.find((t) => {
+      let target: (typeof trains)[number] | undefined;
+      for (const t of trains) {
         const leg = legOn(t);
-        return (
-          leg != null &&
+        if (leg == null) continue;
+        if (useSmart) {
+          const probe = this.smartBulkNeed(booking, wagonDims, t.stock, leg, scarcityRank);
+          if (probe != null && t.budget.fits(probe.need, leg)) {
+            smart = probe;
+            target = t;
+            break;
+          }
+        } else if (
           t.budget.fits(need, leg) &&
           this.hasWagonStock(t.stock, wagonTypeIds, need.wagons, leg)
-        );
-      });
+        ) {
+          target = t;
+          break;
+        }
+      }
 
       // Per-unit trace: chosen train + each train's remaining room on this leg.
       this.logger.debug(
@@ -2645,10 +2744,18 @@ export class BookingBatchService implements OnModuleInit {
           target.armed = true;
           commercialReserved += 1;
         }
-        target.budget.subtract(need, legOn(target)!);
+        target.budget.subtract(smart?.need ?? need, legOn(target)!);
         // Hold the physical wagons too, so the next unit in this pass sees them
-        // gone — otherwise two bookings both "fit" the same 16 NW5.
-        target.stock.consume(wagonTypeIds, need.wagons, legOn(target)!);
+        // gone — otherwise two bookings both "fit" the same 16 NW5. The smart
+        // gate holds the EXACT per-type counts it seated (10 PW2 + 17 NW5),
+        // not a type-blind total drained deepest-first.
+        if (smart) {
+          for (const part of smart.perType) {
+            target.stock.consume([part.wagonTypeId], part.wagons, legOn(target)!);
+          }
+        } else {
+          target.stock.consume(wagonTypeIds, need.wagons, legOn(target)!);
+        }
         target.changed = true;
         reservedThisPass += 1;
       } catch (err) {
@@ -2724,6 +2831,21 @@ export class BookingBatchService implements OnModuleInit {
     wagonTypeIds: string[] = [],
   ): Promise<boolean> {
     if (!this.isSplitEligible(booking, isPair)) return false;
+    const wagonDims = await this.loadWagonDims();
+    // PER_TON bulk partials are sized on ONE concrete wagon type at the
+    // cargo's per-wagon cap — sizing on the first type's raw 70T rating
+    // offered tonnage the wagons could never carry (Perishable caps at
+    // 20/30T), taking payment for cargo that stalls at allocation.
+    // ponytail: single-type bulk partials; a multi-type partial (PW2+NW5
+    // mixed) is the upgrade path if offers come out too small.
+    const perItemBulk =
+      Number(booking.bulkTotalWeightTons ?? 0) > 0 &&
+      Number(booking.cargoTotalWeightVgm ?? 0) > 0;
+    const cappedBulk =
+      !isPair &&
+      booking.freightType === "BULK" &&
+      !perItemBulk &&
+      wagonTypeIds.length > 0;
     const target = candidates
       .map((c) => {
         const leg = c.budget.legOf(booking.originYardId, booking.destinationYardId);
@@ -2734,12 +2856,39 @@ export class BookingBatchService implements OnModuleInit {
         // them NW5" into an offer for 16 — the customer pays for 16 and the
         // other 4 leave as the usual remainder booking, instead of paying for
         // 20 and stalling at allocation on wagon 17.
+        if (cappedBulk) {
+          const best = this.allowedDimsWithTypes(booking, wagonDims)
+            .filter((o): o is { wagonTypeId: string; dims: PerWagonDims } =>
+              o.wagonTypeId != null,
+            )
+            .map((o) => ({
+              ...o,
+              free: c.stock?.availableFor([o.wagonTypeId], leg) ?? 0,
+              takePerWagon: bulkTonsPerWagon(
+                booking.cargoType,
+                o.wagonTypeId,
+                o.dims.capacityTons,
+              ),
+            }))
+            .filter((o) => o.free > 0 && o.takePerWagon > 0)
+            .sort((a, b) => b.takePerWagon - a.takePerWagon)[0];
+          if (!best) return null;
+          return {
+            c,
+            leg,
+            room: { ...room, wagons: Math.min(room.wagons, best.free) },
+            seat: {
+              wagonTypeId: best.wagonTypeId,
+              perWagon: { ...best.dims, capacityTons: best.takePerWagon },
+            },
+          };
+        }
         const physical = wagonTypeIds.length
           ? c.stock?.availableFor(wagonTypeIds, leg)
           : undefined;
         const wagons =
           physical == null ? room.wagons : Math.min(room.wagons, physical);
-        return { c, leg, room: { ...room, wagons } };
+        return { c, leg, room: { ...room, wagons }, seat: undefined };
       })
       .filter((x): x is NonNullable<typeof x> => x != null && x.room.wagons >= 1)
       .sort((a, b) => b.room.wagons - a.room.wagons)[0];
@@ -2749,10 +2898,15 @@ export class BookingBatchService implements OnModuleInit {
       target.c.id,
       target.room,
       need,
+      target.seat,
     );
     if (!offered) return false;
     target.c.budget.subtract(offered, target.leg);
-    target.c.stock?.consume(wagonTypeIds, offered.wagons, target.leg);
+    target.c.stock?.consume(
+      target.seat ? [target.seat.wagonTypeId] : wagonTypeIds,
+      offered.wagons,
+      target.leg,
+    );
     target.c.armed = true;
     return true;
   }
@@ -2767,6 +2921,12 @@ export class BookingBatchService implements OnModuleInit {
     scheduleId: string,
     budget: Capacity,
     need: Capacity,
+    /**
+     * Capped-bulk seating (see maybeOfferPartial): the ONE wagon type this
+     * offer rides, with capacityTons already reduced to the cargo's per-wagon
+     * cap — so the offered tonnage is what those wagons can really carry.
+     */
+    seat?: { wagonTypeId: string; perWagon: PerWagonDims },
   ): Promise<Capacity | null> {
     if (!this.splitService) return null;
     // A consolidated booking is already half of a shared wagon — never split it.
@@ -2784,8 +2944,14 @@ export class BookingBatchService implements OnModuleInit {
     // measured on the booking's REAL wagon type — the same one allocation
     // validates against. Bulk splits ride FULL wagons only: the offer never
     // part-loads its last wagon.
-    const perWagon = this.dimsFor(booking, wagonDims);
-    const partial = sizePartialOfferWagons(budget, need.wagons, perWagon, {
+    const perWagon = seat?.perWagon ?? this.dimsFor(booking, wagonDims);
+    // With a capped seat, the whole booking's wagon count follows the cap too
+    // (695T at 30T/wagon = 24, not 10 at the raw rating) — the offer must be a
+    // strict subset of THAT count.
+    const wholeWagons = seat
+      ? Math.max(1, Math.ceil(bookingCargoTons(booking) / perWagon.capacityTons))
+      : need.wagons;
+    const partial = sizePartialOfferWagons(budget, wholeWagons, perWagon, {
       fullWagonsOnly: booking.freightType === "BULK",
     });
     if (!partial) return null;
@@ -2793,7 +2959,7 @@ export class BookingBatchService implements OnModuleInit {
     const sized = await this.splitService.sizeOffer(
       booking,
       partial.wagons,
-      need.wagons,
+      wholeWagons,
       perWagon.capacityTons,
       partial.maxCargoTons,
     );
@@ -2832,8 +2998,9 @@ export class BookingBatchService implements OnModuleInit {
    * Settle a schedule's reserved bookings. `expireUnpaidUnknownDeadline` decides
    * how to treat a reservation with no deadline (durable path: leave it; timeout
    * path: expire it). Consolidated pairs settle atomically: both allocate only
-   * when both paid; if either partner expires, both expire (a half-paid shared
-   * wagon must not ship). Returns whether anything changed.
+   * when both paid; when neither paid, both expire. A half-paid pair splits:
+   * the paid half keeps the whole wagon, the lapsed half expires and owes the
+   * cancellation fee (expire()'s pair cascade). Returns whether anything changed.
    */
   private async settleReserved(
     scheduleId: string,
@@ -2876,8 +3043,10 @@ export class BookingBatchService implements OnModuleInit {
           await this.allocate(scheduleId, partner, "paid");
           anySettled = true;
         } else if (isExpired(booking) || isExpired(partner)) {
+          // One call is enough: expire()'s pair cascade settles both sides —
+          // both expire when neither paid; a paid half is rescued (keeps the
+          // whole wagon) while the lapsed half expires with its fee.
           await this.expire(booking);
-          await this.expire(partner);
           anySettled = true;
         }
         continue;
@@ -3816,6 +3985,55 @@ export class BookingBatchService implements OnModuleInit {
     booking: Booking,
     reason: "payment" | "no-capacity" = "payment",
   ): Promise<void> {
+    // Consolidated pair: break the link FIRST, then settle each side singly.
+    //  - neither paid  → both expire, no fee.
+    //  - one side paid → the paid half keeps the whole wagon (rescued by the
+    //    paid guard below at no extra cost); the lapsed half expires and owes
+    //    the cancellation fee (the 'partnerLapsed' event opens the fee invoice
+    //    in BookingWagonCancellationService).
+    //  - both paid     → nothing to expire; the paid guard rescues.
+    if (booking.consolidationPartnerId) {
+      const partnerId = booking.consolidationPartnerId;
+      const bookingRepo = this.dataSource.getRepository(Booking);
+      const partnerRow = await bookingRepo.findOne({
+        where: { id: partnerId },
+        relations: { company: true },
+      });
+      const freshSelf = await bookingRepo.findOne({
+        where: { id: booking.id },
+      });
+      const paidOf = (b: Booking | null) =>
+        b != null && (b.paymentStatus === "PAID" || b.status === "PAID");
+      const selfPaid = paidOf(freshSelf);
+      const partnerPaid = paidOf(partnerRow);
+
+      await this.bookingsRepository.clearConsolidationPair(
+        booking.id,
+        partnerId,
+      );
+      booking.consolidationPartnerId = null;
+      if (partnerRow) partnerRow.consolidationPartnerId = null;
+
+      if (selfPaid && !partnerPaid) {
+        // Wrong side called first: the lapsed partner is the one that expires
+        // (with its fee); this paid booking falls through to the rescue below.
+        if (partnerRow && !["EXPIRED", "CANCELLED"].includes(partnerRow.status)) {
+          this.eventEmitter?.emit("booking.consolidation.partnerLapsed", {
+            expiredBookingId: partnerRow.id,
+          });
+          await this.expire(partnerRow, reason);
+        }
+      } else if (!selfPaid && partnerPaid) {
+        this.eventEmitter?.emit("booking.consolidation.partnerLapsed", {
+          expiredBookingId: booking.id,
+        });
+        // fall through: this side expires below; the paid partner is untouched.
+      } else if (!selfPaid && !partnerPaid) {
+        if (partnerRow && !["EXPIRED", "CANCELLED"].includes(partnerRow.status)) {
+          await this.expire(partnerRow, reason);
+        }
+      }
+    }
     if (!booking.consolidationPartnerId) {
       const fresh = await this.dataSource
         .getRepository(Booking)
@@ -4097,7 +4315,50 @@ export class BookingBatchService implements OnModuleInit {
     // and push once per schedule after the sweep (most unaccepted rows are
     // unpinned under day-level pooling, so this usually emits nothing).
     const touchedScheduleIds = new Set<string>();
+    const swept = new Set<string>();
     for (const booking of unaccepted) {
+      if (swept.has(booking.id)) continue;
+      swept.add(booking.id);
+      // Consolidated pair: the partner may sit outside this route-day's result
+      // set (different yards/day/status), so cascade explicitly — an unpaid
+      // partner expires with this booking; a PAID partner keeps the whole
+      // wagon and this booking owes the cancellation fee (partnerLapsed).
+      if (booking.consolidationPartnerId) {
+        const partner = await this.dataSource.getRepository(Booking).findOne({
+          where: { id: booking.consolidationPartnerId },
+          relations: { company: true },
+        });
+        await this.bookingsRepository.clearConsolidationPair(
+          booking.id,
+          booking.consolidationPartnerId,
+        );
+        booking.consolidationPartnerId = null;
+        if (partner) {
+          const partnerPaid =
+            partner.paymentStatus === "PAID" || partner.status === "PAID";
+          if (partnerPaid) {
+            this.eventEmitter?.emit("booking.consolidation.partnerLapsed", {
+              expiredBookingId: booking.id,
+            });
+          } else if (!["EXPIRED", "CANCELLED"].includes(partner.status)) {
+            swept.add(partner.id);
+            partner.consolidationPartnerId = null;
+            if (partner.trainScheduleId) touchedScheduleIds.add(partner.trainScheduleId);
+            await this.bookingsRepository.update(partner.id, {
+              status: "EXPIRED",
+              schedulingStatus: "ELIGIBLE",
+              scheduledDate: null,
+            } as never);
+            await this.billing
+              .expirePayable(Freight.InvoiceSource.Booking, partner.id, "PREPAID")
+              .catch(() => undefined);
+            this.notifier.expired(partner);
+            this.logger.log(
+              `[BATCH] EXPIRED (unaccepted, with consolidation partner) ${partner.reference}:${partner.id} at doc-review end`,
+            );
+          }
+        }
+      }
       if (booking.trainScheduleId) touchedScheduleIds.add(booking.trainScheduleId);
       await this.bookingsRepository.update(booking.id, {
         status: "EXPIRED",
@@ -4798,12 +5059,34 @@ export class BookingBatchService implements OnModuleInit {
       this.loadAllowedWagonTypeIds(),
     ]);
     const anyType = [...stock.remainingByTypeId.keys()];
-    for (const b of await this.committedBookings(schedule, excludeBookingIds)) {
+    const committed = await this.committedBookings(schedule, excludeBookingIds);
+    // Debit committed PER_TON bulk the way it was SEATED — per type at the
+    // cargo's caps, scarcest type first — not a one-type wagon count drained
+    // deepest-first (which mis-charged 695T Perishable as 24 NW5 when it holds
+    // 10 PW2 + 17 NW5, so later passes over-counted free PW2 and sold NW5 that
+    // were already spoken for).
+    const rank = this.scarcityRankForPool(committed, allowed);
+    for (const b of committed) {
       const typeIds = this.allowedWagonTypeIdsFor(b, allowed);
+      const leg = budget.legForYards(b.originYardId, b.destinationYardId);
+      const perItemBulk =
+        Number(b.bulkTotalWeightTons ?? 0) > 0 &&
+        Number(b.cargoTotalWeightVgm ?? 0) > 0;
+      if (b.freightType === "BULK" && !perItemBulk && typeIds.length) {
+        const smart = this.smartBulkNeed(b, wagonDims, ledger, leg, rank);
+        if (smart) {
+          for (const part of smart.perType) {
+            ledger.consume([part.wagonTypeId], part.wagons, leg);
+          }
+          continue;
+        }
+        // Over-committed (stock cannot seat it any more) — drain what exists,
+        // same as before, so the shortage stays visible to the gates.
+      }
       ledger.consume(
         typeIds.length ? typeIds : anyType,
         this.wagonsFor(b, wagonDims),
-        budget.legForYards(b.originYardId, b.destinationYardId),
+        leg,
       );
     }
     return ledger;
@@ -4825,6 +5108,112 @@ export class BookingBatchService implements OnModuleInit {
     return stock.availableFor(wagonTypeIds, leg) >= wagonsNeeded;
   }
 
+  /**
+   * Scarcity rank over the day pool: how many distinct demand groups (bulk
+   * cargo types / container types among these bookings) may ride each wagon
+   * type. The batch seats least-shareable types first, so bulk with a
+   * bulk-only alternative (PW2) never eats the container-capable stock (NW5)
+   * that containers cannot substitute.
+   */
+  private scarcityRankForPool(
+    pool: Booking[],
+    allowed: {
+      byCargoTypeId: Map<string, string[]>;
+      byContainerTypeId: Map<string, string[]>;
+    },
+  ): Map<string, number> {
+    const groups = new Map<string, string[]>();
+    for (const b of pool) {
+      if (b.freightType === "BULK") {
+        const cargoTypeId = b.cargoTypeId ?? b.cargoType?.id;
+        if (cargoTypeId) {
+          groups.set(`B:${cargoTypeId}`, allowed.byCargoTypeId.get(cargoTypeId) ?? []);
+        }
+      } else {
+        for (const line of b.bookingContainers ?? []) {
+          const containerTypeId = line.containerTypeId ?? line.containerType?.id;
+          if (containerTypeId) {
+            groups.set(
+              `C:${containerTypeId}`,
+              allowed.byContainerTypeId.get(containerTypeId) ?? [],
+            );
+          }
+        }
+      }
+    }
+    const rank = new Map<string, number>();
+    for (const ids of groups.values()) {
+      for (const id of ids) rank.set(id, (rank.get(id) ?? 0) + 1);
+    }
+    return rank;
+  }
+
+  /**
+   * Cap-aware, scarcity-ordered seating of a PER_TON bulk booking across the
+   * wagon types this train actually has free on its leg — the same policy the
+   * wagon planner applies at allocation time (least-shareable type first, each
+   * wagon filled to the cargo type's per-wagon cap, one booking per wagon).
+   *
+   * This is the payment gate's real fit check for bulk: the generic
+   * `hasWagonStock` sums free wagons across allowed types against a count
+   * sized on ONE type, so 695T Perishable read "24 wagons needed, 28 free"
+   * when seating it across 10 PW2 (20T) + NW5 (30T) really takes 27 wagons.
+   * Returns the exact per-type counts and the three-axis capacity they
+   * consume, or null when the free stock cannot seat the whole booking.
+   */
+  private smartBulkNeed(
+    booking: Booking,
+    wagonDims: WagonDims,
+    stock: WagonStockLedger,
+    leg: CorridorLeg,
+    scarcityRank: Map<string, number>,
+  ): { need: Capacity; perType: Array<{ wagonTypeId: string; wagons: number }> } | null {
+    const options = this.allowedDimsWithTypes(booking, wagonDims)
+      .filter((o): o is { wagonTypeId: string; dims: PerWagonDims } => o.wagonTypeId != null)
+      .map((o) => ({
+        ...o,
+        free: stock.availableFor([o.wagonTypeId], leg),
+        takePerWagon: bulkTonsPerWagon(
+          booking.cargoType,
+          o.wagonTypeId,
+          o.dims.capacityTons,
+        ),
+      }))
+      .filter((o) => o.free > 0 && o.takePerWagon > 0)
+      .sort(
+        (a, b) =>
+          (scarcityRank.get(a.wagonTypeId) ?? 1) -
+            (scarcityRank.get(b.wagonTypeId) ?? 1) ||
+          b.takePerWagon - a.takePerWagon,
+      );
+
+    let remaining = bookingCargoTons(booking);
+    if (remaining <= 0) return null;
+    const perType: Array<{ wagonTypeId: string; wagons: number }> = [];
+    let weightTons = remaining; // gross: cargo plus each seated wagon's tare
+    let lengthMeters = 0;
+    let wagons = 0;
+    for (const option of options) {
+      if (remaining <= 1e-9) break;
+      const take = Math.min(option.free, Math.ceil(remaining / option.takePerWagon));
+      if (take <= 0) continue;
+      remaining = roundTons(Math.max(0, remaining - take * option.takePerWagon));
+      wagons += take;
+      weightTons += take * option.dims.tareWeightTons;
+      lengthMeters += take * option.dims.lengthMeters;
+      perType.push({ wagonTypeId: option.wagonTypeId, wagons: take });
+    }
+    if (remaining > 1e-9) return null;
+    return {
+      need: {
+        wagons,
+        weightTons: roundTons(weightTons),
+        lengthMeters: roundTons(lengthMeters),
+      },
+      perType,
+    };
+  }
+
   private allowedWagonTypeCache: {
     byCargoTypeId: Map<string, string[]>;
     byContainerTypeId: Map<string, string[]>;