diff --git a/submitqueue/entity/BUILD.bazel b/submitqueue/entity/BUILD.bazel
index b7578c1dd..a2dc0a684 100644
--- a/submitqueue/entity/BUILD.bazel
+++ b/submitqueue/entity/BUILD.bazel
@@ -24,11 +24,13 @@ go_library(
         "request_log.go",
         "request_summary.go",
         "speculation.go",
+        "subject.go",
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/entity",
     visibility = ["//visibility:public"],
     deps = [
         "//platform/base/change:go_default_library",
+        "//platform/base/failure:go_default_library",
         "//platform/base/mergestrategy:go_default_library",
     ],
 )
diff --git a/submitqueue/entity/subject.go b/submitqueue/entity/subject.go
new file mode 100644
index 000000000..7aea6f602
--- /dev/null
+++ b/submitqueue/entity/subject.go
@@ -0,0 +1,46 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package entity
+
+import "github.com/uber/submitqueue/platform/base/failure"
+
+// Subject types name what a failure is about. The queue layer treats the type
+// as opaque; these are SubmitQueue's own vocabulary for it.
+const (
+	// SubjectTypeBatch identifies a subject by batch ID.
+	SubjectTypeBatch = "batch"
+	// SubjectTypeQueue identifies a subject by queue name. Used where no single
+	// batch is at fault — a failure reading or planning the queue as a whole.
+	SubjectTypeQueue = "queue"
+	// SubjectTypeRequest identifies a subject by request ID.
+	SubjectTypeRequest = "request"
+)
+
+// BatchSubject names a batch as what a failure is about.
+func BatchSubject(batchID string) failure.Subject {
+	return failure.Subject{Type: SubjectTypeBatch, ID: batchID}
+}
+
+// QueueSubject names a queue as what a failure is about. It is the honest
+// subject for work that spans the queue — listing it, planning it — where
+// blaming any one batch would be a guess.
+func QueueSubject(queue string) failure.Subject {
+	return failure.Subject{Type: SubjectTypeQueue, ID: queue}
+}
+
+// RequestSubject names a request as what a failure is about.
+func RequestSubject(requestID string) failure.Subject {
+	return failure.Subject{Type: SubjectTypeRequest, ID: requestID}
+}
diff --git a/submitqueue/orchestrator/controller/dlq/BUILD.bazel b/submitqueue/orchestrator/controller/dlq/BUILD.bazel
index 85768f22a..c8840e5ff 100644
--- a/submitqueue/orchestrator/controller/dlq/BUILD.bazel
+++ b/submitqueue/orchestrator/controller/dlq/BUILD.bazel
@@ -10,6 +10,7 @@ go_library(
         "mergeconflictsignal.go",
         "mergesignal.go",
         "request.go",
+        "speculate.go",
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/dlq",
     visibility = ["//visibility:public"],
@@ -18,7 +19,9 @@ go_library(
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
         "//submitqueue/core/batch:go_default_library",
+        "//submitqueue/core/publish:go_default_library",
         "//submitqueue/core/request:go_default_library",
+        "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
@@ -37,11 +40,13 @@ go_test(
         "mergesignal_test.go",
         "publisher_test.go",
         "request_test.go",
+        "speculate_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
         "//api/runway/messagequeue:go_default_library",
         "//api/runway/messagequeue/protopb:go_default_library",
+        "//platform/base/failure:go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go
index 7fe0b5db1..5111926bf 100644
--- a/submitqueue/orchestrator/controller/dlq/batch.go
+++ b/submitqueue/orchestrator/controller/dlq/batch.go
@@ -27,15 +27,21 @@ import (
 )
 
 // batchController is the DLQ reconciler for batch-scoped pipeline stages
-// (speculate, build, merge, conclude). All four topics carry a
-// BatchID payload, so this controller is registered four times — one per
-// topic, each with the matching DLQ topic key and consumer group.
+// (build, merge, conclude). All three topics carry a BatchID payload, so this
+// controller is registered three times — one per topic, each with the matching
+// DLQ topic key and consumer group.
 //
 // On each delivery the controller decodes the BatchID, transitions the batch
 // to BatchStateFailed (idempotent if already halted), and fans out by
 // transitioning each member request to RequestStateError. The fan-out exists
 // because conclude — which normally drives request state from batch state —
 // will not run for a DLQ'd batch.
+//
+// Blaming the batch on the message is right for these stages because their
+// work is that batch: whatever failed, it failed doing this batch's build,
+// merge, or conclusion. The speculate stage is not like that — it re-plans a
+// whole queue from a message that names one batch — so it has its own
+// reconciler; see speculate.go.
 type batchController struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
@@ -92,6 +98,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err)
 	}
 
+	lastError, failureMeta := failureContext(delivery)
 	dmeta := delivery.Metadata()
 	c.logger.Warnw("dlq message received",
 		"batch_id", bid.ID,
@@ -101,7 +108,7 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver
 		"dlq_last_error", dmeta["dlq.last_error"],
 	)
 
-	if err := failBatch(ctx, store, c.registry, c.logger, bid.ID, dmeta["dlq.last_error"]); err != nil {
+	if _, err := failBatch(ctx, store, c.registry, c.logger, bid.ID, lastError, failureMeta); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
 		return err
 	}
diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go
index 07dc69c55..f6b61c44c 100644
--- a/submitqueue/orchestrator/controller/dlq/buildsignal.go
+++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go
@@ -90,6 +90,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", buildID.Queue, err)
 	}
 
+	lastError, failureMeta := failureContext(delivery)
 	dmeta := delivery.Metadata()
 	c.logger.Warnw("dlq message received",
 		"build_id", buildID.ID,
@@ -121,7 +122,7 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D
 		return nil
 	}
 
-	if err := failBatch(ctx, store, c.registry, c.logger, build.BatchID, dmeta["dlq.last_error"]); err != nil {
+	if _, err := failBatch(ctx, store, c.registry, c.logger, build.BatchID, lastError, failureMeta); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
 		return err
 	}
diff --git a/submitqueue/orchestrator/controller/dlq/dlq.go b/submitqueue/orchestrator/controller/dlq/dlq.go
index 689360de7..ee9c6b387 100644
--- a/submitqueue/orchestrator/controller/dlq/dlq.go
+++ b/submitqueue/orchestrator/controller/dlq/dlq.go
@@ -37,6 +37,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/uber/submitqueue/platform/consumer"
 	corebatch "github.com/uber/submitqueue/submitqueue/core/batch"
@@ -62,19 +63,81 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey {
 	return consumer.TopicKey(string(main) + topicSuffix)
 }
 
