diff --git a/doc/rfc/stovepipe/steps/buildanalysis.md b/doc/rfc/stovepipe/steps/buildanalysis.md
index bca6c68f2..7c088d7bb 100644
--- a/doc/rfc/stovepipe/steps/buildanalysis.md
+++ b/doc/rfc/stovepipe/steps/buildanalysis.md
@@ -1,6 +1,6 @@
-# Build issue observation
+# Build issue detection
 
-Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success or recovery.
+Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success. The reference host registers an issue-tracking publication route but does not yet consume those notifications.
 
 ## Start and completion notices
 
@@ -10,9 +10,9 @@ Start and completion notices use distinct deterministic message IDs and the same
 
 Publication errors are returned through the producing stage's retry/DLQ policy. Analysis processing does not gate record, but notification publication is still part of the producing stage: a failed start publication can send the build-stage message to its existing fail-closed recovery path.
 
-## Observations and polling
+## Build issues and polling
 
-The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build observation persistence, polling, and the stop condition for Requests that end without a recorded build result. It only accumulates historical observations; it does not maintain open issues or infer recovery from absence.
+The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build issue persistence, polling, and the stop condition for Requests that end without a recorded build result. It accumulates per-build issues and publishes queue-partitioned issue-tracking notifications after persistence; it does not maintain open issues itself. Successful terminal analysis requests final reconciliation only after the Request records that Build as its winning outcome. Terminal detection must return the complete cumulative issue set, including issues returned by earlier calls, and analysis persists that set before publishing final reconciliation. Issue tracking applies all accumulated build issues before permitting any absence-based closure; a task-wide failure preserves existing issues across all tasks. See the controller documentation for handoff and deduplication rules.
 
 Status-polling DLQ recovery does not publish a build-completion notice. The analysis polling loop observes the ended Request through storage instead.
 
diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go
index 8a9279fe3..6e3f98b1d 100644
--- a/service/stovepipe/server/main.go
+++ b/service/stovepipe/server/main.go
@@ -409,7 +409,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	analysisController := buildanalysis.NewController(logger, scope, store, detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
+	analysisController := buildanalysis.NewController(logger, scope, store, detectorFactory, registry, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
 	if err := c.Register(analysisController); err != nil {
 		return count, fmt.Errorf("failed to register buildanalysis controller: %w", err)
 	}
@@ -562,6 +562,11 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Queue:        q,
 			Subscription: extqueue.DLQSubscriptionConfig(subscriberName, "stovepipe-hook-dlq"),
 		},
+		{
+			Key:   stovepipemq.TopicKeyIssueTrack,
+			Name:  "issuetrack",
+			Queue: q,
+		},
 		{
 			Key:          stovepipemq.TopicKeyBuildAnalysis,
 			Name:         "buildanalysis",
diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go
index 8387c21ed..3f458b896 100644
--- a/service/stovepipe/server/main_test.go
+++ b/service/stovepipe/server/main_test.go
@@ -92,6 +92,15 @@ func TestEveryRegisteredControllerResolvesInTheTopicRegistry(t *testing.T) {
 	}
 }
 
+func TestIssueTrackingHasPublicationRouteWithoutConsumption(t *testing.T) {
+	registry, primary, deadLetter := registeredControllers(t)
+	name, ok := registry.TopicName(stovepipemq.TopicKeyIssueTrack)
+	require.True(t, ok)
+	assert.Equal(t, "issuetrack", name)
+	assert.NotContains(t, topicKeys(primary), stovepipemq.TopicKeyIssueTrack)
+	assert.NotContains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyIssueTrack))
+}
+
 func TestHookStage(t *testing.T) {
 	registry, primary, deadLetter := registeredControllers(t)
 
diff --git a/stovepipe/controller/build/build_analysis_test.go b/stovepipe/controller/build/build_analysis_test.go
index e6b2cfcfd..0609168d0 100644
--- a/stovepipe/controller/build/build_analysis_test.go
+++ b/stovepipe/controller/build/build_analysis_test.go
@@ -110,15 +110,16 @@ func TestBuildAnalysisPipeline(t *testing.T) {
 
 	detectorFactory := buildissuemock.NewMockDetectorFactory(ctrl)
 	detector := buildissuemock.NewMockDetector(ctrl)
-	detectorFactory.EXPECT().For(buildissue.DetectorConfig{QueueName: testQueue}).Return(detector, nil).Times(3)
+	detectorFactory.EXPECT().For(buildissue.DetectorConfig{QueueName: testQueue}).Return(detector, nil).Times(4)
 	gomock.InOrder(
 		detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return([]entity.BuildIssue{a}, nil),
-		detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return([]entity.BuildIssue{a, b}, nil),
-		detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return([]entity.BuildIssue{a, b}, nil),
+		detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return([]entity.BuildIssue{b}, nil),
+		detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(nil, nil),
+		detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(nil, nil),
 	)
 
 	analysisController := buildanalysis.NewController(zap.NewNop().Sugar(), tally.NoopScope,
-		staticStorageFactory{store: m.store}, detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "analysis")
+		staticStorageFactory{store: m.store}, detectorFactory, buildController.registry, stovepipemq.TopicKeyBuildAnalysis, "analysis")
 	statusController := buildsignal.NewController(zap.NewNop().Sugar(), tally.NoopScope,
 		staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, buildController.registry,
 		stovepipemq.TopicKeyBuildSignal, "signal")
@@ -139,17 +140,37 @@ func TestBuildAnalysisPipeline(t *testing.T) {
 	assert.True(t, persisted[a])
 	assert.False(t, persisted[b])
 
+	secondPoll := delivered(published["buildanalysis"][0])
+	secondPoll.EXPECT().Hold(buildanalysis.PollDelayMs)
+	require.NoError(t, analysisController.Process(ctx, secondPoll))
+	assert.True(t, persisted[b])
+
 	require.NoError(t, statusController.Process(ctx, delivered(published["buildsignal"][0])))
 	require.Len(t, published["record"], 1)
 	require.Len(t, published["buildanalysis"], 2)
 	assert.Equal(t, entity.BuildStatusFailed, savedBuild.Status)
 	assert.Equal(t, entity.RequestStateFailed, request.State)
 	assert.Zero(t, queue.InFlightCount)
-	assert.False(t, persisted[b])
+	assert.True(t, persisted[b])
 	assert.NotEqual(t, published["buildanalysis"][0].ID, published["buildanalysis"][1].ID)
 
 	require.NoError(t, analysisController.Process(ctx, delivered(published["buildanalysis"][0])))
 	require.NoError(t, analysisController.Process(ctx, delivered(published["buildanalysis"][1])))
 	assert.Len(t, persisted, 2)
 	assert.True(t, persisted[b])
+	assert.True(t, persisted[a])
+
+	require.Len(t, published["issuetrack"], 4)
+	for i, msg := range published["issuetrack"] {
+		assert.Equal(t, testQueue, msg.PartitionKey)
+		assert.Equal(t, testQueue, msg.Tenant)
+		notification := &stovepipemq.IssueTrack{}
+		require.NoError(t, stovepipemq.Unmarshal(msg.Payload, notification))
+		assert.Equal(t, testBuildID, notification.GetBuildId())
+		assert.Equal(t, testQueue, notification.GetQueueName())
+		assert.Equal(t, i > 1, notification.GetFinalize())
+	}
+	assert.NotEqual(t, published["issuetrack"][0].ID, published["issuetrack"][1].ID)
+	assert.NotEqual(t, published["issuetrack"][1].ID, published["issuetrack"][2].ID)
+	assert.Equal(t, published["issuetrack"][2].ID, published["issuetrack"][3].ID)
 }
diff --git a/stovepipe/controller/build/build_test.go b/stovepipe/controller/build/build_test.go
index 6b83a5a7b..e98089e8e 100644
--- a/stovepipe/controller/build/build_test.go
+++ b/stovepipe/controller/build/build_test.go
@@ -99,6 +99,7 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMoc
 	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
 		{Key: stovepipemq.TopicKeyBuildSignal, Name: "buildsignal", Queue: queue},
 		{Key: stovepipemq.TopicKeyBuildAnalysis, Name: "buildanalysis", Queue: queue},
+		{Key: stovepipemq.TopicKeyIssueTrack, Name: "issuetrack", Queue: queue},
 		{Key: stovepipemq.TopicKeyRecord, Name: "record", Queue: queue},
 	})
 	require.NoError(t, err)
