diff --git a/submitqueue/entity/batch.go b/submitqueue/entity/batch.go
index e1b4431f..8975b513 100644
--- a/submitqueue/entity/batch.go
+++ b/submitqueue/entity/batch.go
@@ -187,6 +187,8 @@ func BatchFromBytes(data []byte) (Batch, error) {
 type BatchID struct {
 	// ID is the globally unique identifier for the batch.
 	ID string `json:"id"`
+	// Queue is the name of the queue processing the batch. Empty on payloads written before the field existed.
+	Queue string `json:"queue"`
 }
 
 // ToBytes serializes the BatchID to JSON bytes for queue message payload.
diff --git a/submitqueue/entity/build.go b/submitqueue/entity/build.go
index 11a62c18..b3bbb029 100644
--- a/submitqueue/entity/build.go
+++ b/submitqueue/entity/build.go
@@ -81,6 +81,8 @@ func BuildFromBytes(data []byte) (Build, error) {
 type BuildID struct {
 	// ID is the globally unique identifier for the build.
 	ID string `json:"id"`
+	// Queue is the name of the queue processing the batch this build verifies. Empty on payloads written before the field existed.
+	Queue string `json:"queue"`
 }
 
 // ToBytes serializes the BuildID to JSON bytes for queue message payload.
diff --git a/submitqueue/entity/cancel.go b/submitqueue/entity/cancel.go
index 3937982a..3c87c196 100644
--- a/submitqueue/entity/cancel.go
+++ b/submitqueue/entity/cancel.go
@@ -21,6 +21,8 @@ import "encoding/json"
 type CancelRequest struct {
 	// ID is the globally unique identifier of the request to cancel. Format: "<queue>/<counter_value>".
 	ID string `json:"id"`
+	// Queue is the name of the queue processing the request to cancel. Empty on payloads written before the field existed.
+	Queue string `json:"queue"`
 	// Reason is an optional free-form explanation of why the cancellation was requested.
 	Reason string `json:"reason"`
 }
diff --git a/submitqueue/entity/request.go b/submitqueue/entity/request.go
index d741d1bc..52cbb89c 100644
--- a/submitqueue/entity/request.go
+++ b/submitqueue/entity/request.go
@@ -113,6 +113,8 @@ func RequestFromBytes(data []byte) (Request, error) {
 type RequestID struct {
 	// ID is the globally unique identifier for the land request.
 	ID string `json:"id"`
+	// Queue is the name of the queue processing the land request. Empty on payloads written before the field existed.
+	Queue string `json:"queue"`
 }
 
 // ToBytes serializes the RequestID to JSON bytes for queue message payload.
diff --git a/submitqueue/extension/storage/mock/request_batch_store_mock.go b/submitqueue/extension/storage/mock/request_batch_store_mock.go
index 2c4c15f0..24e87e12 100644
--- a/submitqueue/extension/storage/mock/request_batch_store_mock.go
+++ b/submitqueue/extension/storage/mock/request_batch_store_mock.go
@@ -1,9 +1,9 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: submitqueue/extension/storage/request_batch_store.go
+// Source: request_batch_store.go
 //
 // Generated by this command:
 //
-//	mockgen -source=submitqueue/extension/storage/request_batch_store.go -destination=submitqueue/extension/storage/mock/request_batch_store_mock.go -package=mock
+//	mockgen -source=request_batch_store.go -destination=mock/request_batch_store_mock.go -package=mock
 //
 
 // Package mock is a generated GoMock package.
diff --git a/submitqueue/gateway/controller/cancel.go b/submitqueue/gateway/controller/cancel.go
index ead31cc3..6bfcd60e 100644
--- a/submitqueue/gateway/controller/cancel.go
+++ b/submitqueue/gateway/controller/cancel.go
@@ -86,7 +86,8 @@ func (c *cancelController) Cancel(ctx context.Context, req entity.CancelRequest)
 	)
 
 	// Verify the sqid exists before recording intent or publishing.
-	if _, err := c.requestSummaryStore.Get(ctx, req.ID); err != nil {
+	summary, err := c.requestSummaryStore.Get(ctx, req.ID)
+	if err != nil {
 		if storage.IsNotFound(err) {
 			metrics.NamedCounter(c.metricsScope, opName, "not_found", 1)
 			return errs.NewUserError(&RequestNotFoundError{Sqid: req.ID})
@@ -94,6 +95,10 @@ func (c *cancelController) Cancel(ctx context.Context, req entity.CancelRequest)
 		return fmt.Errorf("failed to look up request summary for sqid=%s: %w", req.ID, err)
 	}
 
+	// Stamp the authoritative queue from the stored summary onto the payload,
+	// overriding whatever the caller supplied.
+	req.Queue = summary.Queue
+
 	// Record the user's intent in the request log before publishing. Writing direct to the
 	// store (rather than via the log topic) keeps the gateway-emitted entry consistent with
 	// the Land "accepted" entry and guarantees the entry is visible the moment Cancel returns.
diff --git a/submitqueue/orchestrator/controller/batch/batch.go b/submitqueue/orchestrator/controller/batch/batch.go
index 74a1d6ec..f038a29c 100644
--- a/submitqueue/orchestrator/controller/batch/batch.go
+++ b/submitqueue/orchestrator/controller/batch/batch.go
@@ -95,6 +95,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get request %s: %w", rid.ID, err)
 	}
 
+	// The payload's queue must match the request's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if rid.Queue != "" && rid.Queue != request.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of request %s", rid.Queue, request.Queue, request.ID)
+	}
+
 	c.logger.Infow("received batch event",
 		"request_id", request.ID,
 		"queue", request.Queue,
@@ -355,15 +362,16 @@ func (c *Controller) populateBatch(ctx context.Context, batch entity.Batch) (ent
 	return batch, nil
 }
 
-// publish publishes a batch ID to the specified topic key.
-func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
-	bid := entity.BatchID{ID: batchID}
+// publish publishes a batch ID to the specified topic key, stamped with and
+// partitioned by the batch's queue.
+func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error {
+	bid := entity.BatchID{ID: batchID, Queue: queue}
 	payload, err := bid.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {
diff --git a/submitqueue/orchestrator/controller/batch/batch_test.go b/submitqueue/orchestrator/controller/batch/batch_test.go
index 524d2d6f..357221ba 100644
--- a/submitqueue/orchestrator/controller/batch/batch_test.go
+++ b/submitqueue/orchestrator/controller/batch/batch_test.go
@@ -195,6 +195,100 @@ func TestController_Process_Success(t *testing.T) {
 	require.NoError(t, err)
 }
 
+// TestController_Process_QueueMismatchRejected asserts a payload whose queue
+// disagrees with the request's authoritative queue is rejected without
+// touching the counter, the batch store, or the publisher.
+func TestController_Process_QueueMismatchRejected(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	request := testRequest()
+
+	mockReqStore := storagemock.NewMockRequestStore(ctrl)
+	mockReqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil)
+
+	mockStorage := storagemock.NewMockStorage(ctrl)
+	mockStorage.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
+
+	// Counter with no EXPECTs — must not be called.
+	cnt := countermock.NewMockCounter(ctrl)
+	controller := newTestController(t, ctrl, cnt, mockStorage, nil, fmt.Errorf("should not publish"))
+
+	payload, err := entity.RequestID{ID: request.ID, Queue: "some-other-queue"}.ToBytes()
+	require.NoError(t, err)
+	msg := entityqueue.NewMessage(request.ID, payload, request.Queue, nil)
+	delivery := consumermock.NewMockDelivery(ctrl)
+	delivery.EXPECT().Message().Return(msg).AnyTimes()
+	delivery.EXPECT().Attempt().Return(1).AnyTimes()
+
+	require.Error(t, controller.Process(context.Background(), delivery))
+}
+
+// TestController_Process_StampsQueueOnSpeculatePayload asserts the batch ID
+// published to speculate carries the batch's queue.
+func TestController_Process_StampsQueueOnSpeculatePayload(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	request := testRequest()
+
+	var speculateMsgs []entityqueue.Message
+	mockPub := queuemock.NewMockPublisher(ctrl)
+	mockPub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, topic string, msg entityqueue.Message) error {
+			if topic == "speculate" {
+				speculateMsgs = append(speculateMsgs, msg)
+			}
+			return nil
+		},
+	).AnyTimes()
+	mockQ := queuemock.NewMockQueue(ctrl)
+	mockQ.EXPECT().Publisher().Return(mockPub).AnyTimes()
+
+	registry, err := consumer.NewTopicRegistry(
+		[]consumer.TopicConfig{
+			{Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: mockQ},
+			{Key: topickey.TopicKeyLog, Name: "log", Queue: mockQ},
+		},
+	)
+	require.NoError(t, err)
+
+	mockBatchStore := storagemock.NewMockBatchStore(ctrl)
+	mockBatchStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockBatchStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockReqStore := storagemock.NewMockRequestStore(ctrl)
+	mockReqStore.EXPECT().Get(gomock.Any(), request.ID).Return(request, nil).AnyTimes()
+	mockReqStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockBatchDependentStore := storagemock.NewMockBatchDependentStore(ctrl)
+	mockBatchDependentStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	mockRequestBatchStore := storagemock.NewMockRequestBatchStore(ctrl)
+	mockRequestBatchStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+
+	mockStorage := storagemock.NewMockStorage(ctrl)
+	mockStorage.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
+	mockStorage.EXPECT().GetBatchStore().Return(mockBatchStore).AnyTimes()
+	mockStorage.EXPECT().GetBatchDependentStore().Return(mockBatchDependentStore).AnyTimes()
+	mockStorage.EXPECT().GetRequestBatchStore().Return(mockRequestBatchStore).AnyTimes()
+	mockStorage.EXPECT().GetRequestStore().Return(mockReqStore).AnyTimes()
+
+	analyzerFactory := conflictmock.NewMockFactory(ctrl)
+	analyzerFactory.EXPECT().For(gomock.Any()).Return(all.New(), nil).AnyTimes()
+	controller := NewController(
+		zaptest.NewLogger(t).Sugar(), tally.NoopScope, registry, newSequentialCounter(ctrl),
+		mockStorage, analyzerFactory, topickey.TopicKeyBatch, "orchestrator-batch",
+	)
+
+	msg := entityqueue.NewMessage(request.ID, requestIDPayload(t, request.ID), request.Queue, nil)
+	delivery := consumermock.NewMockDelivery(ctrl)
+	delivery.EXPECT().Message().Return(msg).AnyTimes()
+	delivery.EXPECT().Attempt().Return(1).AnyTimes()
+
+	require.NoError(t, controller.Process(context.Background(), delivery))
+	require.Len(t, speculateMsgs, 1)
+	bid, err := entity.BatchIDFromBytes(speculateMsgs[0].Payload)
+	require.NoError(t, err)
+	assert.Equal(t, request.Queue, bid.Queue)
+	assert.Equal(t, request.Queue, speculateMsgs[0].PartitionKey)
+}
+
 // TestController_Process_PublishesBatchedLog asserts the controller emits a
 // "batched" request log carrying the request ID, the post-CAS request version,
 // and the batch ID it was placed into.
diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go
index 0c4c0d28..0a1a1e36 100644
--- a/submitqueue/orchestrator/controller/build/build.go
+++ b/submitqueue/orchestrator/controller/build/build.go
@@ -89,6 +89,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get batch %s: %w", bid.ID, err)
 	}
 
+	// The payload's queue must match the batch's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if bid.Queue != "" && bid.Queue != batch.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID)
+	}
+
 	c.logger.Infow("received build event",
 		"batch_id", batch.ID,
 		"queue", batch.Queue,
@@ -151,7 +158,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	// Hand off to the buildsignal poll loop; it calls Status, updates the
 	// persisted Build, publishes to speculate, and holds its delivery
 	// between polls until terminal.
-	if err := c.publish(ctx, topickey.TopicKeyBuildSignal, build); err != nil {
+	if err := c.publish(ctx, topickey.TopicKeyBuildSignal, build, batch.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to buildsignal: %w", err)
 	}
@@ -184,11 +191,12 @@ func (c *Controller) loadBatches(ctx context.Context, batchIDs []string) ([]enti
 	return batches, nil
 }
 
-// publish publishes a build's ID to the specified topic key. Only the
-// identifier travels on the queue; the consumer loads the full Build from
-// storage, keeping the message small and the store the single source of truth.
-func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, build entity.Build) error {
-	payload, err := entity.BuildID{ID: build.ID}.ToBytes()
+// publish publishes a build's ID to the specified topic key, stamped with the
+// batch's queue and partitioned by the batch ID. Only the identifier and its
+// queue travel on the queue; the consumer loads the full Build from storage,
+// keeping the message small and the store the single source of truth.
+func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, build entity.Build, queue string) error {
+	payload, err := entity.BuildID{ID: build.ID, Queue: queue}.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize build ID: %w", err)
 	}
diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
index 9c7ecbcd..fd3ec821 100644
--- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
+++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go
@@ -135,6 +135,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get batch %s: %w", build.BatchID, err)
 	}
 