+// failureContext reads everything the queue recorded about a dead-lettered
+// message: the human-readable reason, and a metadata map to carry alongside it
+// on the terminal request log.
+//
+// The map is what makes a dead letter diagnosable after the fact. The queue
+// already hands the reconciler the failure count, the topic it failed on, and
+// when — and, when the producer attributed it, which entities it was about.
+// All of that used to be logged and then dropped; the request log is where a
+// user can actually see it, through the gateway's status and history.
+//
+// Values are flattened to strings because RequestLog.Metadata and the gateway's
+// wire contract are both string maps. Nested detail becomes dotted keys, which
+// read well in a display surface where a JSON blob would not.
+func failureContext(delivery consumer.Delivery) (string, map[string]string) {
+	dmeta := delivery.Metadata()
+	metadata := make(map[string]string, len(dmeta))
+	for _, key := range []string{"dlq.original_topic", "dlq.failure_count", "dlq.failed_at"} {
+		if v, ok := dmeta[key]; ok && v != "" {
+			metadata[key] = v
+		}
+	}
+
+	lastError := dmeta["dlq.last_error"]
+
+	f, failed := delivery.Failure()
+	if !failed {
+		return lastError, metadata
+	}
+	if f.Message != "" {
+		lastError = f.Message
+	}
+
+	// One key per subject type, so several batches at fault read as a list
+	// rather than overwriting each other.
+	byType := make(map[string][]string, len(f.Subjects))
+	for _, s := range f.Subjects {
+		byType[s.Type] = append(byType[s.Type], s.ID)
+	}
+	for subjectType, ids := range byType {
+		metadata["dlq.subject."+subjectType] = strings.Join(ids, ",")
+	}
+
+	flattenDetail("dlq.detail", f.Detail, metadata)
+
+	return lastError, metadata
+}
+
+// flattenDetail writes a JSON-shaped document into a string map, joining nested
+// keys with dots. Values are rendered with %v: this is a display surface, not a
+// contract, so a readable rendering beats a faithful one.
+func flattenDetail(prefix string, detail map[string]any, out map[string]string) {
+	for k, v := range detail {
+		key := prefix + "." + k
+		if nested, ok := v.(map[string]any); ok {
+			flattenDetail(key, nested, out)
+			continue
+		}
+		out[key] = fmt.Sprintf("%v", v)
+	}
+}
+
 // failRequest transitions a non-terminal request to RequestStateError and
 // appends the matching RequestStatusError log. Redelivery for an existing Error
 // state repeats materialization to repair a previous partial attempt. A
 // different terminal outcome is left unchanged.
 // lastError is the failure reason preserved by the queue in DLQ delivery
-// metadata and is exposed through Status and History for diagnosis.
+// metadata and is exposed through Status and History for diagnosis, alongside
+// metadata carrying the rest of the failure context.
 //
 // A request in RequestStateCancelling is reconciled to RequestStateError, not
 // left in place: DLQ means the pipeline failed to converge, so we cannot
 // confirm the cancel completed cleanly. Writing Error is the honest signal and
 // keeps the request from being stuck in a non-terminal state forever.
-func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string) error {
-	res, err := requestcore.TerminateRequest(ctx, store, registry, requestID, entity.RequestStateError, lastError, nil)
+func failRequest(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, requestID, lastError string, metadata map[string]string) error {
+	res, err := requestcore.TerminateRequest(ctx, store, registry, requestID, entity.RequestStateError, lastError, metadata)
 	if err != nil {
 		return fmt.Errorf("dlq reconcile request %s failed: %w", requestID, err)
 	}
@@ -112,19 +175,27 @@ func failRequest(ctx context.Context, store storage.Storage, registry consumer.T
 // Idempotency: an existing Failed batch repeats fan-out because a previous
 // attempt may have crashed after updating the batch. Succeeded and Cancelled
 // are different terminal outcomes and do not fan out errors.
-// lastError is propagated to each member request's terminal Error log.
-func failBatch(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string) error {
+// lastError and metadata are propagated to each member request's terminal
+// Error log.
+//
+// It reports whether it transitioned the batch. A caller that only wants to act
+// on real progress — republishing to wake the queue, say — can then tell a
+// first reconcile from a redelivery of one already done, and avoid doing it
+// again forever.
+func failBatch(ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, logger *zap.SugaredLogger, batchID, lastError string, metadata map[string]string) (bool, error) {
 	batch, err := store.GetBatchStore().Get(ctx, batchID)
 	if err != nil {
 		if errors.Is(err, storage.ErrNotFound) {
 			logger.Warnw("dlq reconcile: batch not found, skipping",
 				"batch_id", batchID,
 			)
-			return nil
+			return false, nil
 		}
-		return fmt.Errorf("failed to get batch %s: %w", batchID, err)
+		return false, fmt.Errorf("failed to get batch %s: %w", batchID, err)
 	}
 
+	transitioned := false
+
 	switch batch.State {
 	case entity.BatchStateFailed:
 		logger.Infow("dlq reconcile: batch already failed, repairing request fan-out",
@@ -133,21 +204,22 @@ func failBatch(ctx context.Context, store storage.Storage, registry consumer.Top
 		// A prior attempt may have CAS'd to Failed without completing the
 		// membership record move; repair it alongside the fan-out.
 		if err := corebatch.EnsureRecord(ctx, store, batch); err != nil {
-			return err
+			return false, err
 		}
 	case entity.BatchStateSucceeded, entity.BatchStateCancelled:
 		logger.Infow("dlq reconcile: batch has a different terminal outcome, skipping",
 			"batch_id", batchID,
 			"state", string(batch.State),
 		)
-		return nil
+		return false, nil
 	default:
 		previousState := batch.State
 		updated, err := corebatch.Transition(ctx, store, batch, entity.BatchStateFailed)
 		if err != nil {
-			return err
+			return false, err
 		}
 		batch = updated
+		transitioned = true
 		logger.Infow("dlq reconcile: batch marked failed",
 			"batch_id", batchID,
 			"previous_state", string(previousState),
@@ -155,9 +227,9 @@ func failBatch(ctx context.Context, store storage.Storage, registry consumer.Top
 	}
 
 	for _, requestID := range batch.Contains {
-		if err := failRequest(ctx, store, registry, logger, requestID, lastError); err != nil {
-			return fmt.Errorf("fan-out for batch %s: %w", batchID, err)
+		if err := failRequest(ctx, store, registry, logger, requestID, lastError, metadata); err != nil {
+			return transitioned, fmt.Errorf("fan-out for batch %s: %w", batchID, err)
 		}
 	}
-	return nil
+	return transitioned, nil
 }
diff --git a/submitqueue/orchestrator/controller/dlq/dlq_test.go b/submitqueue/orchestrator/controller/dlq/dlq_test.go
index e673d7cc8..c77a804af 100644
--- a/submitqueue/orchestrator/controller/dlq/dlq_test.go
+++ b/submitqueue/orchestrator/controller/dlq/dlq_test.go
@@ -90,7 +90,7 @@ func TestFailRequest_TerminalStates(t *testing.T) {
 				})
 			}
 
-			err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "")
+			err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "", nil)
 			require.NoError(t, err)
 		})
 	}
@@ -122,7 +122,7 @@ func TestFailRequest_CancellingTransitionsToError(t *testing.T) {
 	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
 
-	err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "")
+	err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "", nil)
 	require.NoError(t, err)
 }
 
