feat: express-batch dispatch — manual trigger for the AI agent

Adds the backend half of the ExpressDispatchAgent flow. Express orders can now
be created batch by batch (bulk create only accumulates them, unassigned), then
an operator hits one endpoint to hand the whole pending set to the agent for
tenant-scoped assignment + road sequencing. The normal B2C flow is untouched.

- POST /admin/expressbooking/dispatch: manual trigger. Console-auth, tenant-
  scoped; gathers the tenant's pending unassigned express orders (or a chosen
  subset) and publishes express.dispatch_requested.
- internal API for the agent: GET /internal/express/riders (tenant's available
  riders), GET /internal/express/bookings, POST /internal/express/assign (writes
  the agent's decided assignments with their sequence; re-checks the already-
  assigned guard so the agent can't double-assign).
- booking_assignment_service.go: extracted a behavior-preserving assignMilerTx
  core; AssignMilerToBooking is unchanged in behavior. assignExpressStops writes
  a batch, one FCM per rider instead of one per stop.
- EXPRESS JetStream stream / express.dispatch_requested subject.
- Gated behind EXPRESS_AGENT_ENABLED (default off): deploying this changes
  nothing until the agent is confirmed running and the flag is flipped.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

constants/constants.go

@@ -13,6 +13,13 @@ const (
 	MilerBlocked    = "Blocked"
 )
 
+// Booking sources — where a booking originated. Left as their stored literals:
+// "CRM_Console" predates the outward express rename and is an existing DB value.
+const (
+	BookingSourceCustomerApp = "Customer_App"
+	BookingSourceExpress     = "CRM_Console"
+)
+
 // App (B2C) Customer Statuses
 const (
 	CustomerStatusActive  = "Active"

controllers/adminController.go

@@ -2150,7 +2150,12 @@ func (e *expressBookingValidationError) Error() string { return e.msg }
 // AdminBulkCreateBookings (many, used by CSV/bulk import). Takes no
 // *fiber.Ctx — the original function never touched c after BodyParser, so
 // both callers can use this identically.
-func createExpressBooking(req AdminBookingRequest) (*models.PickupBooking, error) {
+// createExpressBooking creates one console booking. autoAssign controls whether
+// it kicks off the per-booking assignment engine (AssignCRMMiler) inline. Single
+// bookings assign inline; a bulk batch passes false, because the whole batch is
+// handed to the ExpressDispatchAgent as one unit instead — assigning inline as
+// well would assign every booking twice by two different logics.
+func createExpressBooking(req AdminBookingRequest, autoAssign bool) (*models.PickupBooking, error) {
 	if len(req.Parcels) == 0 {
 		return nil, &expressBookingValidationError{"at least one parcel is required"}
 	}
@@ -2386,7 +2391,9 @@ func createExpressBooking(req AdminBookingRequest) (*models.PickupBooking, error
 		return nil, fmt.Errorf("failed to create booking")
 	}
 
-	go assignment.AssignCRMMiler(booking.Bookingid)
+	if autoAssign {
+		go assignment.AssignCRMMiler(booking.Bookingid)
+	}
 
 	if db.Js != nil {
 		payload := map[string]interface{}{
@@ -2421,7 +2428,7 @@ func CreateExpressBooking(c *fiber.Ctx) error {
 		req.Tenantid = own
 	}
 
-	booking, err := createExpressBooking(*req)
+	booking, err := createExpressBooking(*req, true)
 	if err != nil {
 		if _, ok := err.(*expressBookingValidationError); ok {
 			return utils.BadRequest(c, err.Error())
@@ -2461,13 +2468,20 @@ func AdminBulkCreateBookings(c *fiber.Ctx) error {
 
 	ownTenant := consoleTenantID(c)
 
+	// When the express agent is enabled, a bulk create only *accumulates* pending
+	// bookings — inline per-booking assignment is suppressed so orders can pile up
+	// batch by batch, and the operator later fires the agent over the whole lot
+	// with POST /admin/expressbooking/dispatch. When disabled (default), nothing
+	// changes from the old behavior: each booking auto-assigns inline.
+	agentEnabled := expressAgentEnabled()
+
 	for i, item := range req.Bookings {
 		// Same tenant pin as the single-booking path — a bulk import must not
 		// be a way around it.
 		if ownTenant != 0 {
 			item.Tenantid = ownTenant
 		}
-		booking, err := createExpressBooking(item)
+		booking, err := createExpressBooking(item, !agentEnabled)
 		if err != nil {
 			results = append(results, result{Index: i, Success: false, Error: err.Error()})
 			continue

controllers/booking_assignment_service.go

@@ -10,18 +10,32 @@ import (
 	"doormile/internal/notify"
 	"doormile/models"
 	"doormile/utils"
+
+	"gorm.io/gorm"
 )
 
-// AssignMilerToBooking is the single source of truth for manually assigning a
+// milerStopSequence carries the road-optimized ordering the Route Optimization
-// miler to a pickup booking — shared by the admin console (AdminAssignMiler)
+// API produced for one stop. Passed to assignMilerTx when the caller already
-// and the hub console (HubAssignMiler) so both go through identical DB
+// knows the sequence (the express-batch path), nil when it does not (a plain
-// updates, NATS publish, and FCM notify instead of duplicating the logic.
+// manual assignment, which leaves the stop unsequenced at step 0).
-func AssignMilerToBooking(bookingID, milerUserID int, assignedByUserID *int) (*models.PickupBooking, error) {
+type milerStopSequence struct {
-	tx := db.DB.Begin()
+	Step          int
+	Previouskms   float64
+	Cumulativekms float64
+	Etaminutes    int
+	Cumulativeeta int
+}
 
+// assignMilerTx performs the DB half of a miler assignment inside the given
+// transaction: flip the booking to Miler_Assigned, create the BookingAssignment
+// (optionally already sequenced), and mark the miler Assigned. It does not
+// commit, publish, or notify — those are the caller's job, so the same writes
+// can be reused by the single-booking path (AssignMilerToBooking) and the
+// batch express path (assignExpressStops) without duplicating the SQL or firing
+// one notification per stop.
+func assignMilerTx(tx *gorm.DB, bookingID, milerUserID int, assignedByUserID *int, seq *milerStopSequence) (*models.PickupBooking, error) {
 	var booking models.PickupBooking
 	if err := tx.First(&booking, bookingID).Error; err != nil {
-		tx.Rollback()
 		return nil, fmt.Errorf("booking not found")
 	}
 
@@ -29,7 +43,6 @@ func AssignMilerToBooking(bookingID, milerUserID int, assignedByUserID *int) (*m
 	booking.Assignedmileruserid = &milerUserID
 	booking.Updatedat = time.Now()
 	if err := tx.Save(&booking).Error; err != nil {
-		tx.Rollback()
 		return nil, fmt.Errorf("failed to update booking: %w", err)
 	}
 
@@ -39,21 +52,31 @@ func AssignMilerToBooking(bookingID, milerUserID int, assignedByUserID *int) (*m
 		Assignedbyuserid: assignedByUserID,
 		Assignmentstatus: constants.AssignmentAssigned,
 	}
+	if seq != nil {
+		now := time.Now()
+		assignment.Step = seq.Step
+		assignment.Previouskms = seq.Previouskms
+		assignment.Cumulativekms = seq.Cumulativekms
+		assignment.Etaminutes = seq.Etaminutes
+		assignment.Cumulativeeta = seq.Cumulativeeta
+		assignment.Sequencedat = &now
+	}
 	if err := tx.Create(&assignment).Error; err != nil {
-		tx.Rollback()
 		return nil, fmt.Errorf("failed to create assignment: %w", err)
 	}
 
 	if err := tx.Model(&models.MilerProfile{}).Where("userid = ?", milerUserID).
 		Update("availabilitystatus", constants.MilerAssigned).Error; err != nil {
-		tx.Rollback()
 		return nil, fmt.Errorf("failed to update miler availability: %w", err)
 	}
 
-	if err := tx.Commit().Error; err != nil {
+	return &booking, nil
-		return nil, fmt.Errorf("failed to commit miler assignment: %w", err)
+}
-	}
 
+// publishAssignmentUpdate emits the best-effort booking.update NATS event and
+// pushes the FCM notification for one assignment. Shared by both assignment
+// paths so the outward side effects stay identical.
+func publishAssignmentUpdate(booking *models.PickupBooking, milerUserID int) {
 	if db.Js != nil {
 		payload := map[string]interface{}{
 			"booking_id": booking.Bookingid,
@@ -77,10 +100,137 @@ func AssignMilerToBooking(bookingID, milerUserID int, assignedByUserID *int) (*m
 			"New booking assigned — tap to view details",
 			map[string]string{"booking_id": fmt.Sprintf("%d", booking.Bookingid)},
 		); err != nil {
-			utils.Warn("FCM: failed to notify miler on manual assignment",
+			utils.Warn("FCM: failed to notify miler on assignment",
 				"miler_id", milerUserID, "booking_id", booking.Bookingid, "error", err)
 		}
 	}
-
+}
-	return &booking, nil
+
+// AssignMilerToBooking is the single source of truth for manually assigning a
+// miler to a pickup booking — shared by the admin console (AdminAssignMiler)
+// and the hub console (HubAssignMiler) so both go through identical DB
+// updates, NATS publish, and FCM notify instead of duplicating the logic.
+func AssignMilerToBooking(bookingID, milerUserID int, assignedByUserID *int) (*models.PickupBooking, error) {
+	tx := db.DB.Begin()
+
+	booking, err := assignMilerTx(tx, bookingID, milerUserID, assignedByUserID, nil)
+	if err != nil {
+		tx.Rollback()
+		return nil, err
+	}
+
+	if err := tx.Commit().Error; err != nil {
+		return nil, fmt.Errorf("failed to commit miler assignment: %w", err)
+	}
+
+	publishAssignmentUpdate(booking, milerUserID)
+	return booking, nil
+}
+
+// ExpressStop is one already-decided assignment from the ExpressDispatchAgent:
+// which miler carries which booking, in what road-optimized order. The agent
+// chose the miler and called the Route Optimization API for the sequence; this
+// struct is the writeback contract.
+type ExpressStop struct {
+	BookingID     int     `json:"booking_id"`
+	MilerUserID   int     `json:"miler_user_id"`
+	Step          int     `json:"step"`
+	Previouskms   float64 `json:"previouskms"`
+	Cumulativekms float64 `json:"cumulativekms"`
+	Etaminutes    int     `json:"etaminutes"`
+	Cumulativeeta int     `json:"cumulativeeta"`
+}
+
+// ExpressAssignResult is the per-booking outcome of a batch writeback.
+type ExpressAssignResult struct {
+	BookingID   int    `json:"booking_id"`
+	MilerUserID int    `json:"miler_user_id"`
+	Success     bool   `json:"success"`
+	Error       string `json:"error,omitempty"`
+}
+
+// assignExpressStops writes a batch of agent-decided assignments. Each stop is
+// its own transaction so one bad booking id cannot roll back the whole batch —
+// the same per-row-independence the bulk-create endpoint gives. Each carries its
+// sequence, so the assignment lands already ordered rather than needing a second
+// sequencing pass. A miler is notified once for the whole batch, not once per
+// stop, so a rider handed five stops gets one push, not five.
+func assignExpressStops(stops []ExpressStop) []ExpressAssignResult {
+	results := make([]ExpressAssignResult, 0, len(stops))
+	// Preserve first-seen miler order so the summary notification is deterministic.
+	notifyBooking := map[int]*models.PickupBooking{}
+	notifyOrder := []int{}
+
+	for _, s := range stops {
+		tx := db.DB.Begin()
+		booking, err := assignMilerTx(tx, s.BookingID, s.MilerUserID, nil, &milerStopSequence{
+			Step:          s.Step,
+			Previouskms:   s.Previouskms,
+			Cumulativekms: s.Cumulativekms,
+			Etaminutes:    s.Etaminutes,
+			Cumulativeeta: s.Cumulativeeta,
+		})
+		if err != nil {
+			tx.Rollback()
+			results = append(results, ExpressAssignResult{
+				BookingID: s.BookingID, MilerUserID: s.MilerUserID, Success: false, Error: err.Error()})
+			continue
+		}
+		if err := tx.Commit().Error; err != nil {
+			results = append(results, ExpressAssignResult{
+				BookingID: s.BookingID, MilerUserID: s.MilerUserID, Success: false, Error: "commit failed"})
+			continue
+		}
+
+		// booking.update per stop keeps live trackers accurate; the FCM push is
+		// deferred and coalesced per miler below.
+		if db.Js != nil {
+			payload := map[string]interface{}{
+				"booking_id": booking.Bookingid,
+				"booking_no": booking.Bookingno,
+				"status":     booking.Status,
+				"miler_id":   s.MilerUserID,
+				"updated_at": time.Now().UnixMilli(),
+			}
+			if data, err := json.Marshal(payload); err == nil {
+				if _, err := db.Js.Publish("api.v1.bookings.update", data); err != nil {
+					utils.Warn("Failed to publish booking.update to NATS", "booking_id", booking.Bookingid, "error", err)
+				}
+			}
+		}
+
+		if _, seen := notifyBooking[s.MilerUserID]; !seen {
+			notifyOrder = append(notifyOrder, s.MilerUserID)
+		}
+		notifyBooking[s.MilerUserID] = booking
+		results = append(results, ExpressAssignResult{
+			BookingID: s.BookingID, MilerUserID: s.MilerUserID, Success: true})
+	}
+
+	for _, milerUserID := range notifyOrder {
+		count := 0
+		for _, r := range results {
+			if r.MilerUserID == milerUserID && r.Success {
+				count++
+			}
+		}
+		var miler models.MilerProfile
+		if db.DB.Where("userid = ?", milerUserID).First(&miler).Error == nil && miler.Devicetoken != "" {
+			msg := "New pickups assigned — tap to view your route"
+			if count == 1 {
+				msg = "New booking assigned — tap to view details"
+			}
+			if err := notify.SendToDevice(
+				miler.Devicetoken,
+				"New Pickups Assigned",
+				msg,
+				map[string]string{"count": fmt.Sprintf("%d", count)},
+			); err != nil {
+				utils.Warn("FCM: failed to notify miler on express batch assignment",
+					"miler_id", milerUserID, "error", err)
+			}
+		}
+	}
+
+	return results
 }

controllers/expressDispatchController.go

@@ -0,0 +1,317 @@
+package controllers
+
+import (
+	"encoding/json"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"doormile/constants"
+	"doormile/db"
+	"doormile/models"
+	"doormile/utils"
+
+	"github.com/gofiber/fiber/v2"
+)
+
+// expressAgentEnabled gates the whole express-batch handoff. Default OFF, so
+// deploying this code changes nothing: a bulk create keeps assigning each
+// booking inline exactly as before, and no batch event is emitted. Flip
+// EXPRESS_AGENT_ENABLED=true ONLY once the ExpressDispatchAgent is confirmed
+// running and consuming express.batch_created — otherwise batches would suppress
+// inline assignment with nothing to pick them up, and bulk bookings would sit
+// unassigned. Read at request time so it can be toggled without a redeploy.
+func expressAgentEnabled() bool {
+	return strings.EqualFold(os.Getenv("EXPRESS_AGENT_ENABLED"), "true")
+}
+
+// expressDispatchSubject is the JetStream subject the ExpressDispatchAgent binds
+// to. It must be present on the EXPRESS stream (db/streams.go) before anything
+// publishes here — JetStream silently drops messages on uncovered subjects.
+const expressDispatchSubject = "express.dispatch_requested"
+
+// publishExpressBatch emits one express.dispatch_requested event per tenant when
+// an operator dispatches their pending orders. Best-effort like every publish in
+// this codebase: a NATS outage degrades to hand-assignment from the console, it
+// never fails the request.
+func publishExpressBatch(byTenant map[int][]int) {
+	if db.Js == nil {
+		if len(byTenant) > 0 {
+			utils.Warn("Express: JetStream unavailable, dispatch not handed to agent",
+				"tenants", len(byTenant))
+		}
+		return
+	}
+	for tenantID, bookingIDs := range byTenant {
+		if len(bookingIDs) == 0 {
+			continue
+		}
+		payload := map[string]interface{}{
+			"tenantid":    tenantID,
+			"booking_ids": bookingIDs,
+			"created_at":  time.Now().UnixMilli(),
+		}
+		data, err := json.Marshal(payload)
+		if err != nil {
+			utils.Warn("Express: failed to marshal dispatch event", "tenantid", tenantID, "error", err)
+			continue
+		}
+		if _, err := db.Js.Publish(expressDispatchSubject, data); err != nil {
+			utils.Warn("Express: failed to publish dispatch event", "tenantid", tenantID, "error", err)
+			continue
+		}
+		utils.Info("Express: published "+expressDispatchSubject,
+			"tenantid", tenantID, "bookings", len(bookingIDs))
+	}
+}
+
+// This file is the internal API surface the ExpressDispatchAgent (logistics-ai)
+// uses to run the express-batch flow: read the tenant's available riders, read
+// the batch's bookings, and write back the assignments it decided after calling
+// the Route Optimization API. All three sit under /internal (InternalKeyAuth),
+// never exposed to a console or app token.
+//
+// The division of labour: the agent decides *who* and *what order* (rider pool +
+// routes.workolik). Go stays the single writer of assignment state — the agent
+// never writes the DB directly, it posts its decision here and this reuses the
+// same transactional assignment path the consoles use.
+
+// DispatchExpressBatch is the manual trigger the console operator hits once a
+// batch of express orders has piled up. Orders are created batch by batch (bulk
+// create only accumulates them, unassigned); this hands the whole pending set
+// for the tenant to the ExpressDispatchAgent in one go — the "take over from
+// here" button.
+//
+// Console auth, tenant-scoped: a client login dispatches only its own pending
+// orders; Doormile staff pass ?tenantid= to dispatch for one client. An optional
+// body {"booking_ids":[...]} dispatches a chosen subset instead of everything
+// pending — always still pinned to the resolved tenant so no cross-tenant id can
+// be smuggled in.
+func DispatchExpressBatch(c *fiber.Ctx) error {
+	tenantID, allowed := effectiveTenantID(c)
+	if !allowed {
+		return utils.Forbidden(c, "you can only dispatch your own tenant")
+	}
+	if tenantID == 0 {
+		return utils.BadRequest(c, "tenantid is required (staff must pass ?tenantid=)")
+	}
+
+	var body struct {
+		BookingIDs []int `json:"booking_ids"`
+	}
+	_ = c.BodyParser(&body) // body is optional
+
+	// Only ever the tenant's own, unassigned, still-pending express orders.
+	q := db.DB.Model(&models.PickupBooking{}).
+		Where("tenantid = ? AND bookingsource = ? AND assignedmileruserid IS NULL AND status = ?",
+			tenantID, constants.BookingSourceExpress, constants.BookingPendingPickup)
+	if len(body.BookingIDs) > 0 {
+		q = q.Where("bookingid IN ?", body.BookingIDs)
+	}
+
+	var bookingIDs []int
+	if err := q.Pluck("bookingid", &bookingIDs).Error; err != nil {
+		return utils.Internal(c, "failed to gather pending bookings")
+	}
+	if len(bookingIDs) == 0 {
+		return utils.OK(c, fiber.Map{
+			"queued":      0,
+			"booking_ids": []int{},
+			"message":     "no pending express bookings to dispatch",
+		})
+	}
+
+	publishExpressBatch(map[int][]int{tenantID: bookingIDs})
+
+	resp := fiber.Map{
+		"queued":      len(bookingIDs),
+		"booking_ids": bookingIDs,
+	}
+	if !expressAgentEnabled() {
+		// The batch was published, but with the flag off the bulk path may have
+		// already assigned these inline and the agent may not be consuming — make
+		// that visible rather than implying work was dispatched.
+		resp["warning"] = "EXPRESS_AGENT_ENABLED is off; the agent may not be consuming this event"
+	}
+	return utils.OK(c, resp)
+}
+
+// GetExpressRiders returns a tenant's riders that are free to take work, with the
+// location and hub the agent needs to distribute stops. Optional ?city= narrows
+// to one operating zone (applocationid) so a Coimbatore batch is never handed to
+// a Nagercoil rider.
+func GetExpressRiders(c *fiber.Ctx) error {
+	tenantID := c.QueryInt("tenantid", 0)
+	if tenantID == 0 {
+		return utils.BadRequest(c, "tenantid is required")
+	}
+
+	type riderRow struct {
+		Userid             int     `json:"miler_user_id"`
+		Displayname        string  `json:"displayname"`
+		Phone              string  `json:"phone"`
+		Currentlatitude    float64 `json:"latitude"`
+		Currentlongitude   float64 `json:"longitude"`
+		Hubid              *int    `json:"hubid"`
+		Applocationid      int     `json:"applocationid"`
+		Availabilitystatus string  `json:"availabilitystatus"`
+		Devicetoken        string  `json:"has_device_token"`
+	}
+
+	q := db.DB.Table("milerprofiles AS mp").
+		Select(`mp.userid, mp.displayname, mp.phone, mp.currentlatitude,
+		        mp.currentlongitude, mp.hubid, mp.applocationid,
+		        mp.availabilitystatus, mp.device_token`).
+		Joins("JOIN appusers AS u ON u.userid = mp.userid").
+		Where("u.tenantid = ? AND u.roleid = ? AND mp.availabilitystatus = ?",
+			tenantID, 5, constants.MilerAvailable)
+
+	if city := c.QueryInt("city", 0); city != 0 {
+		q = q.Where("mp.applocationid = ?", city)
+	}
+
+	var rows []riderRow
+	if err := q.Scan(&rows).Error; err != nil {
+		return utils.Internal(c, "failed to load riders")
+	}
+
+	// Expose only whether a device token exists, never the token itself.
+	out := make([]fiber.Map, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, fiber.Map{
+			"miler_user_id":      r.Userid,
+			"displayname":        r.Displayname,
+			"phone":              r.Phone,
+			"latitude":           r.Currentlatitude,
+			"longitude":          r.Currentlongitude,
+			"hubid":              r.Hubid,
+			"applocationid":      r.Applocationid,
+			"availabilitystatus": r.Availabilitystatus,
+			"has_device_token":   r.Devicetoken != "",
+		})
+	}
+	return c.JSON(fiber.Map{"success": true, "riders": out, "total": len(out)})
+}
+
+// GetExpressBookings returns the coordinates and kitchen for a set of booking
+// ids — everything the agent needs to feed the optimizer, and nothing it does
+// not. ?ids=1,2,3.
+func GetExpressBookings(c *fiber.Ctx) error {
+	idsParam := c.Query("ids")
+	if idsParam == "" {
+		return utils.BadRequest(c, "ids is required, e.g. ?ids=1,2,3")
+	}
+
+	ids := make([]int, 0)
+	for _, part := range strings.Split(idsParam, ",") {
+		part = strings.TrimSpace(part)
+		if part == "" {
+			continue
+		}
+		n, err := strconv.Atoi(part)
+		if err != nil {
+			return utils.BadRequest(c, "ids must be a comma-separated list of integers")
+		}
+		ids = append(ids, n)
+	}
+	if len(ids) == 0 {
+		return utils.BadRequest(c, "ids is required")
+	}
+
+	type bookingRow struct {
+		Bookingid           int     `json:"booking_id"`
+		Bookingno           string  `json:"booking_no"`
+		Tenantid            *int    `json:"tenantid"`
+		Tenantlocationid    *int    `json:"tenantlocationid"`
+		Pickuplatitude      float64 `json:"pickuplatitude"`
+		Pickuplongitude     float64 `json:"pickuplongitude"`
+		Deliverylatitude    float64 `json:"deliverylatitude"`
+		Deliverylongitude   float64 `json:"deliverylongitude"`
+		Pickuppincode       string  `json:"pickuppincode"`
+		Deliverypincode     string  `json:"deliverypincode"`
+		Status              string  `json:"status"`
+		Assignedmileruserid *int    `json:"assignedmileruserid"`
+	}
+
+	var rows []bookingRow
+	if err := db.DB.Model(&models.PickupBooking{}).
+		Select(`bookingid, bookingno, tenantid, tenantlocationid,
+		        pickuplatitude, pickuplongitude, deliverylatitude, deliverylongitude,
+		        pickuppincode, deliverypincode, status, assignedmileruserid`).
+		Where("bookingid IN ?", ids).
+		Scan(&rows).Error; err != nil {
+		return utils.Internal(c, "failed to load bookings")
+	}
+
+	return c.JSON(fiber.Map{"success": true, "bookings": rows, "total": len(rows)})
+}
+
+// AssignExpressBatch writes the agent's decided assignments. Body:
+//
+//	{ "assignments": [ {booking_id, miler_user_id, step, previouskms,
+//	                    cumulativekms, etaminutes, cumulativeeta}, ... ] }
+//
+// Each row is assigned in its own transaction with its sequence already set, and
+// each miler is notified once for the whole batch. Per-row results mirror the
+// bulk-create shape so a single bad booking id never fails the batch.
+func AssignExpressBatch(c *fiber.Ctx) error {
+	var req struct {
+		Assignments []ExpressStop `json:"assignments"`
+	}
+	if err := c.BodyParser(&req); err != nil {
+		return utils.BadRequest(c, "invalid request body")
+	}
+	if len(req.Assignments) == 0 {
+		return utils.BadRequest(c, "assignments is required and must not be empty")
+	}
+
+	// Guard against writing to a booking that is not actually the tenant's or is
+	// already assigned — the agent is trusted but the writeback must still be the
+	// place assignment invariants are enforced, not the agent.
+	bookingIDs := make([]int, 0, len(req.Assignments))
+	for _, a := range req.Assignments {
+		bookingIDs = append(bookingIDs, a.BookingID)
+	}
+	assignable := map[int]bool{}
+	var existing []struct {
+		Bookingid           int
+		Assignedmileruserid *int
+		Status              string
+	}
+	db.DB.Model(&models.PickupBooking{}).
+		Select("bookingid, assignedmileruserid, status").
+		Where("bookingid IN ?", bookingIDs).
+		Scan(&existing)
+	for _, b := range existing {
+		assignable[b.Bookingid] = b.Assignedmileruserid == nil &&
+			b.Status != constants.BookingCancelled
+	}
+
+	toAssign := make([]ExpressStop, 0, len(req.Assignments))
+	results := make([]ExpressAssignResult, 0, len(req.Assignments))
+	for _, a := range req.Assignments {
+		if !assignable[a.BookingID] {
+			results = append(results, ExpressAssignResult{
+				BookingID: a.BookingID, MilerUserID: a.MilerUserID,
+				Success: false, Error: "booking already assigned or not eligible"})
+			continue
+		}
+		toAssign = append(toAssign, a)
+	}
+
+	results = append(results, assignExpressStops(toAssign)...)
+
+	assigned := 0
+	for _, r := range results {
+		if r.Success {
+			assigned++
+		}
+	}
+	return c.JSON(fiber.Map{
+		"success":  true,
+		"assigned": assigned,
+		"total":    len(req.Assignments),
+		"results":  results,
+	})
+}

db/streams.go

@@ -44,6 +44,14 @@ var streamSubjects = map[string][]string{
 		// missing from the stream, bookings are never assigned at all.
 		"booking.assignment_requested",
 	},
+	"EXPRESS": {
+		// The console's manual "dispatch" action publishes this; the
+		// ExpressDispatchAgent (logistics-ai) consumes it to run tenant-scoped
+		// batch assignment + route sequencing over the pending orders. Missing
+		// subject → the dispatch is never handed off and bookings sit unassigned
+		// until someone assigns them by hand.
+		"express.dispatch_requested",
+	},
 	"STATUS": {
 		"booking.status.updated",
 		"booking.cancelled",

routes/routes.go

@@ -270,6 +270,10 @@ func RegisterRoutes(app *fiber.App, cfg *config.Config) {
 	adminAuth.Get("/bookings", controllers.GetAdminBookings)
 	adminAuth.Post("/expressbooking", middlewares.CityGateMiddleware, controllers.CreateExpressBooking)
 	adminAuth.Post("/expressbooking/bulk", middlewares.CityGateMiddleware, controllers.AdminBulkCreateBookings)
+	// Manual trigger: hand all pending express orders for the tenant to the
+	// ExpressDispatchAgent (assign + road-sequence). Operator-controlled, after
+	// orders have accumulated batch by batch.
+	adminAuth.Post("/expressbooking/dispatch", controllers.DispatchExpressBatch)
 	adminAuth.Get("/bookings/:id", controllers.GetAdminBookingDetails)
 	adminAuth.Get("/bookings/:id/track", controllers.GetAdminBookingTrack)
 	adminAuth.Post("/bookings/:id/assign-miler", controllers.AdminAssignMiler)
@@ -418,6 +422,12 @@ func RegisterRoutes(app *fiber.App, cfg *config.Config) {
 	internal.Get("/agent-decisions/similar", controllers.FindSimilarDecisions)
 	internal.Patch("/agent-decisions/:id/outcome", controllers.UpdateDecisionOutcome)
 
+	// Express-batch dispatch: the ExpressDispatchAgent reads a tenant's riders
+	// and the batch's bookings, then writes back the assignments it decided.
+	internal.Get("/express/riders", controllers.GetExpressRiders)
+	internal.Get("/express/bookings", controllers.GetExpressBookings)
+	internal.Post("/express/assign", controllers.AssignExpressBatch)
+
 	// --------------------
 	// WEBSOCKET — live miler tracking (no auth, public tracking link)
 	// --------------------