+	// The payload's queue must match the batch's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if buildID.Queue != "" && buildID.Queue != batch.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of batch %s", buildID.Queue, batch.Queue, batch.ID)
+	}
+
 	buildRunner, err := c.buildRunners.For(buildrunner.Config{QueueName: batch.Queue})
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "status_errors", 1)
@@ -169,8 +176,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to update status for build %s: %w", build.ID, err)
 	}
 
-	// Re-evaluate the batch state machine with the latest build status.
-	if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, updatedBuild.BatchID, msg.PartitionKey); err != nil {
+	// Re-evaluate the batch state machine with the latest build status. The
+	// speculate topic is partitioned by queue like every other speculate
+	// publisher, so a queue's batches keep their serial processing guarantee.
+	if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, updatedBuild.BatchID, batch.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to speculate: %w", err)
 	}
@@ -210,15 +219,16 @@ func pollDelay(status entity.BuildStatus) int64 {
 	}
 }
 
-// publishBatchID publishes a batch ID to the topic identified by key.
-func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
-	bid := entity.BatchID{ID: batchID}
+// publishBatchID publishes a batch ID to the topic identified by key, stamped
+// with and partitioned by the batch's queue.
+func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error {
+	bid := entity.BatchID{ID: batchID, Queue: queue}
 	payload, err := bid.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {
diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go
index 69366193..9629e212 100644
--- a/submitqueue/orchestrator/controller/cancel/cancel.go
+++ b/submitqueue/orchestrator/controller/cancel/cancel.go
@@ -119,6 +119,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get request %s: %w", cancelReq.ID, err)
 	}
 
+	// The payload's queue must match the request's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if cancelReq.Queue != "" && cancelReq.Queue != request.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of request %s", cancelReq.Queue, request.Queue, request.ID)
+	}
+
 	c.logger.Infow("received cancel event",
 		"request_id", request.ID,
 		"queue", request.Queue,
@@ -343,15 +350,16 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error
 	return nil
 }
 