@@ -147,7 +147,7 @@ func TestFailRequest_TransitionsToError(t *testing.T) {
 	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
 
-	err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "")
+	err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "", nil)
 	require.NoError(t, err)
 }
 
@@ -172,7 +172,7 @@ func TestFailRequest_LogPublishErrorPropagates(t *testing.T) {
 	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
 
-	err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "")
+	err := failRequest(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/1", "", nil)
 	require.Error(t, err)
 }
 
@@ -186,7 +186,7 @@ func TestFailRequest_NotFoundIsNoOp(t *testing.T) {
 	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
 
-	err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "")
+	err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "", nil)
 	require.NoError(t, err)
 }
 
@@ -200,7 +200,7 @@ func TestFailRequest_GenericGetErrorIsNonRetryable(t *testing.T) {
 	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
 
-	err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "")
+	err := failRequest(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/1", "", nil)
 	require.Error(t, err)
 	assert.False(t, errs.IsRetryable(err))
 }
@@ -239,7 +239,7 @@ func TestFailBatch_TransitionsAndFansOut(t *testing.T) {
 	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
 
-	err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "")
+	_, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil)
 	require.NoError(t, err)
 }
 
@@ -269,7 +269,7 @@ func TestFailBatch_FailedFansOutForRepair(t *testing.T) {
 	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
 
-	err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "")
+	_, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil)
 	require.NoError(t, err)
 }
 
@@ -286,7 +286,7 @@ func TestFailBatch_DifferentTerminalOutcomeSkipsFanOut(t *testing.T) {
 			store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 			store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
 
-			err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "")
+			_, err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil)
 			require.NoError(t, err)
 		})
 	}
@@ -325,7 +325,7 @@ func TestFailBatch_CancellingTransitionsToFailed(t *testing.T) {
 	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
 	store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
 
-	err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "")
+	_, err := failBatch(context.Background(), store, registry, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil)
 	require.NoError(t, err)
 }
 
@@ -339,7 +339,7 @@ func TestFailBatch_NotFoundIsNoOp(t *testing.T) {
 	store.EXPECT().GetQueueBatchStateStore().Return(newQueueBatchStateStore(ctrl)).AnyTimes()
 	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
 
-	err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "")
+	_, err := failBatch(context.Background(), store, consumer.TopicRegistry{}, zaptest.NewLogger(t).Sugar(), "q/batch/1", "", nil)
 	require.NoError(t, err)
 }
 
diff --git a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
index e57f14311..c51561288 100644
--- a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
+++ b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go
@@ -82,6 +82,7 @@ func (c *mergeConflictSignalController) Process(ctx context.Context, delivery co
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", result.GetQueueName(), err)
 	}
 
+	lastError, failureMeta := failureContext(delivery)
 	dmeta := delivery.Metadata()
 	c.logger.Warnw("dlq message received",
 		"request_id", result.Id,
@@ -91,7 +92,7 @@ func (c *mergeConflictSignalController) Process(ctx context.Context, delivery co
 		"dlq_last_error", dmeta["dlq.last_error"],
 	)
 
-	if err := failRequest(ctx, store, c.registry, c.logger, result.Id, dmeta["dlq.last_error"]); err != nil {
+	if err := failRequest(ctx, store, c.registry, c.logger, result.Id, lastError, failureMeta); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
 		return err
 	}
diff --git a/submitqueue/orchestrator/controller/dlq/mergesignal.go b/submitqueue/orchestrator/controller/dlq/mergesignal.go
index 4243df8bb..8eb3d9e4f 100644
--- a/submitqueue/orchestrator/controller/dlq/mergesignal.go
+++ b/submitqueue/orchestrator/controller/dlq/mergesignal.go
@@ -81,6 +81,7 @@ func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.D
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", result.GetQueueName(), err)
 	}
 
+	lastError, failureMeta := failureContext(delivery)
 	dmeta := delivery.Metadata()
 	c.logger.Warnw("dlq message received",
 		"batch_id", result.Id,
@@ -90,7 +91,7 @@ func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.D
 		"dlq_last_error", dmeta["dlq.last_error"],
 	)
 
-	if err := failBatch(ctx, store, c.registry, c.logger, result.Id, dmeta["dlq.last_error"]); err != nil {
+	if _, err := failBatch(ctx, store, c.registry, c.logger, result.Id, lastError, failureMeta); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
 		return err
 	}
diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go
index f5fafec9c..46b68743b 100644
--- a/submitqueue/orchestrator/controller/dlq/request.go
+++ b/submitqueue/orchestrator/controller/dlq/request.go
@@ -129,6 +129,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", rid.Queue, err)
 	}
 
+	lastError, failureMeta := failureContext(delivery)
 	dmeta := delivery.Metadata()
 	c.logger.Warnw("dlq message received",
 		"request_id", rid.ID,
@@ -138,7 +139,7 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv
 		"dlq_last_error", dmeta["dlq.last_error"],
 	)
 
-	if err := failRequest(ctx, store, c.registry, c.logger, rid.ID, dmeta["dlq.last_error"]); err != nil {
+	if err := failRequest(ctx, store, c.registry, c.logger, rid.ID, lastError, failureMeta); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
 		return err
 	}
diff --git a/submitqueue/orchestrator/controller/dlq/request_test.go b/submitqueue/orchestrator/controller/dlq/request_test.go
index 4b317d972..5f88c6bcc 100644
--- a/submitqueue/orchestrator/controller/dlq/request_test.go
+++ b/submitqueue/orchestrator/controller/dlq/request_test.go
@@ -20,6 +20,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/uber/submitqueue/platform/base/failure"
 	queue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