diff --git a/stovepipe/controller/buildanalysis/BUILD.bazel b/stovepipe/controller/buildanalysis/BUILD.bazel
index 37afa9972..ae5acadb4 100644
--- a/stovepipe/controller/buildanalysis/BUILD.bazel
+++ b/stovepipe/controller/buildanalysis/BUILD.bazel
@@ -12,6 +12,7 @@ go_library(
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
+        "//platform/publish:go_default_library",
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
@@ -36,7 +37,9 @@ go_test(
         "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
         "//platform/errs:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
         "//platform/metrics:go_default_library",
+        "//platform/publish:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/core/requestlog/mock:go_default_library",
diff --git a/stovepipe/controller/buildanalysis/README.md b/stovepipe/controller/buildanalysis/README.md
index 33cb3b3d2..fb5f71027 100644
--- a/stovepipe/controller/buildanalysis/README.md
+++ b/stovepipe/controller/buildanalysis/README.md
@@ -1,14 +1,22 @@
 # Build analysis
 
-## Observations
+## Build issues
 
-The controller reloads the queue-scoped Build, invokes the queue's issue detector, and persists each returned observation individually. Observations already stored for the same build, task, and unit within a queue are not persisted again, so repeated polls and retries after partial writes are safe. Detector results accompanying an error are discarded.
+The controller reloads the queue-scoped Build, invokes the queue's issue detector, and persists each returned build issue individually. Build issues already stored for the same build, task, and unit within a queue are not persisted again, so repeated polls and retries after partial writes are safe. Detector results accompanying an error are discarded.
 
-This controller only accumulates historical observations. An empty detection result neither deletes earlier observations nor establishes recovery. Open-issue projection, lifecycle history, and resolution are outside its scope.
+After persistence, the controller publishes an issue-tracking notification partitioned by queue. Issue tracking maintains open issues and lifecycle history; analysis never writes them directly. An empty detection result does not delete earlier build issues.
+
+## Issue-tracking handoff
+
+Refresh notifications let issue tracking open issues reported by the build without closing other open issues. Final notifications let it also close open issues absent from the build's accumulated issues, unless any task-wide failure was recorded or the build was canceled. A successful terminal detection must return the complete cumulative issue set for the build, including issues returned by earlier calls. Persisting that complete result before publication makes finalization independent of whether an earlier worker finished saving its results. An empty terminal result is valid only if the build has no reportable issues; inability to determine the complete set must return an error instead. See the [detector contract](../../extension/buildissue/README.md).
+
+All analysis deliveries for a build must share its build-ID partition and consumer group. The controller validates that partition key. Finalization requires a successful pass that began with a terminal Build and the Request's recorded winning build ID. If status becomes terminal before that outcome is recorded, analysis publishes a refresh and holds until the outcome is available. Known losing builds do not publish. A terminal Request without a recorded winning build permits a refresh but not closing issues based on absence.
+
+Every refresh gets a fresh notification ID because held polling deliveries do not identify individual passes. A fixed ID would suppress later refreshes after earlier notifications were consumed. Retries can therefore publish duplicate refreshes; issue tracking is idempotent. Final notifications use one deterministic ID per build, so duplicate terminal passes deduplicate. Publication errors return through the normal analysis retry/DLQ policy; no hold or successful completion occurs before publication succeeds.
 
 ## Polling and completion
 
-Every delivery reloads the Build and its Request before detection. After a successful pass, a nonterminal Build holds the delivery for two minutes; a terminal Build completes the delivery after persistence. The status read before detection determines whether the pass holds: if the Build finishes during detection, another delivery is needed to analyze it after observing terminal status. More than one delivery may analyze the same terminal Build; observation persistence remains idempotent.
+Every delivery reloads the Build and its Request before detection. After a successful pass, a nonterminal Build holds the delivery for two minutes; a terminal Build completes after persistence and publication once its Request outcome is known, or the Request has ended without one. A pass that begins with a nonterminal Build always holds, even if the Build finishes during detection, so another delivery analyzes it after observing terminal status. More than one delivery may analyze the same terminal Build; build-issue persistence remains idempotent.
 
 If the Request is terminal but the loaded Build is nonterminal, analysis reloads the Build before deciding to stop. The Build may have finished between the initial Build read and the Request read; reloading avoids skipping its final analysis. If it is still nonterminal, analysis stops without detection or a hold. This can occur when the status-polling DLQ handler fails the Request without learning the actual build result, leaving the Build at its last observed status.
 
@@ -16,11 +24,11 @@ Status polling and updates remain the responsibility of buildsignal. Analysis ne
 
 ## Errors
 
-Detection, persistence, configuration, state-loading, and message-validation errors are returned to the consumer for its normal retry or DLQ policy, whether the Build is running or terminal. Analysis errors do not use the two-minute polling hold. The consumer classifies errors and applies its configured retry limits and backoff; a final analysis can still fail and dead-letter.
+Detection, persistence, publication, configuration, state-loading, and message-validation errors are returned to the consumer for its normal retry or DLQ policy, whether the Build is running or terminal. Analysis errors do not use the two-minute polling hold. The consumer classifies errors and applies its configured retry limits and backoff; a final analysis can still fail and dead-letter.
 
 ## DLQ recovery
 
-The recovery controller records a `build_analysis_abandoned` event in the Request's history, with the build ID, to indicate that analysis could not complete and issue observations may be incomplete. It does not rerun detection, delete earlier observations, change Request or Build state, or release a build slot. Original failure details are logged rather than included in the immutable history event.
+The recovery controller records a `build_analysis_abandoned` event in the Request's history, with the build ID, to indicate that analysis could not complete and the recorded build issues may be incomplete. It does not rerun detection, delete earlier build issues, change Request or Build state, or release a build slot. Original failure details are logged rather than included in the immutable history event.
 
 History identity is stable per build, so repeated recovery deliveries and separate analysis notices for the same build retain one event. Different builds retain different events, even for the same Request. A later successful analysis does not erase the historical occurrence.
 
@@ -28,4 +36,4 @@ State-read and history-write errors are returned for retry. Hosts must run recov
 
 ## Host integration
 
-The reference service supplies a no-op detector and publishes analysis notices at build start and completion. It registers the normal analysis controller with bounded retries and the recovery controller with the existing DLQ consumer. See [build issue observation](../../../doc/rfc/stovepipe/steps/buildanalysis.md) for the publication rules and retry configuration.
+The reference service supplies a no-op detector and publishes analysis notices at build start and completion. It also registers a publish-only issue-tracking topic; issue-tracking consumption and recovery are not enabled yet. It registers the normal analysis controller with bounded retries and the recovery controller with the existing DLQ consumer. See [build issue detection](../../../doc/rfc/stovepipe/steps/buildanalysis.md) for the publication rules and retry configuration.
diff --git a/stovepipe/controller/buildanalysis/buildanalysis.go b/stovepipe/controller/buildanalysis/buildanalysis.go
index 4ebaff55e..8f28044fa 100644
--- a/stovepipe/controller/buildanalysis/buildanalysis.go
+++ b/stovepipe/controller/buildanalysis/buildanalysis.go
@@ -25,6 +25,7 @@ import (
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/platform/publish"
 	"github.com/uber/submitqueue/stovepipe/core/loader"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
@@ -33,17 +34,18 @@ import (
 	"go.uber.org/zap"
 )
 
-// PollDelayMs is the delay in milliseconds after a successful nonterminal pass.
+// PollDelayMs is the delay in milliseconds between successful polling passes.
 // Two minutes limits provider traffic during long-running builds.
 const PollDelayMs int64 = 120000
 
-// Controller accumulates observations while builds run and after completion.
-// Terminal analysis must finish persistence before its delivery completes.
+// Controller accumulates build issues while builds run and after completion.
+// Terminal analysis must finish persistence and publication before its delivery completes.
 type Controller struct {
 	logger          *zap.SugaredLogger
 	metricsScope    tally.Scope
 	stores          storage.Factory
 	detectorFactory buildissue.DetectorFactory
+	registry        consumer.TopicRegistry
 	topicKey        consumer.TopicKey
 	consumerGroup   string
 }
@@ -53,6 +55,7 @@ func NewController(
 	scope tally.Scope,
 	stores storage.Factory,
 	detectorFactory buildissue.DetectorFactory,
+	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
 ) *Controller {
@@ -61,13 +64,14 @@ func NewController(
 		metricsScope:    scope.SubScope("buildanalysis_controller"),
 		stores:          stores,
 		detectorFactory: detectorFactory,
+		registry:        registry,
 		topicKey:        topicKey,
 		consumerGroup:   consumerGroup,
 	}
 }
 
-// Process holds after a successful nonterminal pass. Errors are returned for
-// normal queue retry or DLQ handling; partial writes are safe to retry.
+// Process persists before publication and holds while further analysis is needed.
+// Errors use normal queue retry or DLQ handling; partial writes are safe to retry.
 func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) {
 	op := metrics.Begin(c.metricsScope, "process", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
 	defer func() { op.Complete(retErr) }()
@@ -85,6 +89,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	if analysis.GetId() == "" {
 		return fmt.Errorf("build analysis requires a build ID")
 	}
+	if msg.PartitionKey != analysis.GetId() {
+		return fmt.Errorf("build analysis must be partitioned by build %q", analysis.GetId())
+	}
 
 	store, err := c.stores.For(storage.Config{QueueName: analysis.GetQueueName()})
 	if err != nil {
@@ -104,6 +111,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	if request.Queue != analysis.GetQueueName() {
 		return fmt.Errorf("request queue %q does not match analysis queue %q", request.Queue, analysis.GetQueueName())
 	}
+	if request.TerminalBuildID != "" && request.TerminalBuildID != build.ID {
+		return nil
+	}
 
 	// The Build may finish between the initial Build read and the Request read.
 	// Recheck its status so stopping here does not skip final analysis.
@@ -137,9 +147,29 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		}
 	}
 
-	// A nonterminal status read before detection requires another pass,
-	// even if the build finishes while detection is running.
-	if !build.Status.IsTerminal() {
+	hold := !build.Status.IsTerminal()
+	finalize := false
+	if build.Status.IsTerminal() {
+		// Build status can become terminal before buildsignal records the winning build.
+		request, err = loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request")
+		if err != nil {
+			return err
+		}
+		if request.Queue != analysis.GetQueueName() {
+			return fmt.Errorf("request queue %q does not match analysis queue %q", request.Queue, analysis.GetQueueName())
+		}
+		if request.TerminalBuildID != "" && request.TerminalBuildID != build.ID {
+			return nil
+		}
+		finalize = request.TerminalBuildID == build.ID
+		hold = !finalize && !request.State.IsTerminal()
+	}
+
+	if err := c.publishIssueTracking(ctx, request.Queue, build.ID, finalize); err != nil {
+		return err
+	}
+
+	if hold {
 		delivery.Hold(PollDelayMs)
 	}
 
@@ -153,6 +183,27 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	return nil
 }
 
+func (c *Controller) publishIssueTracking(ctx context.Context, queue, buildID string, finalize bool) error {
+	payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{
+		BuildId: buildID, QueueName: queue, Finalize: finalize,
+	})
+	if err != nil {
+		return fmt.Errorf("serialize issue tracking notification: %w", err)
+	}
+
+	// Polls reuse one delivery, so no stable occurrence ID distinguishes refreshes.
+	messageID := publish.UniqueID(buildID)
+	if finalize {
+		messageID = publish.IntentID(buildID, "finished")
+	}
+	if err := publish.Message(ctx, c.registry, stovepipemq.TopicKeyIssueTrack, publish.MessageParams{
+		Tenant: queue, ID: messageID, Payload: payload, PartitionKey: queue,
+	}); err != nil {
+		return fmt.Errorf("publish issue tracking notification: %w", err)
+	}
+	return nil
+}
+
 func (c *Controller) Name() string { return "buildanalysis" }
 
 func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey }
diff --git a/stovepipe/controller/buildanalysis/buildanalysis_test.go b/stovepipe/controller/buildanalysis/buildanalysis_test.go
index 2cb3376e3..42d9a0ac3 100644
--- a/stovepipe/controller/buildanalysis/buildanalysis_test.go
+++ b/stovepipe/controller/buildanalysis/buildanalysis_test.go
@@ -23,9 +23,12 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
 	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
 	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
 	"github.com/uber/submitqueue/platform/errs"
+	mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
 	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/platform/publish"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/buildissue"
@@ -48,9 +51,11 @@ type analysisMocks struct {
 	store           *storagemock.MockStorage
 	builds          *storagemock.MockBuildStore
 	requests        *storagemock.MockRequestStore
+	requestRead     *gomock.Call
 	issues          *storagemock.MockBuildIssueStore
 	detectorFactory *buildissuemock.MockDetectorFactory
 	detector        *buildissuemock.MockDetector
+	publisher       *mqmock.MockPublisher
 	scope           tally.TestScope
 }
 
@@ -68,15 +73,23 @@ func newTestController(t *testing.T) (*Controller, analysisMocks) {
 		issues:          storagemock.NewMockBuildIssueStore(ctrl),
 		detectorFactory: buildissuemock.NewMockDetectorFactory(ctrl),
 		detector:        buildissuemock.NewMockDetector(ctrl),
+		publisher:       mqmock.NewMockPublisher(ctrl),
 		scope:           tally.NewTestScope("test", nil),
 	}
 
 	m.store.EXPECT().GetBuildStore().Return(m.builds).AnyTimes()
 	m.store.EXPECT().GetRequestStore().Return(m.requests).AnyTimes()
-	m.requests.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(context.Context, string) (entity.Request, error) { return *m.request, nil }).AnyTimes()
+	m.requestRead = m.requests.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(context.Context, string) (entity.Request, error) { return *m.request, nil }).AnyTimes()
 	m.store.EXPECT().GetBuildIssueStore().Return(m.issues).AnyTimes()
 
-	c := NewController(zap.NewNop().Sugar(), m.scope, m.stores, m.detectorFactory,
+	queue := mqmock.NewMockQueue(ctrl)
+	queue.EXPECT().Publisher().Return(m.publisher).AnyTimes()
+	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{Key: stovepipemq.TopicKeyIssueTrack, Name: "issue_track", Queue: queue},
+	})
+	require.NoError(t, err)
+
+	c := NewController(zap.NewNop().Sugar(), m.scope, m.stores, m.detectorFactory, registry,
 		stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
 
 	return c, m
@@ -102,13 +115,41 @@ func (m analysisMocks) delivery(t *testing.T) *consumermock.MockDelivery {
 }
 
 func (m analysisMocks) expectPass(status entity.BuildStatus, issues []entity.BuildIssue, err error) *gomock.Call {
+	if status.IsTerminal() {
+		m.request.TerminalBuildID = testBuildID
+	}
 	resolve := m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil)
 	load := m.builds.EXPECT().Get(gomock.Any(), testBuildID).
-		Return(entity.Build{ID: testBuildID, Status: status}, nil).After(resolve)
+		Return(entity.Build{ID: testBuildID, RequestID: m.request.ID, Status: status}, nil).After(resolve)
 	detector := m.detectorFactory.EXPECT().For(buildissue.DetectorConfig{QueueName: testQueue}).Return(m.detector, nil).After(load)
 	return m.detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(issues, err).After(detector)
 }
 