-// publishBatchID publishes a BatchID-payload message to the specified topic key.
-func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
-	bid := entity.BatchID{ID: batchID}
+// publishBatchID publishes a BatchID-payload message to the specified topic
+// key, stamped with and partitioned by the batch's queue.
+func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error {
+	bid := entity.BatchID{ID: batchID, Queue: queue}
 	payload, err := bid.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {
diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go
index 5a1810a4..453dce46 100644
--- a/submitqueue/orchestrator/controller/conclude/conclude.go
+++ b/submitqueue/orchestrator/controller/conclude/conclude.go
@@ -81,6 +81,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get batch %s: %w", bid.ID, err)
 	}
 
+	// The payload's queue must match the batch's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if bid.Queue != "" && bid.Queue != batch.Queue {
+		metrics.NamedCounter(c.metricsScope, "process", "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID)
+	}
+
 	c.logger.Infow("received conclude event",
 		"batch_id", batch.ID,
 		"queue", batch.Queue,
diff --git a/submitqueue/orchestrator/controller/merge/merge.go b/submitqueue/orchestrator/controller/merge/merge.go
index d9c87930..484ba9c5 100644
--- a/submitqueue/orchestrator/controller/merge/merge.go
+++ b/submitqueue/orchestrator/controller/merge/merge.go
@@ -107,6 +107,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get batch %s: %w", bid.ID, err)
 	}
 
+	// The payload's queue must match the batch's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if bid.Queue != "" && bid.Queue != batch.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID)
+	}
+
 	c.logger.Infow("received merge event",
 		"batch_id", batch.ID,
 		"queue", batch.Queue,
diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
index 82561bda..761957b1 100644
--- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
+++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go
@@ -188,14 +188,15 @@ func (c *Controller) failRequest(ctx context.Context, request entity.Request, re
 	return nil
 }
 
-// publishRequestID publishes a request ID to the given topic key, partitioned by queue.
-func (c *Controller) publishRequestID(ctx context.Context, key consumer.TopicKey, requestID string, partitionKey string) error {
-	payload, err := entity.RequestID{ID: requestID}.ToBytes()
+// publishRequestID publishes a request ID to the given topic key, stamped with
+// and partitioned by the request's queue.
+func (c *Controller) publishRequestID(ctx context.Context, key consumer.TopicKey, requestID string, queue string) error {
+	payload, err := entity.RequestID{ID: requestID, Queue: queue}.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize request ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(requestID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(requestID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {
diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
index 51c5c015..6617ba0a 100644
--- a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
+++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go
@@ -155,26 +155,27 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 
 // fanout publishes the batch ID to conclude (so requests are updated) and to
 // speculate (so dependents can re-evaluate now that this batch is done).
-func (c *Controller) fanout(ctx context.Context, batchID, partitionKey string) error {
-	if err := c.publish(ctx, topickey.TopicKeyConclude, batchID, partitionKey); err != nil {
+func (c *Controller) fanout(ctx context.Context, batchID, queue string) error {
+	if err := c.publish(ctx, topickey.TopicKeyConclude, batchID, queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, "process", "publish_conclude_errors", 1)
 		return fmt.Errorf("failed to publish to conclude: %w", err)
 	}
-	if err := c.publish(ctx, topickey.TopicKeySpeculate, batchID, partitionKey); err != nil {
+	if err := c.publish(ctx, topickey.TopicKeySpeculate, batchID, queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, "process", "publish_speculate_errors", 1)
 		return fmt.Errorf("failed to publish to speculate: %w", err)
 	}
 	return nil
 }
 
-// publish publishes a batch ID to the given topic key, partitioned by queue.
-func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
-	payload, err := entity.BatchID{ID: batchID}.ToBytes()
+// publish publishes a batch ID to the given topic key, stamped with and
+// partitioned by the batch's queue.
+func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error {
+	payload, err := entity.BatchID{ID: batchID, Queue: queue}.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {
diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go
index 43ce01b9..3ed8b186 100644
--- a/submitqueue/orchestrator/controller/speculate/speculate.go
+++ b/submitqueue/orchestrator/controller/speculate/speculate.go
@@ -105,6 +105,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get batch %s: %w", bid.ID, err)
 	}
 
+	// The payload's queue must match the batch's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if bid.Queue != "" && bid.Queue != batch.Queue {
+		metrics.NamedCounter(c.metricsScope, opName, "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID)
+	}
+
 	// Cancelling intent: the cancel controller has handed this batch off to
 	// speculate to drive to terminal. Cancel in-flight builds, fan out to
 	// dependents, CAS to terminal Cancelled, and publish to conclude.
@@ -421,23 +428,24 @@ func (c *Controller) fetchDependencies(ctx context.Context, batch entity.Batch)
 // fanout re-publishes downstream events for a batch that has already reached
 // a terminal state. Used for self-healing when a previous publish was lost:
 // re-sending to conclude guarantees request-state reconciliation.
-func (c *Controller) fanout(ctx context.Context, batchID, partitionKey string) error {
-	if err := c.publish(ctx, topickey.TopicKeyConclude, batchID, partitionKey); err != nil {
+func (c *Controller) fanout(ctx context.Context, batchID, queue string) error {
+	if err := c.publish(ctx, topickey.TopicKeyConclude, batchID, queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to conclude: %w", err)
 	}
 	return nil
 }
 
-// publish publishes a batch ID to the specified topic key.
-func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, partitionKey string) error {
-	bid := entity.BatchID{ID: batchID}
+// publish publishes a batch ID to the specified topic key, stamped with and
+// partitioned by the batch's queue.
+func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, batchID string, queue string) error {
+	bid := entity.BatchID{ID: batchID, Queue: queue}
 	payload, err := bid.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize batch ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(batchID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(batchID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {
diff --git a/submitqueue/orchestrator/controller/start/start.go b/submitqueue/orchestrator/controller/start/start.go
index a37e4244..b21d01e2 100644
--- a/submitqueue/orchestrator/controller/start/start.go
+++ b/submitqueue/orchestrator/controller/start/start.go
@@ -130,15 +130,16 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	return nil
 }
 
-// publish publishes a request ID to the specified topic key.
-func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, requestID string, partitionKey string) error {
-	rid := entity.RequestID{ID: requestID}
+// publish publishes a request ID to the specified topic key, stamped with and
+// partitioned by the request's queue.
+func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, requestID string, queue string) error {
+	rid := entity.RequestID{ID: requestID, Queue: queue}
 	payload, err := rid.ToBytes()
 	if err != nil {
 		return fmt.Errorf("failed to serialize request ID: %w", err)
 	}
 
-	msg := entityqueue.NewMessage(requestID, payload, partitionKey, nil)
+	msg := entityqueue.NewMessage(requestID, payload, queue, nil)
 
 	q, ok := c.registry.Queue(key)
 	if !ok {
diff --git a/submitqueue/orchestrator/controller/validate/validate.go b/submitqueue/orchestrator/controller/validate/validate.go
index eb2e0655..6d5496b0 100644
--- a/submitqueue/orchestrator/controller/validate/validate.go
+++ b/submitqueue/orchestrator/controller/validate/validate.go
@@ -105,6 +105,13 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		return fmt.Errorf("failed to get request %s: %w", rid.ID, err)
 	}
 
+	// The payload's queue must match the request's authoritative queue; a
+	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
+	if rid.Queue != "" && rid.Queue != request.Queue {
+		coremetrics.NamedCounter(c.metricsScope, "process", "queue_mismatch", 1)
+		return fmt.Errorf("payload queue %q does not match queue %q of request %s", rid.Queue, request.Queue, request.ID)
+	}
+
 	c.logger.Infow("received validate event",
 		"request_id", request.ID,
 		"queue", request.Queue,