@@ -179,6 +180,12 @@ func TestDLQRequestController_Process_EmptyIDFails(t *testing.T) {
 // newMockDelivery returns a MockDelivery wired up enough to be passed through
 // the DLQ controller Process flow.
 func newMockDelivery(ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery {
+	return newMockDeliveryWithFailure(ctrl, payload, failure.Failure{}, false)
+}
+
+// newMockDeliveryWithFailure builds a delivery that also reports a recorded
+// failure, as a redelivery from a DLQ topic does.
+func newMockDeliveryWithFailure(ctrl *gomock.Controller, payload []byte, f failure.Failure, failed bool) *consumermock.MockDelivery {
 	msg := queue.NewMessage("dlq-msg-1", payload, "", nil)
 	d := consumermock.NewMockDelivery(ctrl)
 	d.EXPECT().Message().Return(msg).AnyTimes()
@@ -188,5 +195,6 @@ func newMockDelivery(ctrl *gomock.Controller, payload []byte) *consumermock.Mock
 		"dlq.failure_count":  "3",
 		"dlq.last_error":     "boom",
 	}).AnyTimes()
+	d.EXPECT().Failure().Return(f, failed).AnyTimes()
 	return d
 }
diff --git a/submitqueue/orchestrator/controller/dlq/speculate.go b/submitqueue/orchestrator/controller/dlq/speculate.go
new file mode 100644
index 000000000..3c41c6b5f
--- /dev/null
+++ b/submitqueue/orchestrator/controller/dlq/speculate.go
@@ -0,0 +1,233 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dlq
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/metrics"
+	corebatch "github.com/uber/submitqueue/submitqueue/core/batch"
+	"github.com/uber/submitqueue/submitqueue/core/publish"
+	"github.com/uber/submitqueue/submitqueue/core/topickey"
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/storage"
+	"go.uber.org/zap"
+)
+
+// speculateController is the DLQ reconciler for the speculate topic.
+//
+// Speculate does not share the generic batch reconciler because it is not a
+// batch-scoped stage. Its message names one batch, but a run re-plans the
+// entire queue: it lists every in-flight batch, hands them all to the
+// Speculator, and commits outcomes for any of them. A failure is therefore
+// often about a different batch than the message names, or about the queue
+// itself — and failing the named batch regardless would terminate one that was
+// never at fault while leaving the real culprit running.
+//
+// So this reconciler reads the failure's subjects and acts on those. It also
+// republishes to speculate afterwards, because a dead letter here consumes an
+// edge the queue needed. Speculation is driven only by messages; a batch
+// admitted to Speculating produces no build to signal and no merge to conclude,
+// so once the message that would have funded it is gone, nothing is left to
+// look at it again. Without the republish the failure of one batch silently
+// strands every other batch in the queue.
+type speculateController struct {
+	logger        *zap.SugaredLogger
+	metricsScope  tally.Scope
+	stores        storage.Factory
+	registry      consumer.TopicRegistry
+	topicKey      consumer.TopicKey
+	consumerGroup string
+}
+
+// Verify speculateController implements consumer.Controller at compile time.
+var _ consumer.Controller = (*speculateController)(nil)
+
+// NewDLQSpeculateController builds the DLQ controller for the speculate topic.
+// topicKey must be the DLQ topic key (typically TopicKey(primary)).
+func NewDLQSpeculateController(
+	logger *zap.SugaredLogger,
+	scope tally.Scope,
+	stores storage.Factory,
+	registry consumer.TopicRegistry,
+	topicKey consumer.TopicKey,
+	consumerGroup string,
+) consumer.Controller {
+	name := string(topicKey) + "_controller"
+	return &speculateController{
+		logger:        logger.Named(name),
+		metricsScope:  scope.SubScope(name),
+		stores:        stores,
+		registry:      registry,
+		topicKey:      topicKey,
+		consumerGroup: consumerGroup,
+	}
+}
+
+// Process reconciles a single dead-lettered speculate message.
+func (c *speculateController) Process(ctx context.Context, delivery consumer.Delivery) error {
+	const opName = "process"
+
+	msg := delivery.Message()
+
+	bid, err := entity.BatchIDFromBytes(msg.Payload)
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "deserialize_errors", 1)
+		return fmt.Errorf("failed to decode batch id from dlq payload: %w", err)
+	}
+	if bid.ID == "" {
+		metrics.NamedCounter(c.metricsScope, opName, "empty_id_errors", 1)
+		return fmt.Errorf("dlq payload decoded to empty batch id")
+	}
+
+	store, err := c.stores.For(storage.Config{QueueName: bid.Queue})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "storage_resolve_errors", 1)
+		// Non-retryable: a missing or unresolvable queue is a malformed message.
+		return fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err)
+	}
+
+	lastError, failureMeta := failureContext(delivery)
+	culprits, attribution := c.blame(delivery, bid.ID)
+	failureMeta["dlq.attribution"] = attribution
+
+	c.logger.Warnw("dlq message received",
+		"batch_id", bid.ID,
+		"queue", bid.Queue,
+		"attempt", delivery.Attempt(),
+		"attribution", attribution,
+		"culprits", culprits,
+		"dlq_last_error", lastError,
+	)
+	metrics.NamedCounter(c.metricsScope, opName, "reconciled", 1,
+		metrics.NewTag("attribution", attribution))
+
+	progressed := false
+	for _, batchID := range culprits {
+		transitioned, err := failBatch(ctx, store, c.registry, c.logger, batchID, lastError, failureMeta)
+		if err != nil {
+			metrics.NamedCounter(c.metricsScope, opName, "reconcile_errors", 1)
+			return err
+		}
+		progressed = progressed || transitioned
+	}
+
+	if !progressed {
+		// A redelivery of a reconcile already done. Republishing here would
+		// hand the queue the same message again on every redelivery, forever.
+		return nil
+	}
+
+	return c.retrigger(ctx, store, bid.Queue)
+}
+
+// blame decides which batches to fail, and reports how that was decided so the
+// answer is visible on the request afterwards rather than inferred.
+//
+// A failure that names batches is taken at its word — those are the batches the
+// run was actually working on when it failed. Anything else falls back to the
+// batch on the message: that is a guess, but DLQ reconciliation exists so
+// requests cannot sit non-terminal forever, and that guarantee has to hold even
+// when nothing can say which batch was at fault.
+func (c *speculateController) blame(delivery consumer.Delivery, payloadBatchID string) ([]string, string) {
+	f, failed := delivery.Failure()
+	if !failed {
+		return []string{payloadBatchID}, "unattributed"
+	}
+
+	if batches := f.IDsOfType(entity.SubjectTypeBatch); len(batches) > 0 {
+		return batches, "batch"
+	}
+
+	// A queue subject says no batch was at fault — a failure listing or
+	// planning the queue. There is still no other way to release the message's
+	// requests, so the named batch is failed, labelled for what it is.
+	if len(f.IDsOfType(entity.SubjectTypeQueue)) > 0 {
+		return []string{payloadBatchID}, "queue"
+	}
+
+	return []string{payloadBatchID}, "unattributed"
+}
+
+// retrigger wakes the queue if it still holds live batches, restoring the edge
+// this dead letter consumed.
+//
+// It names a batch that is still live rather than the one just failed, and runs
+// only after a reconcile that actually transitioned something. Together those
+// make it terminate: each pass fails one more batch and hands the queue another
+// run, so a queue that is genuinely broken drains to empty — every batch
+// recorded with a reason — instead of stranding, while a queue whose failure
+// was transient or queue-wide simply recovers on the next run.
+func (c *speculateController) retrigger(ctx context.Context, store storage.Storage, queue string) error {
+	const opName = "process"
+
+	live, err := corebatch.ListByStates(ctx, store, entity.ActiveBatchStates())
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
+		return fmt.Errorf("failed to list live batches of queue %s: %w", queue, err)
+	}
+	if len(live) == 0 {
+		return nil
+	}
+
+	// ListByStates does not promise an order, so pick deterministically. Any
+	// live batch would wake the queue — the run re-plans all of it — but a
+	// stable choice keeps the behaviour reproducible.
+	next := live[0].ID
+	for _, b := range live[1:] {
+		if b.ID < next {
+			next = b.ID
+		}
+	}
+
+	payload, err := entity.BatchID{ID: next, Queue: queue}.ToBytes()
+	if err != nil {
+		return fmt.Errorf("failed to serialize batch ID: %w", err)
+	}
+
+	// A distinct message ID every time: the queue deduplicates on
+	// (topic, partition, ID) against rows it has not collected yet, so reusing
+	// the batch ID would make this wake-up a silent no-op.
+	if err := publish.Message(ctx, c.registry, topickey.TopicKeySpeculate, publish.UniqueID(next), payload, queue); err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
+		return fmt.Errorf("failed to re-trigger speculation for queue %s: %w", queue, err)
+	}
+
+	metrics.NamedCounter(c.metricsScope, opName, "retriggered", 1)
+	c.logger.Infow("re-triggered speculation after dlq reconcile",
+		"queue", queue,
+		"batch_id", next,
+		"live_batches", len(live),
+	)
+	return nil
+}
+
+// Name returns the controller name for logging and metrics.
+func (c *speculateController) Name() string {
+	return string(c.topicKey)
+}
+
+// TopicKey returns the topic key this controller subscribes to.
+func (c *speculateController) TopicKey() consumer.TopicKey {
+	return c.topicKey
+}
+
+// ConsumerGroup returns the consumer group for offset tracking.
+func (c *speculateController) ConsumerGroup() string {
+	return c.consumerGroup
+}
diff --git a/submitqueue/orchestrator/controller/dlq/speculate_test.go b/submitqueue/orchestrator/controller/dlq/speculate_test.go
new file mode 100644
index 000000000..3cb627fbe
--- /dev/null
+++ b/submitqueue/orchestrator/controller/dlq/speculate_test.go
@@ -0,0 +1,284 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dlq
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber/submitqueue/platform/base/failure"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+	queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
+	"github.com/uber/submitqueue/submitqueue/core/topickey"
+	"github.com/uber/submitqueue/submitqueue/entity"
+	storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
+	"go.uber.org/mock/gomock"
+	"go.uber.org/zap/zaptest"
+)
+
+func newSpeculateController(registry consumer.TopicRegistry, store *storagemock.MockStorage, t *testing.T) consumer.Controller {
+	return NewDLQSpeculateController(
+		zaptest.NewLogger(t).Sugar(), testScope(), staticStorageFactory{store: store},
+		registry, TopicKey(topickey.TopicKeySpeculate), "orchestrator-speculate-dlq",
+	)
+}
+
+// speculateRegistry captures the batch IDs republished to the speculate topic,
+// and counts log publishes so the fan-out stays exercised.
+func speculateRegistry(t *testing.T, ctrl *gomock.Controller, logPublishes int, captured *[]string, logs *[]entity.RequestLog) consumer.TopicRegistry {
+	logPublisher := queuemock.NewMockPublisher(ctrl)
+	logPublisher.EXPECT().Publish(gomock.Any(), "log", gomock.Any()).DoAndReturn(
+		func(_ context.Context, _ string, message entityqueue.Message) error {
+			entry, err := entity.RequestLogFromBytes(message.Payload)
+			require.NoError(t, err)
+			if logs != nil {
+				*logs = append(*logs, entry)
+			}
+			return nil
+		},
+	).Times(logPublishes)
+	logQueue := queuemock.NewMockQueue(ctrl)
+	logQueue.EXPECT().Publisher().Return(logPublisher).Times(logPublishes)
+
+	specPublisher := queuemock.NewMockPublisher(ctrl)
+	specPublisher.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).DoAndReturn(
+		func(_ context.Context, _ string, message entityqueue.Message) error {
+			bid, err := entity.BatchIDFromBytes(message.Payload)
+			require.NoError(t, err)
+			*captured = append(*captured, bid.ID)
+			return nil
+		},
+	).AnyTimes()
+	specQueue := queuemock.NewMockQueue(ctrl)
+	specQueue.EXPECT().Publisher().Return(specPublisher).AnyTimes()
+
+	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{Key: topickey.TopicKeyLog, Name: "log", Queue: logQueue},
+		{Key: topickey.TopicKeySpeculate, Name: "speculate", Queue: specQueue},
+	})
+	require.NoError(t, err)
+	return registry
+}
+
+// A speculate run covers the whole queue, so the batch that failed is often not
+// the batch the message names. The subjects on the failure are what say which,
+// and failing the named one regardless would terminate an innocent batch while
+// leaving the real culprit running.
+func TestDLQSpeculateController_Process_Attribution(t *testing.T) {
+	tests := []struct {
+		name            string
+		recordedFailure failure.Failure
+		failed          bool
+		wantFailedBatch string
+		wantAttribution string
+	}{
+		{
+			name:            "batch subject blames that batch, not the message's",
+			recordedFailure: failure.New("read failed", entity.BatchSubject("q/batch/other")),
+			failed:          true,
+			wantFailedBatch: "q/batch/other",
+			wantAttribution: "batch",
+		},
+		{
+			name:            "queue subject falls back to the message's batch",
+			recordedFailure: failure.New("speculator failed", entity.QueueSubject("q")),
+			failed:          true,
+			wantFailedBatch: "q/batch/named",
+			wantAttribution: "queue",
+		},
+		{
+			name:            "no subjects falls back and says so",
+			recordedFailure: failure.New("exceeded retry limit"),
+			failed:          true,
+			wantFailedBatch: "q/batch/named",
+			wantAttribution: "unattributed",
+		},
+		{
+			name:            "no recorded failure at all falls back",
+			wantFailedBatch: "q/batch/named",
+			wantAttribution: "unattributed",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+
+			blamed := entity.Batch{
+				ID: tt.wantFailedBatch, Queue: "q", Contains: []string{"q/1"},
+				State: entity.BatchStateSpeculating, Version: 2,
+			}
+			batchStore := storagemock.NewMockBatchStore(ctrl)
+			batchStore.EXPECT().Get(gomock.Any(), tt.wantFailedBatch).Return(blamed, nil)
+			batchStore.EXPECT().Update(gomock.Any(), batchWithState(blamed, entity.BatchStateFailed), int32(2), int32(3)).Return(nil)
+
+			request := entity.Request{ID: "q/1", Version: 1, State: entity.RequestStateProcessing}
+			requestStore := storagemock.NewMockRequestStore(ctrl)
+			requestStore.EXPECT().Get(gomock.Any(), "q/1").Return(request, nil)
+			requestStore.EXPECT().Update(gomock.Any(), requestWithState(request, entity.RequestStateError), int32(1), int32(2)).Return(nil)
+
+			// The queue drained with the blamed batch, so no re-trigger.
+			queueBatchState := storagemock.NewMockQueueBatchStateStore(ctrl)
+			queueBatchState.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+			queueBatchState.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+			queueBatchState.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+
+			store := storagemock.NewMockStorage(ctrl)
+			store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
+			store.EXPECT().GetRequestStore().Return(requestStore).AnyTimes()
+			store.EXPECT().GetQueueBatchStateStore().Return(queueBatchState).AnyTimes()
+
+			var republished []string
+			var logs []entity.RequestLog
+			registry := speculateRegistry(t, ctrl, 1, &republished, &logs)
+			c := newSpeculateController(registry, store, t)
+
+			payload, err := entity.BatchID{ID: "q/batch/named", Queue: "q"}.ToBytes()
+			require.NoError(t, err)
+
+			delivery := newMockDeliveryWithFailure(ctrl, payload, tt.recordedFailure, tt.failed)
+			require.NoError(t, c.Process(context.Background(), delivery))
+
+			// The attribution is recorded where a user can see it, so a
+			// fallback is never mistaken for a confident answer.
+			require.Len(t, logs, 1)
+			assert.Equal(t, tt.wantAttribution, logs[0].Metadata["dlq.attribution"])
+			assert.Equal(t, "validate", logs[0].Metadata["dlq.original_topic"])
+		})
+	}
+}
+
+// The dead letter that consumed the queue's last edge has to leave one behind,
+// or every other batch admitted to speculating sits there with nothing to drive
+// it — the stranding this reconciler exists to stop.
+func TestDLQSpeculateController_Process_RetriggersQueue(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	failedBatch := entity.Batch{
+		ID: "q/batch/1", Queue: "q", Contains: nil,
+		State: entity.BatchStateSpeculating, Version: 1,
+	}
+	// Two still live afterwards; the lower ID is chosen so the wake-up is
+	// reproducible despite ListByStates promising no order.
+	liveA := entity.Batch{ID: "q/batch/2", Queue: "q", State: entity.BatchStateSpeculating, Version: 1}
+	liveB := entity.Batch{ID: "q/batch/3", Queue: "q", State: entity.BatchStateSpeculating, Version: 1}
+
+	batchStore := storagemock.NewMockBatchStore(ctrl)
+	batchStore.EXPECT().Get(gomock.Any(), "q/batch/1").Return(failedBatch, nil)
+	batchStore.EXPECT().Update(gomock.Any(), batchWithState(failedBatch, entity.BatchStateFailed), int32(1), int32(2)).Return(nil)
+	batchStore.EXPECT().Get(gomock.Any(), "q/batch/3").Return(liveB, nil).AnyTimes()
+	batchStore.EXPECT().Get(gomock.Any(), "q/batch/2").Return(liveA, nil).AnyTimes()
+
+	queueBatchState := storagemock.NewMockQueueBatchStateStore(ctrl)
+	queueBatchState.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	queueBatchState.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	queueBatchState.EXPECT().List(gomock.Any(), entity.BatchStateSpeculating).Return([]entity.QueueBatchState{
+		{Queue: "q", State: entity.BatchStateSpeculating, BatchID: "q/batch/3"},
+		{Queue: "q", State: entity.BatchStateSpeculating, BatchID: "q/batch/2"},
+	}, nil).AnyTimes()
+	queueBatchState.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+
+	store := storagemock.NewMockStorage(ctrl)
+	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
+	store.EXPECT().GetQueueBatchStateStore().Return(queueBatchState).AnyTimes()
+
+	var republished []string
+	registry := speculateRegistry(t, ctrl, 0, &republished, nil)
+	c := newSpeculateController(registry, store, t)
+
+	payload, err := entity.BatchID{ID: "q/batch/1", Queue: "q"}.ToBytes()
+	require.NoError(t, err)
+
+	delivery := newMockDeliveryWithFailure(ctrl, payload, failure.New("boom"), true)
+	require.NoError(t, c.Process(context.Background(), delivery))
+
+	assert.Equal(t, []string{"q/batch/2"}, republished)
+}
+
+// Redelivery of a reconcile already done must publish nothing. Republishing
+// unconditionally would hand the queue a fresh message every time this message
+// came back, and a permanently failing queue would never stop.
+func TestDLQSpeculateController_Process_NoRetriggerWithoutProgress(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	already := entity.Batch{
+		ID: "q/batch/1", Queue: "q", Contains: nil,
+		State: entity.BatchStateFailed, Version: 4,
+	}
+	batchStore := storagemock.NewMockBatchStore(ctrl)
+	batchStore.EXPECT().Get(gomock.Any(), "q/batch/1").Return(already, nil)
+
+	queueBatchState := storagemock.NewMockQueueBatchStateStore(ctrl)
+	queueBatchState.EXPECT().Put(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	queueBatchState.EXPECT().Delete(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	// Live batches remain, so only the progress guard can be what stops it.
+	queueBatchState.EXPECT().List(gomock.Any(), entity.BatchStateSpeculating).Return([]entity.QueueBatchState{
+		{Queue: "q", State: entity.BatchStateSpeculating, BatchID: "q/batch/2"},
+	}, nil).AnyTimes()
+	queueBatchState.EXPECT().List(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+
+	store := storagemock.NewMockStorage(ctrl)
+	store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes()
+	store.EXPECT().GetQueueBatchStateStore().Return(queueBatchState).AnyTimes()
+
+	var republished []string
+	registry := speculateRegistry(t, ctrl, 0, &republished, nil)
+	c := newSpeculateController(registry, store, t)
+
+	payload, err := entity.BatchID{ID: "q/batch/1", Queue: "q"}.ToBytes()
+	require.NoError(t, err)
+
+	delivery := newMockDeliveryWithFailure(ctrl, payload, failure.New("boom"), true)
+	require.NoError(t, c.Process(context.Background(), delivery))
+
+	assert.Empty(t, republished, "an already-failed batch is not progress")
+}
+
+func TestDLQSpeculateController_InterfaceAndAccessors(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	store := storagemock.NewMockStorage(ctrl)
+
+	c := newSpeculateController(consumer.TopicRegistry{}, store, t)
+
+	assert.Equal(t, "speculate_dlq", c.Name())
+	assert.Equal(t, consumer.TopicKey("speculate_dlq"), c.TopicKey())
+	assert.Equal(t, "orchestrator-speculate-dlq", c.ConsumerGroup())
+}
+
+func TestDLQSpeculateController_Process_MalformedPayloadFails(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	store := storagemock.NewMockStorage(ctrl)
+
+	c := newSpeculateController(consumer.TopicRegistry{}, store, t)
+
+	delivery := newMockDelivery(ctrl, []byte("garbage"))
+	require.Error(t, c.Process(context.Background(), delivery))
+}
+
+func TestDLQSpeculateController_Process_EmptyIDFails(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	store := storagemock.NewMockStorage(ctrl)
+
+	c := newSpeculateController(consumer.TopicRegistry{}, store, t)
+
+	payload, err := entity.BatchID{ID: ""}.ToBytes()
+	require.NoError(t, err)
+
+	delivery := newMockDelivery(ctrl, payload)
+	require.Error(t, c.Process(context.Background(), delivery))
+}
diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel
index f4eafb864..fdcf53146 100644
--- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel
+++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel
@@ -15,7 +15,9 @@ go_library(
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/base/failure:go_default_library",
         "//platform/consumer:go_default_library",
+        "//platform/errs:go_default_library",
         "//platform/metrics:go_default_library",
         "//submitqueue/core/batch:go_default_library",
         "//submitqueue/core/publish:go_default_library",
diff --git a/submitqueue/orchestrator/controller/speculate/dispatch.go b/submitqueue/orchestrator/controller/speculate/dispatch.go
index 24f991058..1732eca59 100644
--- a/submitqueue/orchestrator/controller/speculate/dispatch.go
+++ b/submitqueue/orchestrator/controller/speculate/dispatch.go
@@ -87,7 +87,9 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot,
 					)
 					continue
 				}
-				return err
+				// The loop covers every in-flight batch, so the head whose
+				// write failed is usually not the one named on the message.
+				return c.attributed(err, entity.BatchSubject(batch.ID))
 			}
 		}
 