+func (m analysisMocks) expectNotification(t *testing.T, finalize bool, after *gomock.Call) *gomock.Call {
+	t.Helper()
+	call := m.publisher.EXPECT().Publish(gomock.Any(), "issue_track", gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error {
+			assert.Equal(t, testQueue, msg.Tenant)
+			assert.Equal(t, testQueue, msg.PartitionKey)
+			assert.Equal(t, testQueue, msg.Metadata[entityqueue.MetadataKeyQueueName])
+			notification := &stovepipemq.IssueTrack{}
+			require.NoError(t, stovepipemq.Unmarshal(msg.Payload, notification))
+			assert.Equal(t, testBuildID, notification.GetBuildId())
+			assert.Equal(t, testQueue, notification.GetQueueName())
+			assert.Equal(t, finalize, notification.GetFinalize())
+			if finalize {
+				assert.Equal(t, publish.IntentID(testBuildID, "finished"), msg.ID)
+			} else {
+				assert.NotEmpty(t, msg.ID)
+			}
+			return nil
+		})
+	if after != nil {
+		call.After(after)
+	}
+	return call
+}
+
 func TestProcessPersistsBeforeHoldingOrCompleting(t *testing.T) {
 	issue := entity.BuildIssue{TaskID: "validation", UnitID: "unit-a"}
 
@@ -128,6 +169,7 @@ func TestProcessPersistsBeforeHoldingOrCompleting(t *testing.T) {
 				if !empty {
 					last = m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(last)
 				}
+				last = m.expectNotification(t, status.IsTerminal(), last)
 
 				d := m.delivery(t)
 				if !status.IsTerminal() {
@@ -148,6 +190,7 @@ func TestProcessAcceptsRepeatedAndTaskWideObservations(t *testing.T) {
 	last = m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(last)
 	last = m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).
 		Return(fmt.Errorf("already recorded: %w", storage.ErrAlreadyExists)).After(last)
+	last = m.expectNotification(t, false, last)
 
 	d := m.delivery(t)
 	d.EXPECT().Hold(PollDelayMs).After(last)
@@ -181,6 +224,7 @@ func TestProcessRetriesPartialPersistenceAndReloadsStatus(t *testing.T) {
 			require.ErrorIs(t, err, writeErr)
 			assert.True(t, errs.IsRetryable(err))
 		} else {
+			last = m.expectNotification(t, pass.status.IsTerminal(), last)
 			if !pass.status.IsTerminal() {
 				d.EXPECT().Hold(PollDelayMs).After(last)
 			}
@@ -211,10 +255,13 @@ func TestProcessRunsAgainWhenBuildCompletesDuringDetection(t *testing.T) {
 
 	first := m.delivery(t)
 	first.EXPECT().Hold(PollDelayMs)
+	m.expectNotification(t, false, nil)
 
 	require.NoError(t, c.Process(context.Background(), first))
 
 	m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil)
+	m.request.TerminalBuildID = testBuildID
+	m.expectNotification(t, true, nil)
 
 	require.NoError(t, c.Process(context.Background(), m.delivery(t)))
 }
@@ -267,7 +314,7 @@ func TestProcessPropagatesDependencyErrors(t *testing.T) {
 }
 
 func TestProcessRejectsInvalidMessagesBeforeResolvingDependencies(t *testing.T) {
-	for _, kind := range []string{"malformed payload", "empty queue", "empty build ID", "empty tenant", "tenant mismatch", "metadata mismatch"} {
+	for _, kind := range []string{"malformed payload", "empty queue", "empty build ID", "empty tenant", "tenant mismatch", "metadata mismatch", "partition mismatch"} {
 		t.Run(kind, func(t *testing.T) {
 			c, m := newTestController(t)
 			msg := analysisMessage(t)
@@ -284,6 +331,8 @@ func TestProcessRejectsInvalidMessagesBeforeResolvingDependencies(t *testing.T)
 				msg.Tenant = "other/main"
 			case "metadata mismatch":
 				msg.Metadata[entityqueue.MetadataKeyQueueName] = "other/main"
+			case "partition mismatch":
+				msg.PartitionKey = testQueue
 			}
 
 			d := consumermock.NewMockDelivery(m.ctrl)
@@ -356,7 +405,8 @@ func TestProcessStillAnalyzesTerminalBuildOfTerminalRequest(t *testing.T) {
 	issue := entity.BuildIssue{TaskID: "validation", UnitID: "unit"}
 
 	detected := m.expectPass(entity.BuildStatusFailed, []entity.BuildIssue{issue}, nil)
-	m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected)
+	persisted := m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected)
+	m.expectNotification(t, true, persisted)
 
 	require.NoError(t, c.Process(context.Background(), m.delivery(t)))
 }
@@ -374,6 +424,7 @@ func TestProcessRejectsRequestFromDifferentQueue(t *testing.T) {
 func TestCompletionBetweenBuildAndRequestReadStillAnalyzes(t *testing.T) {
 	c, m := newTestController(t)
 	m.request.State = entity.RequestStateFailed
+	m.request.TerminalBuildID = testBuildID
 
 	m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil)
 	gomock.InOrder(
@@ -384,7 +435,8 @@ func TestCompletionBetweenBuildAndRequestReadStillAnalyzes(t *testing.T) {
 	m.detectorFactory.EXPECT().For(buildissue.DetectorConfig{QueueName: testQueue}).Return(m.detector, nil)
 	issue := entity.BuildIssue{TaskID: "validation", UnitID: "unit"}
 	detected := m.detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return([]entity.BuildIssue{issue}, nil)
-	m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected)
+	persisted := m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected)
+	m.expectNotification(t, true, persisted)
 
 	require.NoError(t, c.Process(context.Background(), m.delivery(t)))
 }
@@ -419,6 +471,8 @@ func TestProcessRetriesErrorsAndReloadsStatus(t *testing.T) {
 	assert.True(t, errs.IsRetryable(err))
 
 	m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil)
+	m.request.TerminalBuildID = testBuildID
+	m.expectNotification(t, true, nil)
 
 	require.NoError(t, c.Process(context.Background(), m.delivery(t)))
 }
@@ -464,3 +518,132 @@ func TestCancelledAnalysisDoesNotHold(t *testing.T) {
 		})
 	}
 }
+
+func TestProcessRetriesPublicationAfterPersistence(t *testing.T) {
+	for _, status := range []entity.BuildStatus{entity.BuildStatusRunning, entity.BuildStatusFailed} {
+		t.Run(string(status), func(t *testing.T) {
+			c, m := newTestController(t)
+			issue := entity.BuildIssue{TaskID: "validation", UnitID: "unit"}
+			publishErr := errs.NewRetryableError(assert.AnError)
+			var failedMessageID string
+
+			detected := m.expectPass(status, []entity.BuildIssue{issue}, nil)
+			persisted := m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected)
+			m.publisher.EXPECT().Publish(gomock.Any(), "issue_track", gomock.Any()).
+				DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error {
+					failedMessageID = msg.ID
+					return publishErr
+				}).After(persisted)
+
+			require.ErrorIs(t, c.Process(context.Background(), m.delivery(t)), publishErr)
+
+			detected = m.expectPass(status, []entity.BuildIssue{issue}, nil)
+			persisted = m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).
+				Return(storage.ErrAlreadyExists).After(detected)
+			published := m.expectNotification(t, status.IsTerminal(), persisted).
+				Do(func(_ context.Context, _ string, msg entityqueue.Message) {
+					if status.IsTerminal() {
+						assert.Equal(t, failedMessageID, msg.ID)
+					} else {
+						assert.NotEqual(t, failedMessageID, msg.ID)
+					}
+				})
+			d := m.delivery(t)
+			if !status.IsTerminal() {
+				d.EXPECT().Hold(PollDelayMs).After(published)
+			}
+
+			require.NoError(t, c.Process(context.Background(), d))
+		})
+	}
+}
+
+func TestProcessPublishesDistinctRefreshesFromHeldMessage(t *testing.T) {
+	c, m := newTestController(t)
+	var messageIDs []string
+
+	for range 2 {
+		detected := m.expectPass(entity.BuildStatusRunning, nil, nil)
+		published := m.expectNotification(t, false, detected).
+			Do(func(_ context.Context, _ string, msg entityqueue.Message) {
+				messageIDs = append(messageIDs, msg.ID)
+			})
+		d := m.delivery(t)
+		d.EXPECT().Hold(PollDelayMs).After(published)
+
+		require.NoError(t, c.Process(context.Background(), d))
+	}
+
+	require.Len(t, messageIDs, 2)
+	assert.NotEqual(t, messageIDs[0], messageIDs[1])
+}
+
+func TestProcessWaitsForWinningBuildBeforeFinalizing(t *testing.T) {
+	for _, outcome := range []string{"pending", "recorded during detection", "abandoned", "another build wins"} {
+		t.Run(outcome, func(t *testing.T) {
+			c, m := newTestController(t)
+			detected := m.expectPass(entity.BuildStatusFailed, nil, nil)
+			m.request.TerminalBuildID = ""
+			if outcome == "abandoned" {
+				m.request.State = entity.RequestStateFailed
+			}
+			detected.Do(func(context.Context, entity.BuildID) {
+				switch outcome {
+				case "recorded during detection":
+					m.request.TerminalBuildID = testBuildID
+					m.request.State = entity.RequestStateFailed
+				case "another build wins":
+					m.request.TerminalBuildID = "other-build"
+					m.request.State = entity.RequestStateFailed
+				}
+			})
+
+			d := m.delivery(t)
+			if outcome != "another build wins" {
+				published := m.expectNotification(t, outcome == "recorded during detection", detected)
+				if outcome == "pending" {
+					d.EXPECT().Hold(PollDelayMs).After(published)
+				}
+			}
+
+			require.NoError(t, c.Process(context.Background(), d))
+
+			if outcome == "pending" {
+				next := m.expectPass(entity.BuildStatusFailed, nil, nil)
+				m.request.State = entity.RequestStateFailed
+				m.expectNotification(t, true, next)
+
+				require.NoError(t, c.Process(context.Background(), m.delivery(t)))
+			}
+		})
+	}
+}
+
+func TestProcessIgnoresKnownLosingBuildBeforeDetection(t *testing.T) {
+	c, m := newTestController(t)
+	m.request.TerminalBuildID = "other-build"
+	m.stores.EXPECT().For(storage.Config{QueueName: testQueue}).Return(m.store, nil)
+	m.builds.EXPECT().Get(gomock.Any(), testBuildID).
+		Return(entity.Build{ID: testBuildID, RequestID: m.request.ID, Status: entity.BuildStatusFailed}, nil)
+
+	require.NoError(t, c.Process(context.Background(), m.delivery(t)))
+}
+
+func TestProcessReturnsOutcomeReloadErrorsWithoutPublishing(t *testing.T) {
+	c, m := newTestController(t)
+	m.requestRead.Times(1)
+	detected := m.expectPass(entity.BuildStatusFailed, nil, nil)
+	m.requests.EXPECT().Get(gomock.Any(), m.request.ID).Return(entity.Request{}, assert.AnError).After(detected)
+
+	require.ErrorIs(t, c.Process(context.Background(), m.delivery(t)), assert.AnError)
+}
+
+func TestProcessRejectsQueueMismatchOnOutcomeReload(t *testing.T) {
+	c, m := newTestController(t)
+	detected := m.expectPass(entity.BuildStatusFailed, nil, nil)
+	detected.Do(func(context.Context, entity.BuildID) {
+		m.request.Queue = "other/main"
+	})
+
+	require.Error(t, c.Process(context.Background(), m.delivery(t)))
+}
diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel
index 441031a49..663d00d00 100644
--- a/stovepipe/controller/issuetrack/BUILD.bazel
+++ b/stovepipe/controller/issuetrack/BUILD.bazel
@@ -6,8 +6,12 @@ go_library(
     importpath = "github.com/uber/submitqueue/stovepipe/controller/issuetrack",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/base/id:go_default_library",
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
+        "//platform/metrics:go_default_library",
         "//stovepipe/core/loader:go_default_library",
+        "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
@@ -18,11 +22,14 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "delivery_test.go",
         "fixture_test.go",
         "issuetrack_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer/mock:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md
index 3af6e37f9..e3668a164 100644
--- a/stovepipe/controller/issuetrack/README.md
+++ b/stovepipe/controller/issuetrack/README.md
@@ -1,6 +1,6 @@
 # Issue tracking
 
-Issue tracking maintains current issue state and opening/closing history from persisted build issues. This layer applies one issue at a time; notification handling and bounded passes follow separately. No consumer is registered yet.
+Issue tracking maintains current issue state and opening/closing history from persisted build issues. Each delivery applies a bounded batch of issues and requests immediate redelivery when more work remains. No consumer is registered yet.
 
 ## Applying one issue
 
@@ -14,6 +14,36 @@ Closed issue rows are retained, so their versions cannot reset on reopening. A r
 
 ## Ordering requirements
 
-The caller must finish the current cursor and its history before moving on or admitting another build. If a replacement worker advances, completing that cursor changes the issue version first. A paused worker's earlier issue write then fails its row guard. A worker that reads later is rejected by source ordering or the conditional cursor write.
+During issue application, the caller must finish the current cursor and its history before moving on or admitting another build. If a replacement worker advances, completing that cursor changes the issue version first. A paused worker's earlier issue write then fails its row guard. A worker that reads later is rejected by source ordering or the conditional cursor write.
 
 The post-read cursor write and processed-build updates, including unchanged statuses, are required for this ordering. A progress version alone is not a lock around issue writes. These operations use individual K-V reads, conditional writes, and immutable history creates; no cross-record transaction or stored future-change payload is needed.
+
+## Refreshing and finalizing
+
+Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal notification retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh even when helping it through another notification. Encountering a task-wide issue (empty unit ID) sets `HasTaskWideIssues` in that same versioned record. This fact remains set across batches and retries, including when the issue was already applied by an earlier pass.
+
+Once the entire build-issue range is exhausted:
+
+- An ordinary refresh becomes `refreshed`.
+- A terminal pass with a task-wide failure, or a canceled build, becomes `finalized` without closing anything. Reported closed issues may still have reopened during refreshing; unreported closed issues remain closed.
+- Any other terminal pass becomes `finalizing`, resetting its cursor to the start of the issue-state range.
+
+The `finalizing` phase traverses retained issue states in primary-key order. Each key is looked up in the build's issues: a reported issue remains open, while an absent issue closes. Actual openings and closings are retained in history. Exhausting the issue-state range marks the pass `finalized`, so late refreshes are ignored.
+
+No closure occurs before all accumulated build issues have been processed. A task-wide failure in a later batch therefore protects existing issues across every task, not just keys encountered afterward. A later build makes its own decision; the restriction does not carry over. Both walks consume one ordered range at a time; no merge is required.
+
+## Bounded passes
+
+Every delivery processes at most 100 task/unit keys from one ordered range, including a revisited cursor and retained closed rows. Refreshing reads build issues; finalizing reads issue states. The cursor is always inclusive: revisit it to finish the issue and its history, then continue after it. A full batch retains its active phase and requests redelivery. Moving from refreshing to finalizing conditionally saves a reset cursor before any closing work begins.
+
+An unfinished pass completes before another source is admitted. A notification that helps finish a different pass requests redelivery before handling its own source. Completed notifications and older sources are ignored, as are refreshes after the same Request has finalized. A terminal pass requires a terminal build chosen by its Request in both active phases.
+
+Build IDs are opaque; source order comes from Request IDs. All scans are bounded reads under leading primary-key prefixes. There is no transaction spanning progress, issue state, and history.
+
+## Recovery and activation
+
+An error returns to normal consumer retry. The next delivery resumes the cursor before advancing. Permanently failing cursor work blocks issue tracking for that queue; terminal recovery and consumer registration remain follow-up work. This PR does not enable consumption or publish repair requests.
+
+## Analysis handoff
+
+Build analysis publishes refresh or final notifications after persisting build issues. Final notification requires successful terminal detection and the Request's recorded terminal build. Terminal detection returns the complete cumulative issue set, including issues returned by earlier calls, so finalization does not depend on earlier workers finishing their writes. Nonterminal detection may omit earlier issues without removing them from the build's accumulated results. The reference host registers the publication route but does not enable issue-tracking consumption or recovery.
diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go
new file mode 100644
index 000000000..337c362d5
--- /dev/null
+++ b/stovepipe/controller/issuetrack/delivery_test.go
@@ -0,0 +1,619 @@
+// Copyright (c) 2026 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 issuetrack
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	consumermock "github.com/uber/submitqueue/platform/consumer/mock"
+	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+func (f *fixture) message(build string, final bool) entityqueue.Message {
+	f.t.Helper()
+	f.sequence++
+	payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{BuildId: build, QueueName: testQueue, Finalize: final})
+	require.NoError(f.t, err)
+	msg := entityqueue.NewMessage(fmt.Sprintf("notice/%d", f.sequence), payload, testQueue, nil)
+	msg.Tenant = testQueue
+	return msg
+}
+
+func (f *fixture) deliver(msg entityqueue.Message, hold bool) error {
+	f.t.Helper()
+	delivery := consumermock.NewMockDelivery(f.mocks)
+	delivery.EXPECT().Message().Return(msg)
+	if hold {
+		delivery.EXPECT().Hold(int64(0))
+	}
+	return f.controller.Process(context.Background(), delivery)
+}
+
+func TestRefreshOpensReportedIssuesAndKeepsOtherState(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusRunning)
+	f.seedIssue("old", "", entity.IssueStatusOpen)
+	f.reportIssue("build", "task", "unit")
+	msg := f.message("build", false)
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase)
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", ""}].Status)
+	require.Equal(t, int32(7), f.states[issueKey{"old", ""}].Version)
+	require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version)
+	require.Len(t, f.logs, 1)
+
+	f.reportIssue("build", "task", "next")
+	require.NoError(t, f.deliver(msg, false))
+	require.NotContains(t, f.states, issueKey{"task", "next"})
+	require.NoError(t, f.deliver(f.message("build", false), false))
+	require.Contains(t, f.states, issueKey{"task", "next"})
+	require.Len(t, f.logs, 2)
+	require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version)
+}
+
+func TestFinalizationOpensKeepsAndClosesIssues(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusFailed)
+	f.seedIssue("a", "", entity.IssueStatusOpen)
+	f.seedIssue("b", "unit", entity.IssueStatusOpen)
+	f.seedIssue("closed", "", entity.IssueStatusClosed)
+	f.reportIssue("build", "b", "unit")
+	f.reportIssue("build", "c", "unit")
+	msg := f.message("build", true)
+	require.NoError(t, f.deliver(msg, true))
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"a", ""}].Status)
+	require.Equal(t, int32(8), f.states[issueKey{"a", ""}].Version)
+	require.Equal(t, "older", f.states[issueKey{"b", "unit"}].FirstSeenBuildID)
+	require.Equal(t, int32(8), f.states[issueKey{"b", "unit"}].Version)
+	require.Equal(t, "build", f.states[issueKey{"c", "unit"}].FirstSeenBuildID)
+	require.Equal(t, int32(8), f.states[issueKey{"closed", ""}].Version)
+	require.Len(t, f.logs, 2)
+	require.Contains(t, f.logs, logKey{issueKey{"a", ""}, "build", entity.IssueEventResolved})
+	require.Contains(t, f.logs, logKey{issueKey{"c", "unit"}, "build", entity.IssueEventOpened})
+}
+
+func TestEmptyFinalBuildClosesIssuesButRetainsTheirRows(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusSucceeded)
+	f.seedIssue("task", "", entity.IssueStatusOpen)
+	msg := f.message("build", true)
+	require.NoError(t, f.deliver(msg, true))
+	require.NoError(t, f.deliver(msg, false))
+	require.Len(t, f.states, 1)
+	require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", ""}].Status)
+	require.Len(t, f.logs, 1)
+}
+
+func TestReportedIssueReopensTheSameRecord(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusRunning)
+	f.seedIssue("task", "unit", entity.IssueStatusClosed)
+	f.reportIssue("build", "task", "unit")
+	require.NoError(t, f.deliver(f.message("build", false), false))
+	issue := f.states[issueKey{"task", "unit"}]
+	require.Equal(t, entity.IssueStatusOpen, issue.Status)
+	require.Equal(t, int32(8), issue.Version)
+	require.Equal(t, "build", issue.FirstSeenBuildID)
+}
+
+func TestClosedRecordsStillAdvanceTheCursor(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusSucceeded)
+	for i := 0; i < 101; i++ {
+		f.seedIssue("task", fmt.Sprintf("unit/%03d", i), entity.IssueStatusClosed)
+	}
+	msg := f.message("build", true)
+	require.NoError(t, f.deliver(msg, true))
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, "unit/099", f.progress.CursorUnitID)
+	require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase)
+	require.Empty(t, f.logs)
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, "unit/100", f.progress.CursorUnitID)
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.Empty(t, f.logs)
+}
+
+func TestRefreshCheckpointsAtTheBatchBoundary(t *testing.T) {
+	for _, count := range []int{0, 99, 100, 103} {
+		t.Run(fmt.Sprint(count), func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", entity.BuildStatusRunning)
+			for i := 0; i < count; i++ {
+				f.reportIssue("build", "task", fmt.Sprintf("unit/%03d", i))
+			}
+			msg := f.message("build", false)
+			require.NoError(t, f.deliver(msg, count >= 100))
+			require.Len(t, f.states, min(count, 100))
+			if count >= 100 {
+				require.Equal(t, "unit/099", f.progress.CursorUnitID)
+				require.NoError(t, f.deliver(msg, false))
+			}
+			require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase)
+			require.Len(t, f.states, count)
+			require.Len(t, f.logs, count)
+		})
+	}
+}
+
+func TestFinalizationTraversesBuildIssuesThenExistingStates(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusFailed)
+	for i := 0; i < 99; i++ {
+		f.reportIssue("build", "task", fmt.Sprintf("unit/%03d", 2*i))
+		f.seedIssue("task", fmt.Sprintf("unit/%03d", 2*i+1), entity.IssueStatusOpen)
+	}
+	msg := f.message("build", true)
+	require.NoError(t, f.deliver(msg, true))
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, "unit/099", f.progress.CursorUnitID)
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, "unit/197", f.progress.CursorUnitID)
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	open, closed := 0, 0
+	for _, issue := range f.states {
+		if issue.Status == entity.IssueStatusOpen {
+			open++
+		} else {
+			closed++
+		}
+	}
+	require.Equal(t, 99, open)
+	require.Equal(t, 99, closed)
+	require.Len(t, f.logs, 198)
+}
+
+func TestCanceledBuildKeepsExistingIssuesOpen(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusCancelled)
+	f.seedIssue("old", "", entity.IssueStatusOpen)
+	f.reportIssue("build", "new", "")
+	f.failures["issue/list"] = 1
+	require.NoError(t, f.deliver(f.message("build", true), false))
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", ""}].Status)
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"new", ""}].Status)
+	require.Len(t, f.logs, 1)
+}
+
+func TestFinalNotificationBeforeLateRefreshAndOlderBuild(t *testing.T) {
+	f := newFixture(t)
+	f.source("older-build", "9", entity.BuildStatusSucceeded)
+	f.source("newer-build", "10", entity.BuildStatusSucceeded)
+	msg := f.message("newer-build", true)
+	require.NoError(t, f.deliver(msg, true))
+	require.NoError(t, f.deliver(msg, false))
+	before := f.progress
+	f.reportIssue("newer-build", "late", "")
+	f.reportIssue("older-build", "old", "")
+	require.NoError(t, f.deliver(f.message("newer-build", false), false))
+	require.NoError(t, f.deliver(f.message("older-build", true), false))
+	require.Equal(t, before, f.progress)
+	require.Empty(t, f.states)
+}
+
+func TestAnotherNotificationFinishesTheCursorBeforeItsOwnBuild(t *testing.T) {
+	f := newFixture(t)
+	f.source("build/100", "100", entity.BuildStatusRunning)
+	f.source("build/101", "101", entity.BuildStatusSucceeded)
+	f.reportIssue("build/100", "task", "unit")
+	f.retain(currentCursor())
+	incoming := f.message("build/101", true)
+	require.NoError(t, f.deliver(incoming, true))
+	require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase)
+	require.Equal(t, "build/100", f.progress.BuildID)
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status)
+	require.NoError(t, f.deliver(incoming, true))
+	require.NoError(t, f.deliver(incoming, false))
+	require.Equal(t, "build/101", f.progress.BuildID)
+	require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status)
+	require.Equal(t, int32(2), f.states[issueKey{"task", "unit"}].Version)
+}
+
+func TestIssueMutationWaitsForItsCursor(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusRunning)
+	f.reportIssue("build", "task", "unit")
+	f.failures["progress/update"] = 1
+	msg := f.message("build", false)
+	require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+	require.Empty(t, f.states)
+	require.Empty(t, f.logs)
+	require.NoError(t, f.deliver(msg, false))
+	require.Len(t, f.states, 1)
+	require.Len(t, f.logs, 1)
+}
+
+func TestRetryRepairsHistoryBeforeFinalizing(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusFailed)
+	f.reportIssue("build", "task", "unit")
+	f.failures["history/create"] = 1
+	msg := f.message("build", true)
+	require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+	require.Equal(t, "task", f.progress.CursorTaskID)
+	require.Equal(t, "unit", f.progress.CursorUnitID)
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase)
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.Equal(t, "task", f.progress.CursorTaskID)
+	require.Equal(t, "unit", f.progress.CursorUnitID)
+	require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version)
+	require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build", entity.IssueEventOpened}].TimestampMs)
+}
+
+func TestRetryFinishesTheCursorIssueBeforeScanningTheNextBatch(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusRunning)
+	for i := 0; i < 101; i++ {
+		f.reportIssue("build", "task", fmt.Sprintf("unit/%03d", i))
+	}
+	f.failures["history/create"] = 1
+	msg := f.message("build", false)
+
+	require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+	require.Equal(t, "unit/000", f.progress.CursorUnitID)
+	require.Len(t, f.states, 1)
+	require.Empty(t, f.logs)
+
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, "unit/099", f.progress.CursorUnitID)
+	require.Len(t, f.states, 100)
+	require.Len(t, f.logs, 100)
+	require.Equal(t, int32(1), f.states[issueKey{"task", "unit/000"}].Version)
+
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, "unit/100", f.progress.CursorUnitID)
+	require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase)
+	require.Len(t, f.states, 101)
+	require.Len(t, f.logs, 101)
+}
+
+func TestReadFailuresDoNotChangeIssues(t *testing.T) {
+	for _, operation := range []string{"build-issues/list", "issue/list", "build-issues/get", "issue/get"} {
+		t.Run(operation, func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", entity.BuildStatusSucceeded)
+			f.seedIssue("task", "unit", entity.IssueStatusOpen)
+			msg := f.message("build", true)
+			if operation != "build-issues/list" {
+				require.NoError(t, f.deliver(msg, true))
+			}
+			f.failures[operation] = 1
+			require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+			require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status)
+			require.Empty(t, f.logs)
+		})
+	}
+}
+
+func TestFinalizationRequiresTheWinningTerminalBuild(t *testing.T) {
+	for _, status := range []entity.BuildStatus{entity.BuildStatusRunning, entity.BuildStatusSucceeded} {
+		t.Run(string(status), func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", status)
+			request := f.requests["10"]
+			request.TerminalBuildID = ""
+			f.requests["10"] = request
+			require.Error(t, f.deliver(f.message("build", true), false))
+			require.False(t, f.hasProgress)
+			require.Empty(t, f.states)
+		})
+	}
+	f := newFixture(t)
+	f.source("loser", "10", entity.BuildStatusFailed)
+	request := f.requests["10"]
+	request.TerminalBuildID = "winner"
+	f.requests["10"] = request
+	require.NoError(t, f.deliver(f.message("loser", true), false))
+	require.False(t, f.hasProgress)
+}
+
+func TestProcessRejectsInvalidMessageAndSource(t *testing.T) {
+	for _, kind := range []string{"payload", "partition", "tenant", "missing message ID", "missing build", "request queue", "invalid request ID"} {
+		t.Run(kind, func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", entity.BuildStatusRunning)
+			msg := f.message("build", false)
+			switch kind {
+			case "payload":
+				msg.Payload = []byte("invalid")
+			case "partition":
+				msg.PartitionKey = "other"
+			case "tenant":
+				msg.Tenant = "other"
+			case "missing message ID":
+				msg.ID = ""
+			case "missing build":
+				delete(f.builds, "build")
+			case "request queue":
+				request := f.requests["10"]
+				request.Queue = "other"
+				f.requests["10"] = request
+			case "invalid request ID":
+				request := f.requests["10"]
+				request.ID = "010"
+				f.requests["10"] = request
+			}
+			require.Error(t, f.deliver(msg, false))
+			require.False(t, f.hasProgress)
+			require.Empty(t, f.states)
+		})
+	}
+}
+
+func TestCursorRetriesPartialWrites(t *testing.T) {
+	for _, operation := range []string{"issue/create", "issue/update", "history/create", "complete-pass"} {
+		t.Run(operation, func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", entity.BuildStatusRunning)
+			f.reportIssue("build", "task", "unit")
+			if operation == "issue/update" {
+				f.seedIssue("task", "unit", entity.IssueStatusClosed)
+			}
+			if operation == "complete-pass" {
+				f.failOnPhase = entity.IssueTrackingPhaseRefreshed
+			} else {
+				f.failures[operation] = 1
+			}
+			msg := f.message("build", false)
+			require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+			require.Equal(t, "task", f.progress.CursorTaskID)
+			require.Equal(t, "unit", f.progress.CursorUnitID)
+			require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
+			require.NoError(t, f.deliver(msg, false))
+			require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase)
+			require.Equal(t, "build", f.states[issueKey{"task", "unit"}].LastProcessedBuildID)
+			require.Equal(t, "build", f.states[issueKey{"task", "unit"}].LastChangedBuildID)
+			require.Len(t, f.logs, 1)
+			if operation == "issue/update" {
+				require.Equal(t, int32(8), f.states[issueKey{"task", "unit"}].Version)
+			} else {
+				require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version)
+			}
+		})
+	}
+}
+
+func TestHistoryRecordsTheActualChangingBuildNotTheLatestProcessedBuild(t *testing.T) {
+	f := newFixture(t)
+	f.seedIssue("task", "unit", entity.IssueStatusOpen)
+	f.source("build/100", "100", entity.BuildStatusSucceeded)
+	f.failures["history/create"] = 1
+	msg := f.message("build/100", true)
+	require.NoError(t, f.deliver(msg, true))
+	require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+	changed := f.states[issueKey{"task", "unit"}]
+	require.Equal(t, entity.IssueStatusClosed, changed.Status)
+	require.Equal(t, "build/100", changed.LastChangedBuildID)
+	require.Equal(t, "build/100", changed.LastProcessedBuildID)
+	require.Equal(t, "older", changed.FirstSeenBuildID)
+	require.Empty(t, f.logs)
+	f.controller.nowMs = func() int64 { return 2000 }
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", "unit"}, "build/100", entity.IssueEventResolved}].TimestampMs)
+	require.Equal(t, changed, f.states[issueKey{"task", "unit"}])
+
+	f.source("build/101", "101", entity.BuildStatusSucceeded)
+	next := f.message("build/101", true)
+	require.NoError(t, f.deliver(next, true))
+	require.NoError(t, f.deliver(next, false))
+	retained := f.states[issueKey{"task", "unit"}]
+	require.Equal(t, "build/101", retained.LastProcessedBuildID)
+	require.Equal(t, "build/100", retained.LastChangedBuildID)
+	require.Equal(t, int64(1000), retained.ChangedAtMs)
+	require.Equal(t, int32(9), retained.Version)
+	require.Len(t, f.logs, 1)
+}
+
+func TestTaskWideFailurePreservesAllExistingIssues(t *testing.T) {
+	f := newFixture(t)
+	f.source("build/100", "100", entity.BuildStatusFailed)
+	f.seedIssue("old", "unit", entity.IssueStatusOpen)
+	f.seedIssue("old", "", entity.IssueStatusOpen)
+	f.seedIssue("old", "closed", entity.IssueStatusClosed)
+	f.seedIssue("reopen", "unit", entity.IssueStatusClosed)
+	f.reportIssue("build/100", "new", "unit")
+	f.reportIssue("build/100", "reopen", "unit")
+	f.reportIssue("build/100", "unrelated", "")
+	f.failures["issue/list"] = 1
+	msg := f.message("build/100", true)
+
+	require.NoError(t, f.deliver(msg, false))
+	require.True(t, f.progress.Finalize)
+	require.True(t, f.progress.HasTaskWideIssues)
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	for _, key := range []issueKey{{"old", "unit"}, {"old", ""}, {"new", "unit"}, {"reopen", "unit"}, {"unrelated", ""}} {
+		require.Equal(t, entity.IssueStatusOpen, f.states[key].Status)
+	}
+	require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"old", "closed"}].Status)
+	require.Len(t, f.logs, 3)
+	for key := range f.logs {
+		require.Equal(t, entity.IssueEventOpened, key.event)
+	}
+	require.Equal(t, 1, f.failures["issue/list"])
+
+	f.failures["issue/list"] = 0
+	f.source("build/101", "101", entity.BuildStatusSucceeded)
+	next := f.message("build/101", true)
+	require.NoError(t, f.deliver(next, true))
+	require.False(t, f.progress.HasTaskWideIssues)
+	require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase)
+	require.NoError(t, f.deliver(next, false))
+	for _, issue := range f.states {
+		require.Equal(t, entity.IssueStatusClosed, issue.Status)
+	}
+}
+
+func TestTaskWideFailureInLaterOpeningBatchPreventsAllClosures(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusFailed)
+	f.seedIssue("old", "unit", entity.IssueStatusOpen)
+	for i := 0; i < 105; i++ {
+		f.reportIssue("build", fmt.Sprintf("task/%03d", i), "unit")
+	}
+	f.reportIssue("build", "zzz", "")
+	msg := f.message("build", true)
+
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
+	require.True(t, f.progress.Finalize)
+	require.False(t, f.progress.HasTaskWideIssues)
+	require.Equal(t, "task/099", f.progress.CursorTaskID)
+	require.Len(t, f.logs, 100)
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status)
+
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.True(t, f.progress.HasTaskWideIssues)
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status)
+	require.Len(t, f.logs, 106)
+	for key := range f.logs {
+		require.Equal(t, entity.IssueEventOpened, key.event)
+	}
+}
+
+func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) {
+	f := newFixture(t)
+	f.source("build/100", "100", entity.BuildStatusFailed)
+	for i := 0; i < 104; i++ {
+		unit := fmt.Sprintf("unit/%03d", i)
+		if i == 0 {
+			unit = ""
+		}
+		f.reportIssue("build/100", "task", unit)
+		f.seedIssue("task", unit, entity.IssueStatusOpen)
+		key := issueKey{"task", unit}
+		issue := f.states[key]
+		issue.LastProcessedBuildID = "build/100"
+		f.states[key] = issue
+	}
+	prior := currentCursor()
+	prior.Phase = entity.IssueTrackingPhaseRefreshed
+	f.retain(prior)
+	msg := f.message("build/100", true)
+
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
+	require.True(t, f.progress.Finalize)
+	require.True(t, f.progress.HasTaskWideIssues)
+	require.Equal(t, "unit/099", f.progress.CursorUnitID)
+	require.Empty(t, f.logs)
+	f.failures["build-issues/list"] = 1
+	require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+	require.True(t, f.progress.HasTaskWideIssues)
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.True(t, f.progress.HasTaskWideIssues)
+	for _, issue := range f.states {
+		require.Equal(t, int32(7), issue.Version)
+	}
+}
+
+func TestAnotherNotificationRetainsTheSavedTerminalIntent(t *testing.T) {
+	f := newFixture(t)
+	f.source("build/100", "100", entity.BuildStatusFailed)
+	f.source("build/101", "101", entity.BuildStatusRunning)
+	f.reportIssue("build/100", "task/a", "unit")
+	f.reportIssue("build/100", "task/b", "unit")
+	progress := currentCursor()
+	progress.CursorTaskID, progress.CursorUnitID = "task/a", "unit"
+	progress.Finalize = true
+	f.retain(progress)
+	msg := f.message("build/101", false)
+
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, "build/100", f.progress.BuildID)
+	require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase)
+	require.True(t, f.progress.Finalize)
+	require.Empty(t, f.progress.CursorTaskID)
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, "build/101", f.progress.BuildID)
+	require.False(t, f.progress.Finalize)
+	require.False(t, f.progress.HasTaskWideIssues)
+	require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase)
+}
+
+func TestOpeningCursorCannotCloseAnUnreportedIssue(t *testing.T) {
+	f := newFixture(t)
+	f.source("build/100", "100", entity.BuildStatusFailed)
+	f.seedIssue("task", "unit", entity.IssueStatusOpen)
+	progress := currentCursor()
+	progress.Finalize = true
+	progress.HasTaskWideIssues = true
+	f.retain(progress)
+	msg := f.message("build/100", true)
+	msg.ID = progress.NotificationID
+
+	require.Error(t, f.deliver(msg, false))
+	issue := f.states[issueKey{"task", "unit"}]
+	require.Equal(t, entity.IssueStatusOpen, issue.Status)
+	require.Equal(t, int32(7), issue.Version)
+	require.Empty(t, f.logs)
+}
+
+func TestTaskWideHistoryRetryFinishesWithoutClosing(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusFailed)
+	f.seedIssue("old", "unit", entity.IssueStatusOpen)
+	f.reportIssue("build", "task", "")
+	f.failures["history/create"] = 1
+	f.failures["issue/list"] = 1
+	msg := f.message("build", true)
+
+	require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+	require.True(t, f.progress.HasTaskWideIssues)
+	require.True(t, f.progress.Finalize)
+	require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
+	f.controller.nowMs = func() int64 { return 2000 }
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status)
+	require.Equal(t, int64(1000), f.logs[logKey{issueKey{"task", ""}, "build", entity.IssueEventOpened}].TimestampMs)
+	require.Equal(t, 1, f.failures["issue/list"])
+}
+
+func TestRetryResetsCursorBeforeClosing(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusFailed)
+	f.seedIssue("old", "unit", entity.IssueStatusOpen)
+	f.reportIssue("build", "task", "unit")
+	f.failOnPhase = entity.IssueTrackingPhaseFinalizing
+	msg := f.message("build", true)
+
+	require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+	require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
+	require.Equal(t, "task", f.progress.CursorTaskID)
+	require.True(t, f.progress.Finalize)
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"old", "unit"}].Status)
+	require.NoError(t, f.deliver(msg, true))
+	require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase)
+	require.Empty(t, f.progress.CursorTaskID)
+	require.NoError(t, f.deliver(msg, false))
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"old", "unit"}].Status)
+	require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version)
+	require.Len(t, f.logs, 2)
+}
diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go
index cd24c8f63..6db9a8c3c 100644
--- a/stovepipe/controller/issuetrack/fixture_test.go
+++ b/stovepipe/controller/issuetrack/fixture_test.go
@@ -39,6 +39,7 @@ type logKey struct {
 }
 
 type fixture struct {
+	failOnPhase entity.IssueTrackingPhase
 	t           *testing.T
 	mocks       *gomock.Controller
 	controller  *Controller
@@ -146,6 +147,10 @@ func newFixture(t *testing.T) *fixture {
 		if err := f.fail("progress/update"); err != nil {
 			return err
 		}
+		if f.failOnPhase != "" && state.Phase == f.failOnPhase {
+			f.failOnPhase = ""
+			return assert.AnError
+		}
 		if !f.hasProgress || f.progress.Version != oldVersion {
 			return storage.ErrVersionMismatch
 		}
diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go
index 72cf7b0b9..8902fdc2f 100644
--- a/stovepipe/controller/issuetrack/issuetrack.go
+++ b/stovepipe/controller/issuetrack/issuetrack.go
@@ -23,13 +23,20 @@ import (
 	"time"
 
 	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/base/id"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/metrics"
 	"github.com/uber/submitqueue/stovepipe/core/loader"
+	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	"go.uber.org/zap"
 )
 
+// Each delivery processes at most this many issue keys, including a revisited cursor and closed records.
+const _issueBatchSize = 100
+
 type issueKey struct{ task, unit string }
 
 // Controller retains issue transitions and their queue-scoped progress.
@@ -53,6 +60,213 @@ func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.
 	}
 }
 