@@ -100,7 +102,8 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot,
 		if hasActionablePaths(set) {
 			if err := c.publishBatchID(ctx, topickey.TopicKeyBuild, batch.ID, queue, batch.ID); err != nil {
 				metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
-				return fmt.Errorf("failed to publish batch %s to build: %w", batch.ID, err)
+				return c.attributed(fmt.Errorf("failed to publish batch %s to build: %w", batch.ID, err),
+					entity.BatchSubject(batch.ID))
 			}
 		}
 	}
diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go
index 668f7f0de..206be5455 100644
--- a/submitqueue/orchestrator/controller/speculate/finalize.go
+++ b/submitqueue/orchestrator/controller/speculate/finalize.go
@@ -197,13 +197,20 @@ func (c *Controller) commitOutcome(ctx context.Context, snap *snapshot, batch en
 				snap.markClean(batch.ID)
 				return false, nil
 			}
-			return false, err
+			return false, c.attributed(err, entity.BatchSubject(batch.ID))
 		}
 		snap.pathSets[batch.ID] = stored
 		snap.markClean(batch.ID)
 	}
 
-	return c.applyOutcome(ctx, snap.store, batch, decision, snap.isTrigger(batch.ID))
+	// Attributed here rather than at each leaf: this is the one place that
+	// knows which batch the whole commit is for, and the finalize loop runs it
+	// over every decided batch in the queue — rarely the one on the message.
+	landed, err := c.applyOutcome(ctx, snap.store, batch, decision, snap.isTrigger(batch.ID))
+	if err != nil {
+		return landed, c.attributed(err, entity.BatchSubject(batch.ID))
+	}
+	return landed, nil
 }
 
 // recordOutcome writes an outcome's terminal state back into the snapshot so
diff --git a/submitqueue/orchestrator/controller/speculate/run.go b/submitqueue/orchestrator/controller/speculate/run.go
index 044df5455..b5c108998 100644
--- a/submitqueue/orchestrator/controller/speculate/run.go
+++ b/submitqueue/orchestrator/controller/speculate/run.go
@@ -77,7 +77,8 @@ func (c *Controller) read(ctx context.Context, store storage.Storage, queue stri
 	inFlight, err := corebatch.ListByStates(ctx, store, entity.ActiveBatchStates())
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-		return snapshot{}, fmt.Errorf("failed to list in-flight batches of queue %s: %w", queue, err)
+		return snapshot{}, c.attributed(fmt.Errorf("failed to list in-flight batches of queue %s: %w", queue, err),
+			entity.QueueSubject(queue))
 	}
 
 	snap := snapshot{
@@ -106,7 +107,8 @@ func (c *Controller) read(ctx context.Context, store storage.Storage, queue stri
 			dep, err := store.GetBatchStore().Get(ctx, depID)
 			if err != nil {
 				metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-				return snapshot{}, fmt.Errorf("failed to get dependency batch %s of %s: %w", depID, batch.ID, err)
+				return snapshot{}, c.attributed(fmt.Errorf("failed to get dependency batch %s of %s: %w", depID, batch.ID, err),
+					entity.BatchSubject(depID))
 			}
 			snap.batches[depID] = dep
 		}
@@ -120,7 +122,8 @@ func (c *Controller) read(ctx context.Context, store storage.Storage, queue stri
 				continue
 			}
 			metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-			return snapshot{}, fmt.Errorf("failed to get path set for batch %s: %w", batch.ID, err)
+			return snapshot{}, c.attributed(fmt.Errorf("failed to get path set for batch %s: %w", batch.ID, err),
+				entity.BatchSubject(batch.ID))
 		}
 		changed, err := c.updatePathsFromBuilds(ctx, store, &set)
 		if err != nil {
@@ -180,7 +183,8 @@ func (c *Controller) updatePathsFromBuilds(ctx context.Context, store storage.St
 		}
 		if err != nil {
 			metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-			return false, fmt.Errorf("failed to look up build for path %s attempt %d: %w", entry.ID, entry.Attempt, err)
+			return false, c.attributed(fmt.Errorf("failed to look up build for path %s attempt %d: %w", entry.ID, entry.Attempt, err),
+				entity.BatchSubject(set.Head))
 		}
 
 		build, err := store.GetBuildStore().Get(ctx, link.BuildID)
@@ -193,7 +197,8 @@ func (c *Controller) updatePathsFromBuilds(ctx context.Context, store storage.St
 				continue
 			}
 			metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-			return false, fmt.Errorf("failed to get build %s for path %s: %w", link.BuildID, entry.ID, err)
+			return false, c.attributed(fmt.Errorf("failed to get build %s for path %s: %w", link.BuildID, entry.ID, err),
+				entity.BatchSubject(set.Head))
 		}
 
 		if !build.Status.IsTerminal() {
@@ -260,7 +265,8 @@ func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]en
 	spec, err := c.speculators.For(speculator.Config{QueueName: queue})
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "speculator_errors", 1)
-		return nil, fmt.Errorf("failed to build speculator for queue %s: %w", queue, err)
+		return nil, c.attributed(fmt.Errorf("failed to build speculator for queue %s: %w", queue, err),
+			entity.QueueSubject(queue))
 	}
 
 	sets := make([]entity.SpeculationPathSet, 0, len(snap.pathSets))