+func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error {
+	msg := delivery.Message()
+	notification := &stovepipemq.IssueTrack{}
+	if err := stovepipemq.Unmarshal(msg.Payload, notification); err != nil {
+		return fmt.Errorf("deserialize issue tracking notification: %w", err)
+	}
+	queue := notification.GetQueueName()
+	if err := entityqueue.ValidatePayloadQueue(msg, queue); err != nil {
+		return fmt.Errorf("invalid message identity: %w", err)
+	}
+	if msg.ID == "" || queue == "" || notification.GetBuildId() == "" {
+		return fmt.Errorf("issue tracking requires message, queue, and build identifiers")
+	}
+	if msg.PartitionKey != queue {
+		return fmt.Errorf("issue tracking must be partitioned by queue %q", queue)
+	}
+	store, err := c.stores.For(storage.Config{QueueName: queue})
+	if err != nil {
+		return fmt.Errorf("resolve issue tracking storage: %w", err)
+	}
+	return c.trackBuild(ctx, delivery, store, notification, msg.ID)
+}
+
+func (c *Controller) trackBuild(ctx context.Context, delivery consumer.Delivery, store storage.Storage, notification *stovepipemq.IssueTrack, notificationID string) error {
+	queue := notification.GetQueueName()
+	build, request, err := loadBuildAndRequest(ctx, store, queue, notification.GetBuildId())
+	if err != nil {
+		return err
+	}
+
+	tracking := store.GetIssueTrackingStore()
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		progress, err := tracking.Get(ctx)
+		if errors.Is(err, storage.ErrNotFound) {
+			progress = entity.IssueTrackingState{}
+		} else if err != nil {
+			return fmt.Errorf("load issue tracking progress: %w", err)
+		}
+
+		if progress.Phase != entity.IssueTrackingPhaseRefreshing && progress.Phase != entity.IssueTrackingPhaseFinalizing {
+			if request.TerminalBuildID != "" && request.TerminalBuildID != build.ID {
+				return nil
+			}
+			if progress.Version != 0 {
+				if progress.NotificationID == notificationID {
+					return nil
+				}
+				order, err := entity.CompareRequestID(request.ID, progress.RequestID)
+				if err != nil {
+					return err
+				}
+				if order < 0 || (order == 0 && progress.Phase == entity.IssueTrackingPhaseFinalized) {
+					c.logger.Debugw("ignoring superseded issue tracking notification",
+						"queue", queue, "build_id", build.ID, "request_id", request.ID, "current_request_id", progress.RequestID)
+					return nil
+				}
+			}
+
+			if progress.Version == math.MaxInt32 {
+				return fmt.Errorf("issue tracking version exhausted")
+			}
+			next := entity.IssueTrackingState{
+				RequestID:      request.ID,
+				BuildID:        build.ID,
+				NotificationID: notificationID,
+				Phase:          entity.IssueTrackingPhaseRefreshing,
+				Finalize:       notification.GetFinalize(),
+				Version:        progress.Version + 1,
+			}
+			if next.Finalize {
+				if err := requireTerminalBuild(build, request); err != nil {
+					return err
+				}
+			}
+
+			if progress.Version == 0 {
+				err = tracking.Create(ctx, next)
+			} else {
+				err = tracking.Update(ctx, next, progress.Version, next.Version)
+			}
+			if errors.Is(err, storage.ErrAlreadyExists) || errors.Is(err, storage.ErrVersionMismatch) {
+				continue
+			}
+			if err != nil {
+				return fmt.Errorf("start issue tracking pass: %w", err)
+			}
+			progress = next
+		}
+
+		checkpoint, err := c.reconcileBatch(ctx, store, queue, progress)
+		if err != nil {
+			return err
+		}
+		if checkpoint.NotificationID != notificationID ||
+			checkpoint.Phase == entity.IssueTrackingPhaseRefreshing ||
+			checkpoint.Phase == entity.IssueTrackingPhaseFinalizing {
+			delivery.Hold(0)
+		}
+		return nil
+	}
+}
+
+func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) (checkpoint entity.IssueTrackingState, retErr error) {
+	op := metrics.Begin(c.metricsScope, "batch", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
+	defer func() { op.Complete(retErr) }()
+
+	if err := progress.Validate(); err != nil {
+		return entity.IssueTrackingState{}, fmt.Errorf("invalid issue tracking progress: %w", err)
+	}
+
+	limit := _issueBatchSize
+
+	build, request, err := loadBuildAndRequest(ctx, store, queue, progress.BuildID)
+	if err != nil {
+		return entity.IssueTrackingState{}, err
+	}
+	switch progress.Phase {
+	case entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseFinalizing:
+	default:
+		return entity.IssueTrackingState{}, fmt.Errorf("issue batch requires an active pass")
+	}
+	if request.ID != progress.RequestID {
+		return entity.IssueTrackingState{}, fmt.Errorf("issue batch does not match its source request")
+	}
+	finalizing := progress.Phase == entity.IssueTrackingPhaseFinalizing
+	if progress.Finalize {
+		if err := requireTerminalBuild(build, request); err != nil {
+			return entity.IssueTrackingState{}, err
+		}
+	}
+
+	if finalizing && build.Status == entity.BuildStatusCancelled {
+		return entity.IssueTrackingState{}, fmt.Errorf("canceled builds cannot close missing issues")
+	}
+
+	if progress.CursorTaskID != "" {
+		next, err := c.processIssue(ctx, store, progress, issueKey{progress.CursorTaskID, progress.CursorUnitID}, false)
+		if err != nil {
+			return entity.IssueTrackingState{}, err
+		}
+		progress = next
+		limit--
+	}
+
+	var keys []issueKey
+	if finalizing {
+		states, err := store.GetIssueStateStore().List(ctx, storage.IssueStateRange{
+			After: storage.IssueStateCursor{TaskID: progress.CursorTaskID, UnitID: progress.CursorUnitID},
+			Limit: limit,
+		})
+		if err != nil {
+			return entity.IssueTrackingState{}, fmt.Errorf("list issue states: %w", err)
+		}
+		for _, issue := range states {
+			keys = append(keys, issueKey{issue.TaskID, issue.UnitID})
+		}
+	} else {
+		issues, err := store.GetBuildIssueStore().List(ctx, build.ID, storage.BuildIssueRange{
+			After: storage.BuildIssueCursor{TaskID: progress.CursorTaskID, UnitID: progress.CursorUnitID},
+			Limit: limit,
+		})
+		if err != nil {
+			return entity.IssueTrackingState{}, fmt.Errorf("list build issues: %w", err)
+		}
+		for _, issue := range issues {
+			keys = append(keys, issueKey{issue.TaskID, issue.UnitID})
+		}
+	}
+
+	cursorNeedsCheckpoint := false
+	for _, key := range keys {
+		if err := ctx.Err(); err != nil {
+			return entity.IssueTrackingState{}, err
+		}
+		previousVersion := progress.Version
+		progress, err = c.processIssue(ctx, store, progress, key, !finalizing)
+		if err != nil {
+			return entity.IssueTrackingState{}, err
+		}
+		cursorNeedsCheckpoint = progress.Version == previousVersion
+	}
+
+	if len(keys) < limit {
+		next := progress
+		switch {
+		case finalizing:
+			next.Phase = entity.IssueTrackingPhaseFinalized
+		case !progress.Finalize:
+			next.Phase = entity.IssueTrackingPhaseRefreshed
+		case progress.HasTaskWideIssues || build.Status == entity.BuildStatusCancelled:
+			next.Phase = entity.IssueTrackingPhaseFinalized
+		default:
+			next.Phase = entity.IssueTrackingPhaseFinalizing
+			next.CursorTaskID, next.CursorUnitID = "", ""
+		}
+		return saveProgress(ctx, store.GetIssueTrackingStore(), next)
+	}
+	if cursorNeedsCheckpoint {
+		return saveProgress(ctx, store.GetIssueTrackingStore(), progress)
+	}
+
+	return progress, nil
+}
+
 func (c *Controller) processIssue(ctx context.Context, store storage.Storage, progress entity.IssueTrackingState, key issueKey, reported bool) (entity.IssueTrackingState, error) {
 	if err := ctx.Err(); err != nil {
 		return entity.IssueTrackingState{}, err
@@ -66,7 +280,10 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr
 	}
 
 	if !reported && progress.Phase != entity.IssueTrackingPhaseFinalizing {
-		return entity.IssueTrackingState{}, fmt.Errorf("refresh cursor requires a reported issue")
+		return entity.IssueTrackingState{}, fmt.Errorf("only finalizing passes can close missing issues")
+	}
+	if reported && key.unit == "" && progress.Phase == entity.IssueTrackingPhaseRefreshing {
+		progress.HasTaskWideIssues = true
 	}
 
 	desired := entity.IssueStatusClosed
@@ -182,6 +399,36 @@ func saveProgress(ctx context.Context, store storage.IssueTrackingStore, next en
 	return next, nil
 }
 
+func requireTerminalBuild(build entity.Build, request entity.Request) error {
+	if !build.Status.IsTerminal() {
+		return fmt.Errorf("final issue tracking requires a terminal build")
+	}
+	if request.TerminalBuildID != build.ID {
+		return fmt.Errorf("final issue tracking requires the request's terminal build")
+	}
+	return nil
+}
+
+func loadBuildAndRequest(ctx context.Context, store storage.Storage, queue, buildID string) (entity.Build, entity.Request, error) {
+	build, err := loader.ByID(ctx, buildID, store.GetBuildStore().Get, "build")
+	if err != nil {
+		return entity.Build{}, entity.Request{}, err
+	}
+	request, err := loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request")
+	if err != nil {
+		return entity.Build{}, entity.Request{}, err
+	}
+	if request.Queue != queue {
+		return entity.Build{}, entity.Request{}, fmt.Errorf("issue tracking request belongs to another queue")
+	}
+	if err := id.Validate(request.ID); err != nil {
+		return entity.Build{}, entity.Request{}, fmt.Errorf("invalid issue tracking request ID: %w", err)
+	}
+	return build, request, nil
+}
+
 func (c *Controller) Name() string                { return "issuetrack" }
 func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey }
 func (c *Controller) ConsumerGroup() string       { return c.consumerGroup }
+
+var _ consumer.Controller = (*Controller)(nil)
diff --git a/stovepipe/controller/issuetrack/issuetrack_test.go b/stovepipe/controller/issuetrack/issuetrack_test.go
index 656c0666b..9e104cb00 100644
--- a/stovepipe/controller/issuetrack/issuetrack_test.go
+++ b/stovepipe/controller/issuetrack/issuetrack_test.go
@@ -24,6 +24,43 @@ import (
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 )
 
+func TestBatchRejectsInvalidClosingPass(t *testing.T) {
+	for _, scenario := range []string{"missing terminal intent", "task-wide failure", "canceled build", "running build", "not terminal build"} {
+		t.Run(scenario, func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build/100", "100", entity.BuildStatusSucceeded)
+			f.seedIssue("task", "unit", entity.IssueStatusOpen)
+			original := f.states[issueKey{"task", "unit"}]
+			progress := currentCursor()
+			progress.Phase = entity.IssueTrackingPhaseFinalizing
+			progress.Finalize = true
+
+			switch scenario {
+			case "missing terminal intent":
+				progress.Finalize = false
+			case "task-wide failure":
+				progress.HasTaskWideIssues = true
+			case "canceled build":
+				f.source("build/100", "100", entity.BuildStatusCancelled)
+			case "running build":
+				f.source("build/100", "100", entity.BuildStatusRunning)
+			case "not terminal build":
+				request := f.requests["100"]
+				request.TerminalBuildID = "another-build"
+				f.requests["100"] = request
+			}
+			f.progress = progress
+			f.hasProgress = true
+
+			_, err := f.controller.reconcileBatch(context.Background(), f.store, testQueue, progress)
+			require.Error(t, err)
+			require.Equal(t, original, f.states[issueKey{"task", "unit"}])
+			require.Equal(t, progress, f.progress)
+			require.Empty(t, f.logs)
+		})
+	}
+}
+
 func TestProcessIssueOpensAndCloses(t *testing.T) {
 	for _, reported := range []bool{true, false} {
 		name := "close"
@@ -35,6 +72,7 @@ func TestProcessIssueOpensAndCloses(t *testing.T) {
 			f.source("build/100", "100", entity.BuildStatusSucceeded)
 			progress := currentCursor()
 			progress.Phase = entity.IssueTrackingPhaseFinalizing
+			progress.Finalize = true
 			progress.CursorTaskID, progress.CursorUnitID = "", ""
 			f.retain(progress)
 			if !reported {
diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go
index 544948c3b..664f83264 100644
--- a/stovepipe/entity/issue_tracking_state.go
+++ b/stovepipe/entity/issue_tracking_state.go
@@ -30,7 +30,7 @@ const (
 	IssueTrackingPhaseRefreshing IssueTrackingPhase = "refreshing"
 	// IssueTrackingPhaseRefreshed means a refresh pass completed; later refresh notifications remain eligible.
 	IssueTrackingPhaseRefreshed IssueTrackingPhase = "refreshed"
-	// IssueTrackingPhaseFinalizing means final opening and closing updates are underway.
+	// IssueTrackingPhaseFinalizing means existing issues are being reconciled after all reported issues were applied.
 	IssueTrackingPhaseFinalizing IssueTrackingPhase = "finalizing"
 	// IssueTrackingPhaseFinalized means final reconciliation completed; later refresh notifications are ignored.
 	IssueTrackingPhaseFinalized IssueTrackingPhase = "finalized"
@@ -45,10 +45,14 @@ type IssueTrackingState struct {
 	// NotificationID identifies the message that owns this reconciliation pass.
 	NotificationID string `json:"notification_id"`
 	// CursorTaskID identifies the current issue's task; active passes revisit it on continuation.
-	// Empty with an empty CursorUnitID starts a pass.
+	// Empty with an empty CursorUnitID starts the phase's range.
 	CursorTaskID string `json:"cursor_task_id"`
 	// CursorUnitID identifies the unit at the cursor; empty is task-wide.
 	CursorUnitID string `json:"cursor_unit_id"`
+	// Finalize identifies a terminal reconciliation pass rather than an ordinary refresh.
+	Finalize bool `json:"finalize"`
+	// HasTaskWideIssues records whether this pass has encountered a task-wide issue.
+	HasTaskWideIssues bool `json:"has_task_wide_issues"`
 	// Phase distinguishes running passes from completed refresh or final passes.
 	Phase IssueTrackingPhase `json:"phase"`
 	// Version is the positive optimistic-lock version of this state.
@@ -73,7 +77,22 @@ func (s IssueTrackingState) Validate() error {
 		return fmt.Errorf("issue tracking version must be positive")
 	}
 	switch s.Phase {
-	case IssueTrackingPhaseRefreshing, IssueTrackingPhaseFinalizing, IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalized:
+	case IssueTrackingPhaseRefreshing:
+	case IssueTrackingPhaseRefreshed:
+		if s.Finalize {
+			return fmt.Errorf("a terminal pass cannot finish as an ordinary refresh")
+		}
+	case IssueTrackingPhaseFinalizing:
+		if !s.Finalize {
+			return fmt.Errorf("finalizing requires terminal intent")
+		}
+		if s.HasTaskWideIssues {
+			return fmt.Errorf("task-wide failures prevent closing missing issues")
+		}
+	case IssueTrackingPhaseFinalized:
+		if !s.Finalize {
+			return fmt.Errorf("finalized requires terminal intent")
+		}
 	default:
 		return fmt.Errorf("invalid issue tracking phase %q", s.Phase)
 	}
diff --git a/stovepipe/entity/issue_tracking_state_test.go b/stovepipe/entity/issue_tracking_state_test.go
index 99f9d471b..58eb40f24 100644
--- a/stovepipe/entity/issue_tracking_state_test.go
+++ b/stovepipe/entity/issue_tracking_state_test.go
@@ -23,7 +23,7 @@ import (
 func TestIssueTrackingState(t *testing.T) {
 	for _, phase := range []IssueTrackingPhase{IssueTrackingPhaseRefreshing, IssueTrackingPhaseRefreshed, IssueTrackingPhaseFinalizing, IssueTrackingPhaseFinalized} {
 		t.Run(string(phase), func(t *testing.T) {
-			state := IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: phase, Version: 1}
+			state := IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: phase, Finalize: phase == IssueTrackingPhaseFinalizing || phase == IssueTrackingPhaseFinalized, Version: 1}
 			require.NoError(t, state.Validate())
 		})
 	}
@@ -36,6 +36,14 @@ func TestIssueTrackingState(t *testing.T) {
 		{"nonnumeric request", func(s *IssueTrackingState) { s.RequestID = "request/10" }},
 		{"missing notification", func(s *IssueTrackingState) { s.NotificationID = "" }},
 		{"unit cursor without task", func(s *IssueTrackingState) { s.CursorUnitID = "unit" }},
+		{"closing without terminal intent", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseFinalizing }},
+		{"finalized without terminal intent", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseFinalized }},
+		{"terminal pass marked refreshed", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseRefreshed; s.Finalize = true }},
+		{"closing with task-wide failure", func(s *IssueTrackingState) {
+			s.Phase = IssueTrackingPhaseFinalizing
+			s.Finalize = true
+			s.HasTaskWideIssues = true
+		}},
 		{"missing build", func(s *IssueTrackingState) { s.BuildID = "" }},
 		{"unknown phase", func(s *IssueTrackingState) { s.Phase = IssueTrackingPhaseUnknown }},
 		{"unsupported phase", func(s *IssueTrackingState) { s.Phase = "other" }},
diff --git a/stovepipe/extension/buildissue/README.md b/stovepipe/extension/buildissue/README.md
index 691014210..7b47a88bb 100644
--- a/stovepipe/extension/buildissue/README.md
+++ b/stovepipe/extension/buildissue/README.md
@@ -1,12 +1,14 @@
 # Build issue detection
 
-`Detector` discovers confirmed validation issues from a build. It returns failure observations, not a complete view of build state.
+`Detector` discovers confirmed validation issues from a build. It returns build issues, not a complete view of build state.
 
 Issue values are defined in the [domain model](../../entity/build_issue.go). A validation task is a logical validation operation. A validation unit is an identifiable piece of work within a task.
 
-Within a validation queue, TaskID and UnitID together identify the validation condition. Repeated observations of the same condition use the same pair across builds. UnitID is empty for a task-wide issue.
+Within a validation queue, TaskID and UnitID together identify the validation condition. Repeated reports of the same condition use the same pair across builds. UnitID is empty for a task-wide issue.
 
-Each detection returns issues discoverable at that time and may repeat earlier observations. Detection does not mark issues as handled. An issue's absence from a detection result does not mean it is resolved.
+Nonterminal detection returns issues discoverable at that time and may repeat or omit previously reported issues. Reported issues remain part of that build's accumulated results; later omissions do not retract them. Detection does not mark issues as handled.
+
+A successful detection after build completion must return the complete cumulative issue set for the build within the queue's relevant validation scope, including issues returned by earlier calls. The result must be complete independently of whether earlier calls finished persisting their results. An empty terminal result is valid only if the build has no reportable issues. If the complete set cannot be determined, detection must return an error rather than report success. Implementations decide when a failure qualifies as a reportable issue.
 
 Detectors must support concurrent calls. Dependency failures return plain errors; issue results returned with an error must not be used.
 
diff --git a/stovepipe/extension/buildissue/buildissue.go b/stovepipe/extension/buildissue/buildissue.go
index c5fdcd0b9..f5dba03b6 100644
--- a/stovepipe/extension/buildissue/buildissue.go
+++ b/stovepipe/extension/buildissue/buildissue.go
@@ -24,10 +24,12 @@ import (
 	"github.com/uber/submitqueue/stovepipe/entity"
 )
 
-// Detector discovers confirmed issues without waiting for build completion.
-// Implementations own confirmation and retry rules and must be concurrency-safe.
-// Detection does not mark issues as handled.
-// An issue's absence from a detection result does not mean it is resolved.
+// Detector discovers issues retained in a build's accumulated results.
+// Nonterminal calls may repeat or omit issues returned earlier.
+// A successful terminal call must return the complete cumulative issue set
+// for the build, including issues returned by earlier calls.
+// Return an error if that set cannot be determined.
+// Implementations must be concurrency-safe and must not consume evidence.
 // Return plain dependency errors; issue results returned with an error must not be used.
 type Detector interface {
 	Detect(ctx context.Context, buildID entity.BuildID) ([]entity.BuildIssue, error)
diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md
index 6e672091d..4e0de28ea 100644
--- a/stovepipe/extension/storage/README.md
+++ b/stovepipe/extension/storage/README.md
@@ -62,11 +62,11 @@ History persistence does not alter issue state or validate transitions against i
 
 ## Issue reconciliation progress
 
-`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts a pass. An active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction.
+`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing, while `HasTaskWideIssues` records whether a task-wide issue has been encountered in that pass. Once all build issues have been applied, a terminal pass with task-wide failures finishes without closing; otherwise it resets the cursor and finalizes against issue states. These fields and the cursor are saved together under the same version guard. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction.
 
 The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of cursor work, and completion; no runtime writer is enabled here.
 
-The issue-tracking table must include the notification and cursor columns before updated readers or writers are deployed. No runtime writer is enabled by the storage layer. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores.
+The issue-tracking table must include the notification, cursor, `finalize`, and `has_task_wide_issues` columns before updated readers or writers are deployed. Both boolean columns default to false. They describe one pass, not analysis state for every build in the queue; terminal intent is required before absence-based reconciliation is permitted. No runtime writer is enabled by the storage layer. Notification IDs allow up to 1020 bytes, matching the queue message ID byte capacity; cursor limits match the issue stores.
 
 ### Applied transition facts
 
diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store.go b/stovepipe/extension/storage/mysql/issue_tracking_store.go
index 659a13bb9..9e81deeb9 100644
--- a/stovepipe/extension/storage/mysql/issue_tracking_store.go
+++ b/stovepipe/extension/storage/mysql/issue_tracking_store.go
@@ -43,13 +43,13 @@ func (s *issueTrackingStore) Get(ctx context.Context) (state entity.IssueTrackin
 	err := s.db.QueryRowContext(ctx,
 		`SELECT request_id, build_id, notification_id,
             cursor_task_id, cursor_unit_id,
-            phase, version
+            finalize, has_task_wide_issues, phase, version
          FROM issue_tracking WHERE queue = ?`, s.queue).
 		Scan(
 			&state.RequestID, &state.BuildID, &state.NotificationID,
 			&state.CursorTaskID, &state.CursorUnitID,
 
-			&state.Phase, &state.Version,
+			&state.Finalize, &state.HasTaskWideIssues, &state.Phase, &state.Version,
 		)
 	if errors.Is(err, sql.ErrNoRows) {
 		return entity.IssueTrackingState{}, storage.WrapNotFound(err)
@@ -69,10 +69,10 @@ func (s *issueTrackingStore) Create(ctx context.Context, state entity.IssueTrack
 	_, err := s.db.ExecContext(ctx,
 		`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id,
             cursor_task_id, cursor_unit_id,
-            phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+            finalize, has_task_wide_issues, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		s.queue, state.RequestID, state.BuildID, state.NotificationID,
 		state.CursorTaskID, state.CursorUnitID,
-		state.Phase, state.Version,
+		state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version,
 	)
 	if isDuplicateEntry(err) {
 		return storage.ErrAlreadyExists
@@ -97,11 +97,11 @@ func (s *issueTrackingStore) Update(ctx context.Context, state entity.IssueTrack
 	result, err := s.db.ExecContext(ctx,
 		`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?,
             cursor_task_id = ?, cursor_unit_id = ?,
-            phase = ?, version = ?
+            finalize = ?, has_task_wide_issues = ?, phase = ?, version = ?
          WHERE queue = ? AND version = ?`,
 		state.RequestID, state.BuildID, state.NotificationID,
 		state.CursorTaskID, state.CursorUnitID,
-		state.Phase, newVersion,
+		state.Finalize, state.HasTaskWideIssues, state.Phase, newVersion,
 		s.queue, oldVersion,
 	)
 	if err != nil {
diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go
index 8644d1569..935ea6682 100644
--- a/stovepipe/extension/storage/mysql/issue_tracking_store_test.go
+++ b/stovepipe/extension/storage/mysql/issue_tracking_store_test.go
@@ -31,7 +31,7 @@ import (
 )
 
 func trackingFixture() entity.IssueTrackingState {
-	return entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1}
+	return entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseFinalizing, Finalize: true, Version: 1}
 }
 
 func TestIssueTrackingStoreGet(t *testing.T) {
@@ -43,7 +43,7 @@ func TestIssueTrackingStoreGet(t *testing.T) {
 
 			state := trackingFixture()
 			store := NewIssueTrackingStore(db, testMetrics(), "queue")
-			query := mock.ExpectQuery(regexp.QuoteMeta(`SELECT request_id, build_id, notification_id, cursor_task_id, cursor_unit_id,  phase, version FROM issue_tracking WHERE queue = ?`)).WithArgs("queue")
+			query := mock.ExpectQuery(regexp.QuoteMeta(`SELECT request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, has_task_wide_issues, phase, version FROM issue_tracking WHERE queue = ?`)).WithArgs("queue")
 			switch kind {
 			case "missing":
 				query.WillReturnError(sql.ErrNoRows)
@@ -54,7 +54,7 @@ func TestIssueTrackingStoreGet(t *testing.T) {
 				if kind == "scan error" {
 					request = nil
 				}
-				query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "cursor_task_id", "cursor_unit_id", "phase", "version"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Phase, state.Version))
+				query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "cursor_task_id", "cursor_unit_id", "finalize", "has_task_wide_issues", "phase", "version"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version))
 			}
 
 			got, err := store.Get(context.Background())
@@ -94,8 +94,8 @@ func TestIssueTrackingStoreCreate(t *testing.T) {
 
 			state := trackingFixture()
 			store := NewIssueTrackingStore(db, testMetrics(), "queue")
-			query := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, cursor_task_id, cursor_unit_id,  phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`)).
-				WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Phase, state.Version)
+			query := mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO issue_tracking (queue, request_id, build_id, notification_id, cursor_task_id, cursor_unit_id, finalize, has_task_wide_issues, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)).
+				WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version)
 			if tt.dbErr != nil {
 				query.WillReturnError(tt.dbErr)
 			} else {
@@ -133,8 +133,8 @@ func TestIssueTrackingStoreUpdate(t *testing.T) {
 
 			state := trackingFixture()
 			store := NewIssueTrackingStore(db, testMetrics(), "queue")
-			query := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, cursor_task_id = ?, cursor_unit_id = ?,  phase = ?, version = ? WHERE queue = ? AND version = ?`)).
-				WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Phase, int32(7), "queue", int32(1))
+			query := mock.ExpectExec(regexp.QuoteMeta(`UPDATE issue_tracking SET request_id = ?, build_id = ?, notification_id = ?, cursor_task_id = ?, cursor_unit_id = ?, finalize = ?, has_task_wide_issues = ?, phase = ?, version = ? WHERE queue = ? AND version = ?`)).
+				WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, int32(7), "queue", int32(1))
 			switch {
 			case tt.dbErr != nil:
 				query.WillReturnError(tt.dbErr)
diff --git a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql
index e25f0a788..9d367d84c 100644
--- a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql
+++ b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql
@@ -1,14 +1,16 @@
 -- Persistent source-order boundary; unfinished passes are resumed before admitting another source.
 CREATE TABLE IF NOT EXISTS issue_tracking (
-    queue           VARBINARY(255)  NOT NULL,
-    request_id      VARBINARY(255)  NOT NULL,
-    build_id        VARBINARY(1020) NOT NULL,
-    notification_id VARBINARY(1020) NOT NULL,
+    queue                VARBINARY(255)  NOT NULL,
+    request_id           VARBINARY(255)  NOT NULL,
+    build_id             VARBINARY(1020) NOT NULL,
+    notification_id      VARBINARY(1020) NOT NULL,
 
-    cursor_task_id  VARBINARY(255)  NOT NULL,
-    cursor_unit_id  VARBINARY(1024) NOT NULL,
+    cursor_task_id       VARBINARY(255)  NOT NULL,
+    cursor_unit_id       VARBINARY(1024) NOT NULL,
 
-    phase           VARBINARY(32)   NOT NULL,
-    version         INT             NOT NULL,
+    finalize             BOOLEAN         NOT NULL DEFAULT FALSE,
+    has_task_wide_issues BOOLEAN         NOT NULL DEFAULT FALSE,
+    phase                VARBINARY(32)   NOT NULL,
+    version              INT             NOT NULL,
     PRIMARY KEY (queue)
 ) ENGINE=InnoDB;
diff --git a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go
index 83cd9b4a8..b8b63bd4f 100644
--- a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go
+++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go
@@ -49,6 +49,7 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F
 	require.Equal(t, state, got)
 
 	state.RequestID, state.BuildID, state.Phase = "11", "build/11", entity.IssueTrackingPhaseFinalizing
+	state.Finalize = true
 	require.NoError(t, store.Update(ctx, state, 2, 3))
 	state.Version = 3
 	state.Phase = entity.IssueTrackingPhaseFinalized
@@ -61,6 +62,24 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F
 	require.NoError(t, err)
 	require.Empty(t, opens)
 
+	t.Run("terminal opening checkpoint and task-wide flag", func(t *testing.T) {
+		bound, err := factory.For(storage.Config{QueueName: "tracking/terminal"})
+		require.NoError(t, err)
+		store := bound.GetIssueTrackingStore()
+		opening := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "terminal", Phase: entity.IssueTrackingPhaseRefreshing, Finalize: true, HasTaskWideIssues: true, CursorTaskID: "task", CursorUnitID: "unit", Version: 1}
+		require.NoError(t, store.Create(ctx, opening))
+		got, err := store.Get(ctx)
+		require.NoError(t, err)
+		require.Equal(t, opening, got)
+		finished := opening
+		finished.Phase = entity.IssueTrackingPhaseFinalized
+		require.NoError(t, store.Update(ctx, finished, 1, 2))
+		finished.Version = 2
+		got, err = store.Get(ctx)
+		require.NoError(t, err)
+		require.Equal(t, finished, got)
+	})
+
 	t.Run("queue isolation", func(t *testing.T) {
 		other, err := factory.For(storage.Config{QueueName: "Tracking/main"})
 		require.NoError(t, err)
@@ -79,6 +98,7 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F
 			next := state
 			next.RequestID = requestID
 			next.Phase = entity.IssueTrackingPhaseRefreshing
+			next.Finalize = false
 			go func() { results <- store.Update(ctx, next, 4, 5) }()
 		}
 		successes, conflicts := 0, 0