@@ -273,7 +279,8 @@ func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]en
 	proposals, err := spec.Speculate(ctx, slices.Collect(maps.Values(snap.batches)), sets)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "speculator_errors", 1)
-		return nil, fmt.Errorf("speculator failed for queue %s: %w", queue, err)
+		return nil, c.attributed(fmt.Errorf("speculator failed for queue %s: %w", queue, err),
+			entity.QueueSubject(queue))
 	}
 	return proposals, nil
 }
diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go
index a9720d952..95540770c 100644
--- a/submitqueue/orchestrator/controller/speculate/speculate.go
+++ b/submitqueue/orchestrator/controller/speculate/speculate.go
@@ -19,7 +19,9 @@ import (
 	"fmt"
 
 	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/base/failure"
 	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/errs"
 	"github.com/uber/submitqueue/platform/metrics"
 	corebatch "github.com/uber/submitqueue/submitqueue/core/batch"
 	"github.com/uber/submitqueue/submitqueue/core/publish"
@@ -101,20 +103,23 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "storage_resolve_errors", 1)
 		// Non-retryable: a missing or unresolvable queue is a malformed message.
-		return fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err)
+		return c.attributed(fmt.Errorf("failed to resolve storage for queue %q: %w", bid.Queue, err),
+			entity.QueueSubject(bid.Queue))
 	}
 
 	batch, err := store.GetBatchStore().Get(ctx, bid.ID)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-		return fmt.Errorf("failed to get batch %s: %w", bid.ID, err)
+		return c.attributed(fmt.Errorf("failed to get batch %s: %w", bid.ID, err),
+			entity.BatchSubject(bid.ID))
 	}
 
 	// 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)
+		return c.attributed(fmt.Errorf("payload queue %q does not match queue %q of batch %s", bid.Queue, batch.Queue, batch.ID),
+			entity.BatchSubject(batch.ID))
 	}
 
 	if batch.State.IsTerminal() {
@@ -145,7 +150,7 @@ func (c *Controller) admit(ctx context.Context, store storage.Storage, batch ent
 	updated, err := corebatch.Transition(ctx, store, batch, entity.BatchStateSpeculating)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
-		return batch, err
+		return batch, c.attributed(err, entity.BatchSubject(batch.ID))
 	}
 
 	metrics.NamedCounter(c.metricsScope, opName, "admitted", 1)
@@ -163,7 +168,8 @@ func (c *Controller) admit(ctx context.Context, store storage.Storage, batch ent
 func (c *Controller) fanout(ctx context.Context, batchID, queue string) error {
 	if err := c.publishBatchID(ctx, topickey.TopicKeyConclude, batchID, queue, queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
-		return fmt.Errorf("failed to publish to conclude: %w", err)
+		return c.attributed(fmt.Errorf("failed to publish to conclude: %w", err),
+			entity.BatchSubject(batchID))
 	}
 	return nil
 }
@@ -190,6 +196,25 @@ func (c *Controller) publishBatchID(ctx context.Context, key consumer.TopicKey,
 	return publish.Message(ctx, c.registry, key, publish.UniqueID(batchID), payload, partitionKey)
 }
 
+// attributed records what a failure was about and counts it by subject type.
+//
+// It exists because this stage's message names one batch but its work covers
+// the whole queue: a run reads every in-flight batch, hands them all to the
+// Speculator, and commits outcomes for any of them. So most failures here are
+// not about the batch on the message — they are about some other batch, or
+// about the queue itself — and a reconciler that assumed otherwise would
+// terminate a batch that was never at fault.
+//
+// Only the attribution is added. Whether the error is retryable stays with the
+// classifiers, which read the cause underneath this wrapper unchanged.
+func (c *Controller) attributed(err error, subjects ...failure.Subject) error {
+	for _, s := range subjects {
+		metrics.NamedCounter(c.metricsScope, opName, "attributed_failure", 1,
+			metrics.NewTag("subject_type", s.Type))
+	}
+	return errs.Attribute(err, subjects...)
+}
+
 // Name returns the controller name for logging and metrics.
 func (c *Controller) Name() string {
 	return "speculate"
diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go
index 0f4e55008..60cfef5e2 100644
--- a/submitqueue/orchestrator/pipeline.go
+++ b/submitqueue/orchestrator/pipeline.go
@@ -153,7 +153,7 @@ var Stages = []pipeline.Stage[Deps]{
 			return speculate.NewController(d.Logger, d.Scope, d.Storage, d.Speculator, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
 		},
 		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
-			return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
+			return dlq.NewDLQSpeculateController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
 		},
 	},
 	{