From 928133cfe40815429b9231c1a895dafa320b631e Mon Sep 17 00:00:00 2001
From: mnoah1 <mnoah1@uber.com>
Date: Tue, 27 Oct 2026 22:01:41 +0000
Subject: [PATCH 1/2] feat(stovepipe): hand off newly opened issues for repair

Summary:
Issue tracking now publishes a build-scoped repair request after completing an opening pass with newly opened or reopened issues. Persist the opening flag across batches and retries, and publish before advancing the pass so a failed handoff can be retried. The existing fix controller retains its one-repair-per-build policy.

Adds the fix publication route to the reference host; consumer activation and executor configuration remain separate. Existing MySQL deployments must append the has_new_issues column before deploying updated readers.

Test Plan:
- Run ./tool/bazel test //stovepipe/controller/issuetrack:go_default_test //stovepipe/controller/fix:go_default_test //stovepipe/core/fixrequest:go_default_test //stovepipe/entity:go_default_test //stovepipe/extension/storage/mysql:go_default_test //service/stovepipe/server:go_default_test.
- With Docker available, run ./tool/bazel test //test/integration/stovepipe/extension/storage/mysql:go_default_test --test_filter='TestMySQLStorage/IssueTracking' --test_output=errors to verify persistence of the opening flag against MySQL.

Revert Plan:
Revert this PR; the added MySQL column can remain.
---
 service/stovepipe/server/main.go              |   5 +
 service/stovepipe/server/main_test.go         |   8 +
 stovepipe/controller/issuetrack/BUILD.bazel   |   4 +
 stovepipe/controller/issuetrack/README.md     |   4 +-
 .../controller/issuetrack/fix_handoff_test.go | 187 ++++++++++++++++++
 .../controller/issuetrack/fixture_test.go     |  20 +-
 stovepipe/controller/issuetrack/issuetrack.go |  15 +-
 stovepipe/entity/issue_tracking_state.go      |   2 +
 stovepipe/extension/storage/README.md         |   2 +
 .../storage/mysql/issue_tracking_store.go     |  12 +-
 .../mysql/issue_tracking_store_test.go        |  14 +-
 .../storage/mysql/schema/issue_tracking.sql   |   1 +
 .../storage/mysql/issue_tracking_test.go      |   3 +-
 13 files changed, 260 insertions(+), 17 deletions(-)
 create mode 100644 stovepipe/controller/issuetrack/fix_handoff_test.go

diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go
index 6e3f98b1d..509122734 100644
--- a/service/stovepipe/server/main.go
+++ b/service/stovepipe/server/main.go
@@ -567,6 +567,11 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Name:  "issuetrack",
 			Queue: q,
 		},
+		{
+			Key:   stovepipemq.TopicKeyFix,
+			Name:  "fix",
+			Queue: q,
+		},
 		{
 			Key:          stovepipemq.TopicKeyBuildAnalysis,
 			Name:         "buildanalysis",
diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go
index 3f458b896..6dd54825e 100644
--- a/service/stovepipe/server/main_test.go
+++ b/service/stovepipe/server/main_test.go
@@ -101,6 +101,14 @@ func TestIssueTrackingHasPublicationRouteWithoutConsumption(t *testing.T) {
 	assert.NotContains(t, topicKeys(deadLetter), dlq.TopicKey(stovepipemq.TopicKeyIssueTrack))
 }
 
+func TestFixHasPublicationRouteWithoutConsumption(t *testing.T) {
+	registry, primary, _ := registeredControllers(t)
+	name, ok := registry.TopicName(stovepipemq.TopicKeyFix)
+	require.True(t, ok)
+	assert.Equal(t, "fix", name)
+	assert.NotContains(t, topicKeys(primary), stovepipemq.TopicKeyFix)
+}
+
 func TestHookStage(t *testing.T) {
 	registry, primary, deadLetter := registeredControllers(t)
 
diff --git a/stovepipe/controller/issuetrack/BUILD.bazel b/stovepipe/controller/issuetrack/BUILD.bazel
index 663d00d00..f8d6359fe 100644
--- a/stovepipe/controller/issuetrack/BUILD.bazel
+++ b/stovepipe/controller/issuetrack/BUILD.bazel
@@ -10,6 +10,7 @@ go_library(
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
+        "//stovepipe/core/fixrequest:go_default_library",
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
@@ -23,13 +24,16 @@ go_test(
     name = "go_default_test",
     srcs = [
         "delivery_test.go",
+        "fix_handoff_test.go",
         "fixture_test.go",
         "issuetrack_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
         "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
+        "//platform/extension/messagequeue/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 e3668a164..50681208c 100644
--- a/stovepipe/controller/issuetrack/README.md
+++ b/stovepipe/controller/issuetrack/README.md
@@ -22,6 +22,8 @@ The post-read cursor write and processed-build updates, including unchanged stat
 
 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.
 
+The progress record retains `HasNewIssues` when an issue is first opened or reopened by the source build, including when replaying that opening. Once the entire build-issue range is exhausted, a pass with new issues publishes a build-scoped repair request before saving its next phase. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. Refreshes for the same build share a repair message identity; the fix controller submits one repair per build.
+
 Once the entire build-issue range is exhausted:
 
 - An ordinary refresh becomes `refreshed`.
@@ -42,7 +44,7 @@ Build IDs are opaque; source order comes from Request IDs. All scans are bounded
 
 ## 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.
+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. The host must provide a fix publication route and a repair executor when registering the fix consumer.
 
 ## Analysis handoff
 
diff --git a/stovepipe/controller/issuetrack/fix_handoff_test.go b/stovepipe/controller/issuetrack/fix_handoff_test.go
new file mode 100644
index 000000000..3ee54b79f
--- /dev/null
+++ b/stovepipe/controller/issuetrack/fix_handoff_test.go
@@ -0,0 +1,187 @@
+// 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 (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
+	"github.com/uber/submitqueue/stovepipe/entity"
+)
+
+func TestOpeningPassPublishesOneBuildRepairRequest(t *testing.T) {
+	for _, tt := range []struct {
+		name   string
+		status entity.BuildStatus
+		final  bool
+	}{
+		{name: "running refresh", status: entity.BuildStatusRunning},
+		{name: "terminal opening pass", status: entity.BuildStatusFailed, final: true},
+		{name: "canceled build", status: entity.BuildStatusCancelled, final: true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", tt.status)
+			f.seedIssue("reopened", "unit", entity.IssueStatusClosed)
+			for _, task := range []string{"new", "reopened"} {
+				f.reportIssue("build", task, "unit")
+			}
+			f.onFixPublish = func(msg entityqueue.Message) {
+				require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
+				for _, task := range []string{"new", "reopened"} {
+					require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{task, "unit"}].Status)
+					require.Contains(t, f.logs, logKey{issueKey{task, "unit"}, "build", entity.IssueEventOpened})
+				}
+				payload := &stovepipemq.FixRequest{}
+				require.NoError(t, stovepipemq.Unmarshal(msg.Payload, payload))
+				require.Equal(t, "build", payload.GetId())
+				require.Equal(t, testQueue, payload.GetQueueName())
+				require.Equal(t, testQueue, msg.Tenant)
+				require.Equal(t, "build", msg.PartitionKey)
+			}
+			msg := f.message("build", tt.final)
+			closing := tt.final && tt.status != entity.BuildStatusCancelled
+			require.NoError(t, f.deliver(msg, closing))
+			require.Len(t, f.fixMessages, 1)
+			if closing {
+				require.NoError(t, f.deliver(msg, false))
+				require.Len(t, f.fixMessages, 1)
+			}
+			require.NoError(t, f.deliver(msg, false))
+			require.Len(t, f.fixMessages, 1)
+		})
+	}
+}
+
+func TestOpeningPassWithoutNewIssuesDoesNotPublishRepair(t *testing.T) {
+	for _, reportExisting := range []bool{false, true} {
+		t.Run(fmt.Sprint(reportExisting), func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", entity.BuildStatusRunning)
+			f.seedIssue("existing", "unit", entity.IssueStatusOpen)
+			if reportExisting {
+				f.reportIssue("build", "existing", "unit")
+			}
+			require.NoError(t, f.deliver(f.message("build", false), false))
+			require.Empty(t, f.fixMessages)
+			f.reportIssue("build", "new", "unit")
+			require.NoError(t, f.deliver(f.message("build", false), false))
+			require.Len(t, f.fixMessages, 1)
+		})
+	}
+}
+
+func TestRepairHandoffWaitsForAllOpeningBatches(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusRunning)
+	f.reportIssue("build", "a-new", "unit")
+	for i := 0; i < 100; i++ {
+		unit := fmt.Sprintf("unit/%03d", i)
+		f.seedIssue("existing", unit, entity.IssueStatusOpen)
+		f.reportIssue("build", "existing", unit)
+	}
+	msg := f.message("build", false)
+	require.NoError(t, f.deliver(msg, true))
+	require.Empty(t, f.fixMessages)
+	require.True(t, f.progress.HasNewIssues)
+
+	f.onFixPublish = func(entityqueue.Message) {
+		for _, issue := range f.states {
+			require.Equal(t, "build", issue.LastProcessedBuildID)
+		}
+	}
+	require.NoError(t, f.deliver(msg, false))
+	require.Len(t, f.fixMessages, 1)
+}
+
+func TestRepairHandoffRetriesBeforeCompletingPass(t *testing.T) {
+	for _, tt := range []struct {
+		name             string
+		failure          string
+		failOnPhase      entity.IssueTrackingPhase
+		initialPublishes int
+	}{
+		{name: "issue write", failure: "issue/create"},
+		{name: "opening history", failure: "history/create"},
+		{name: "repair publication", failure: "fix/publish", initialPublishes: 1},
+		{name: "completion checkpoint", failOnPhase: entity.IssueTrackingPhaseRefreshed, initialPublishes: 1},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", entity.BuildStatusRunning)
+			f.reportIssue("build", "new", "unit")
+			f.failures[tt.failure] = 1
+			f.failOnPhase = tt.failOnPhase
+			msg := f.message("build", false)
+			require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
+			require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase)
+			require.Len(t, f.fixMessages, tt.initialPublishes)
+
+			require.NoError(t, f.deliver(msg, false))
+			require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase)
+			require.Len(t, f.fixMessages, tt.initialPublishes+1)
+			require.Len(t, f.logs, 1)
+			if tt.initialPublishes != 0 {
+				require.Equal(t, f.fixMessages[0].ID, f.fixMessages[1].ID)
+				require.Equal(t, f.fixMessages[0].Payload, f.fixMessages[1].Payload)
+			}
+		})
+	}
+}
+
+func TestRepairHandoffUsesResumedBuildAndBuildScopedIdentity(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusRunning)
+	f.source("next-build", "11", entity.BuildStatusRunning)
+	f.reportIssue("build", "new", "unit")
+	f.failures["fix/publish"] = 1
+	require.ErrorIs(t, f.deliver(f.message("build", false), false), assert.AnError)
+
+	nextMessage := f.message("next-build", false)
+	require.NoError(t, f.deliver(nextMessage, true))
+	require.Equal(t, "build", f.progress.BuildID)
+	require.Len(t, f.fixMessages, 2)
+	require.Equal(t, f.fixMessages[0].ID, f.fixMessages[1].ID)
+	require.Equal(t, "build", f.fixMessages[1].PartitionKey)
+
+	require.NoError(t, f.deliver(f.message("build", false), false))
+	require.Len(t, f.fixMessages, 3)
+	require.Equal(t, f.fixMessages[0].ID, f.fixMessages[2].ID)
+
+	f.reportIssue("next-build", "new", "unit")
+	require.NoError(t, f.deliver(nextMessage, false))
+	require.Len(t, f.fixMessages, 3)
+	require.False(t, f.progress.HasNewIssues)
+	f.reportIssue("next-build", "another", "unit")
+	require.NoError(t, f.deliver(f.message("next-build", false), false))
+	require.Len(t, f.fixMessages, 4)
+	require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[3].ID)
+	require.Equal(t, "next-build", f.fixMessages[3].PartitionKey)
+}
+
+func TestTaskWideFailureStillPublishesRepair(t *testing.T) {
+	f := newFixture(t)
+	f.source("build", "10", entity.BuildStatusFailed)
+	f.seedIssue("existing", "unit", entity.IssueStatusOpen)
+	f.reportIssue("build", "setup", "")
+	require.NoError(t, f.deliver(f.message("build", true), false))
+	require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase)
+	require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"existing", "unit"}].Status)
+	require.Len(t, f.fixMessages, 1)
+}
diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go
index 6db9a8c3c..929067ed0 100644
--- a/stovepipe/controller/issuetrack/fixture_test.go
+++ b/stovepipe/controller/issuetrack/fixture_test.go
@@ -22,6 +22,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+	queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
 	"github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
@@ -55,6 +58,9 @@ type fixture struct {
 	requests    map[string]entity.Request
 	buildIssues map[string]map[issueKey]entity.BuildIssue
 	sequence    int
+
+	fixMessages  []entityqueue.Message
+	onFixPublish func(entityqueue.Message)
 }
 
 func newFixture(t *testing.T) *fixture {
@@ -235,7 +241,19 @@ func newFixture(t *testing.T) *fixture {
 		}
 		return result, nil
 	}).AnyTimes()
-	f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, f.factory, messagequeue.TopicKeyIssueTrack, "test")
+	queue := queuemock.NewMockQueue(f.mocks)
+	publisher := queuemock.NewMockPublisher(f.mocks)
+	queue.EXPECT().Publisher().Return(publisher).AnyTimes()
+	publisher.EXPECT().Publish(gomock.Any(), "fix", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error {
+		f.fixMessages = append(f.fixMessages, msg)
+		if f.onFixPublish != nil {
+			f.onFixPublish(msg)
+		}
+		return f.fail("fix/publish")
+	}).AnyTimes()
+	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{Key: messagequeue.TopicKeyFix, Name: "fix", Queue: queue}})
+	require.NoError(t, err)
+	f.controller = NewController(zap.NewNop().Sugar(), tally.NoopScope, f.factory, registry, messagequeue.TopicKeyIssueTrack, "test")
 	f.controller.nowMs = func() int64 { return 1000 }
 	return f
 }
diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go
index 8902fdc2f..0f85d9c05 100644
--- a/stovepipe/controller/issuetrack/issuetrack.go
+++ b/stovepipe/controller/issuetrack/issuetrack.go
@@ -27,6 +27,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/stovepipe/core/fixrequest"
 	"github.com/uber/submitqueue/stovepipe/core/loader"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
@@ -44,16 +45,18 @@ type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	stores        storage.Factory
+	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
 	consumerGroup string
 	nowMs         func() int64
 }
 
-func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, topicKey consumer.TopicKey, consumerGroup string) *Controller {
+func NewController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string) *Controller {
 	return &Controller{
 		logger:        logger.Named("issuetrack_controller"),
 		metricsScope:  scope.SubScope("issuetrack_controller"),
 		stores:        stores,
+		registry:      registry,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
 		nowMs:         func() int64 { return time.Now().UnixMilli() },
@@ -246,6 +249,11 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage,
 	}
 
 	if len(keys) < limit {
+		if !finalizing && progress.HasNewIssues {
+			if err := fixrequest.Publish(ctx, c.registry, queue, progress.BuildID); err != nil {
+				return entity.IssueTrackingState{}, fmt.Errorf("publish build repair request: %w", err)
+			}
+		}
 		next := progress
 		switch {
 		case finalizing:
@@ -316,6 +324,11 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr
 		}
 	}
 
+	if progress.Phase == entity.IssueTrackingPhaseRefreshing && reported &&
+		(missing || current.Status == entity.IssueStatusClosed || current.FirstSeenBuildID == progress.BuildID) {
+		progress.HasNewIssues = true
+	}
+
 	next := progress
 	next.CursorTaskID, next.CursorUnitID = key.task, key.unit
 	if !missing && current.LastProcessedBuildID == progress.BuildID && current.Status == desired {
diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go
index 664f83264..4b3c37567 100644
--- a/stovepipe/entity/issue_tracking_state.go
+++ b/stovepipe/entity/issue_tracking_state.go
@@ -57,6 +57,8 @@ type IssueTrackingState struct {
 	Phase IssueTrackingPhase `json:"phase"`
 	// Version is the positive optimistic-lock version of this state.
 	Version int32 `json:"version"`
+	// HasNewIssues records whether this pass encountered an issue first opened or reopened by its source build.
+	HasNewIssues bool `json:"has_new_issues"`
 }
 
 // Validate verifies the identity, phase, and version of a persisted state.
diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md
index 4e0de28ea..b761f67e8 100644
--- a/stovepipe/extension/storage/README.md
+++ b/stovepipe/extension/storage/README.md
@@ -68,6 +68,8 @@ The store does not decide source ordering, apply issue transitions, or advance v
 
 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.
 
+Existing MySQL deployments must append `has_new_issues BOOLEAN NOT NULL DEFAULT FALSE` at the end of `issue_tracking` before deploying the repair handoff. The flag retains openings across batches and retries until the opening pass can publish its build-scoped repair request. `CREATE TABLE IF NOT EXISTS` does not alter existing tables.
+
 ### Applied transition facts
 
 `IssueState.LastProcessedBuildID` identifies the most recent build applied to an issue, including unchanged results. `LastChangedBuildID` and `ChangedAtMs` identify the actual opening or closing and supply its history identity and timestamp. They do not advance when only the processed build changes. `FirstSeenBuildID` remains the opening identity across closure and changes on reopening. These references must remain readable while used for source ordering. The current-issue cursor and the row's processed source provide the storage needed for retry-safe application without retaining a future write payload.
diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store.go b/stovepipe/extension/storage/mysql/issue_tracking_store.go
index 9e81deeb9..012f5fd6a 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,
-            finalize, has_task_wide_issues, phase, version
+            finalize, has_task_wide_issues, phase, version, has_new_issues
          FROM issue_tracking WHERE queue = ?`, s.queue).
 		Scan(
 			&state.RequestID, &state.BuildID, &state.NotificationID,
 			&state.CursorTaskID, &state.CursorUnitID,
 
-			&state.Finalize, &state.HasTaskWideIssues, &state.Phase, &state.Version,
+			&state.Finalize, &state.HasTaskWideIssues, &state.Phase, &state.Version, &state.HasNewIssues,
 		)
 	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,
-            finalize, has_task_wide_issues, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+            finalize, has_task_wide_issues, phase, version, has_new_issues) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		s.queue, state.RequestID, state.BuildID, state.NotificationID,
 		state.CursorTaskID, state.CursorUnitID,
-		state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version,
+		state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version, state.HasNewIssues,
 	)
 	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 = ?,
-            finalize = ?, has_task_wide_issues = ?, phase = ?, version = ?
+            finalize = ?, has_task_wide_issues = ?, phase = ?, version = ?, has_new_issues = ?
          WHERE queue = ? AND version = ?`,
 		state.RequestID, state.BuildID, state.NotificationID,
 		state.CursorTaskID, state.CursorUnitID,
-		state.Finalize, state.HasTaskWideIssues, state.Phase, newVersion,
+		state.Finalize, state.HasTaskWideIssues, state.Phase, newVersion, state.HasNewIssues,
 		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 935ea6682..81f97c234 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.IssueTrackingPhaseFinalizing, Finalize: true, Version: 1}
+	return entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseFinalizing, Finalize: true, Version: 1, HasNewIssues: true}
 }
 
 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, finalize, has_task_wide_issues, 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, has_new_issues 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", "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))
+				query.WillReturnRows(sqlmock.NewRows([]string{"request_id", "build_id", "notification_id", "cursor_task_id", "cursor_unit_id", "finalize", "has_task_wide_issues", "phase", "version", "has_new_issues"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version, state.HasNewIssues))
 			}
 
 			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, 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)
+			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, has_new_issues) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)).
+				WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version, state.HasNewIssues)
 			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 = ?, 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))
+			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 = ?, has_new_issues = ? WHERE queue = ? AND version = ?`)).
+				WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, int32(7), state.HasNewIssues, "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 9d367d84c..70ad76ca3 100644
--- a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql
+++ b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql
@@ -12,5 +12,6 @@ CREATE TABLE IF NOT EXISTS issue_tracking (
     has_task_wide_issues BOOLEAN         NOT NULL DEFAULT FALSE,
     phase                VARBINARY(32)   NOT NULL,
     version              INT             NOT NULL,
+    has_new_issues       BOOLEAN         NOT NULL DEFAULT FALSE,
     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 b8b63bd4f..aedb8825a 100644
--- a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go
+++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go
@@ -32,7 +32,7 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F
 	store := bound.GetIssueTrackingStore()
 	_, err = store.Get(ctx)
 	require.ErrorIs(t, err, storage.ErrNotFound)
-	state := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1}
+	state := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1, HasNewIssues: true}
 	require.NoError(t, store.Create(ctx, state))
 	require.ErrorIs(t, store.Create(ctx, state), storage.ErrAlreadyExists)
 	got, err := store.Get(ctx)
@@ -41,6 +41,7 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F
 
 	state.CursorTaskID, state.CursorUnitID = "task", "unit"
 	state.Phase = entity.IssueTrackingPhaseRefreshed
+	state.HasNewIssues = false
 	require.NoError(t, store.Update(ctx, state, 1, 2))
 	state.Version = 2
 	require.ErrorIs(t, store.Update(ctx, state, 1, 3), storage.ErrVersionMismatch)

From 87ba6536c3615323604c6e8933b050e0d729640e Mon Sep 17 00:00:00 2001
From: mnoah1 <mnoah1@uber.com>
Date: Tue, 27 Oct 2026 22:16:49 +0000
Subject: [PATCH 2/2] fix(stovepipe): publish repair handoffs per nonempty pass

Use the existing opening cursor to skip empty passes and let the fix controller select eligible issues. Key repair messages by build and persisted pass notification so retries deduplicate without suppressing later passes. Remove the opening flag and its schema change.
---
 stovepipe/controller/fix/README.md            |  2 +-
 stovepipe/controller/issuetrack/README.md     |  2 +-
 .../controller/issuetrack/fix_handoff_test.go | 80 +++++++++++--------
 stovepipe/controller/issuetrack/issuetrack.go |  9 +--
 stovepipe/core/fixrequest/publish.go          | 13 ++-
 stovepipe/core/fixrequest/publish_test.go     | 36 +++++----
 stovepipe/entity/issue_tracking_state.go      |  2 -
 stovepipe/extension/storage/README.md         |  2 -
 .../storage/mysql/issue_tracking_store.go     | 12 +--
 .../mysql/issue_tracking_store_test.go        | 14 ++--
 .../storage/mysql/schema/issue_tracking.sql   |  1 -
 .../storage/mysql/issue_tracking_test.go      |  3 +-
 12 files changed, 90 insertions(+), 86 deletions(-)

diff --git a/stovepipe/controller/fix/README.md b/stovepipe/controller/fix/README.md
index bca017ade..489766326 100644
--- a/stovepipe/controller/fix/README.md
+++ b/stovepipe/controller/fix/README.md
@@ -4,6 +4,6 @@ The controller consumes a `FixRequest` containing a queue name and build ID. It
 
 A returned fix ID is associated with the issue through `BuildIssueStore.Update`, using the version read during selection. A concurrent association is preserved. A submission error without an ID leaves the issue eligible for normal queue retries; an ID returned alongside an error is retained.
 
-Issue tracking calls [`fixrequest.Publish`](../../core/fixrequest/publish.go) after persisting both the build observation and open issue. Publications for the same build share a message identity and partition.
+Issue tracking calls [`fixrequest.Publish`](../../core/fixrequest/publish.go) after completing a nonempty opening pass and persisting its issue state and history. Publications share a build partition, while the message identity includes the pass notification ID so a later pass can trigger another evaluation. A pass with no eligible issue completes without submitting a repair; a recorded fix ID prevents later passes from submitting another repair for the build.
 
 The one-per-build limit is controller policy. Submission precedes persistence, so overlapping deliveries, crashes, and failed writes can still produce duplicate executor calls. Issue recovery remains separate from repair association.
diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md
index 50681208c..abd4fb740 100644
--- a/stovepipe/controller/issuetrack/README.md
+++ b/stovepipe/controller/issuetrack/README.md
@@ -22,7 +22,7 @@ The post-read cursor write and processed-build updates, including unchanged stat
 
 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.
 
-The progress record retains `HasNewIssues` when an issue is first opened or reopened by the source build, including when replaying that opening. Once the entire build-issue range is exhausted, a pass with new issues publishes a build-scoped repair request before saving its next phase. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. Refreshes for the same build share a repair message identity; the fix controller submits one repair per build.
+Once the entire build-issue range is exhausted, an opening pass with a nonempty task cursor publishes a build-scoped repair request before saving its next phase. Empty passes and closing sweeps do not publish. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. The repair message identity combines the build and persisted pass notification IDs: retries share an identity, while later passes can trigger another evaluation. The fix controller selects an eligible issue and submits at most one repair per build.
 
 Once the entire build-issue range is exhausted:
 
diff --git a/stovepipe/controller/issuetrack/fix_handoff_test.go b/stovepipe/controller/issuetrack/fix_handoff_test.go
index 3ee54b79f..1b82ea3bc 100644
--- a/stovepipe/controller/issuetrack/fix_handoff_test.go
+++ b/stovepipe/controller/issuetrack/fix_handoff_test.go
@@ -69,45 +69,59 @@ func TestOpeningPassPublishesOneBuildRepairRequest(t *testing.T) {
 	}
 }
 
-func TestOpeningPassWithoutNewIssuesDoesNotPublishRepair(t *testing.T) {
-	for _, reportExisting := range []bool{false, true} {
-		t.Run(fmt.Sprint(reportExisting), func(t *testing.T) {
+func TestRepairHandoffSkipsEmptyPassAndPublishesLaterPasses(t *testing.T) {
+	for _, tt := range []struct {
+		name             string
+		reportExisting   bool
+		initialPublishes int
+	}{
+		{name: "empty pass"},
+		{name: "previously open issues", reportExisting: true, initialPublishes: 1},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
 			f := newFixture(t)
 			f.source("build", "10", entity.BuildStatusRunning)
 			f.seedIssue("existing", "unit", entity.IssueStatusOpen)
-			if reportExisting {
+			if tt.reportExisting {
 				f.reportIssue("build", "existing", "unit")
 			}
 			require.NoError(t, f.deliver(f.message("build", false), false))
-			require.Empty(t, f.fixMessages)
+			require.Len(t, f.fixMessages, tt.initialPublishes)
 			f.reportIssue("build", "new", "unit")
 			require.NoError(t, f.deliver(f.message("build", false), false))
-			require.Len(t, f.fixMessages, 1)
+			require.Len(t, f.fixMessages, tt.initialPublishes+1)
+			if tt.initialPublishes != 0 {
+				require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[1].ID)
+				require.Equal(t, f.fixMessages[0].PartitionKey, f.fixMessages[1].PartitionKey)
+			}
 		})
 	}
 }
 
 func TestRepairHandoffWaitsForAllOpeningBatches(t *testing.T) {
-	f := newFixture(t)
-	f.source("build", "10", entity.BuildStatusRunning)
-	f.reportIssue("build", "a-new", "unit")
-	for i := 0; i < 100; i++ {
-		unit := fmt.Sprintf("unit/%03d", i)
-		f.seedIssue("existing", unit, entity.IssueStatusOpen)
-		f.reportIssue("build", "existing", unit)
-	}
-	msg := f.message("build", false)
-	require.NoError(t, f.deliver(msg, true))
-	require.Empty(t, f.fixMessages)
-	require.True(t, f.progress.HasNewIssues)
+	for _, issueCount := range []int{_issueBatchSize, _issueBatchSize + 1} {
+		t.Run(fmt.Sprintf("%d issues", issueCount), func(t *testing.T) {
+			f := newFixture(t)
+			f.source("build", "10", entity.BuildStatusRunning)
+			f.reportIssue("build", "a-new", "unit")
+			for i := 1; i < issueCount; i++ {
+				unit := fmt.Sprintf("unit/%03d", i)
+				f.seedIssue("existing", unit, entity.IssueStatusOpen)
+				f.reportIssue("build", "existing", unit)
+			}
+			msg := f.message("build", false)
+			require.NoError(t, f.deliver(msg, true))
+			require.Empty(t, f.fixMessages)
 
-	f.onFixPublish = func(entityqueue.Message) {
-		for _, issue := range f.states {
-			require.Equal(t, "build", issue.LastProcessedBuildID)
-		}
+			f.onFixPublish = func(entityqueue.Message) {
+				for _, issue := range f.states {
+					require.Equal(t, "build", issue.LastProcessedBuildID)
+				}
+			}
+			require.NoError(t, f.deliver(msg, false))
+			require.Len(t, f.fixMessages, 1)
+		})
 	}
-	require.NoError(t, f.deliver(msg, false))
-	require.Len(t, f.fixMessages, 1)
 }
 
 func TestRepairHandoffRetriesBeforeCompletingPass(t *testing.T) {
@@ -145,13 +159,14 @@ func TestRepairHandoffRetriesBeforeCompletingPass(t *testing.T) {
 	}
 }
 
-func TestRepairHandoffUsesResumedBuildAndBuildScopedIdentity(t *testing.T) {
+func TestRepairHandoffUsesResumedBuildAndPassIdentity(t *testing.T) {
 	f := newFixture(t)
 	f.source("build", "10", entity.BuildStatusRunning)
 	f.source("next-build", "11", entity.BuildStatusRunning)
 	f.reportIssue("build", "new", "unit")
 	f.failures["fix/publish"] = 1
-	require.ErrorIs(t, f.deliver(f.message("build", false), false), assert.AnError)
+	msg := f.message("build", false)
+	require.ErrorIs(t, f.deliver(msg, false), assert.AnError)
 
 	nextMessage := f.message("next-build", false)
 	require.NoError(t, f.deliver(nextMessage, true))
@@ -160,19 +175,14 @@ func TestRepairHandoffUsesResumedBuildAndBuildScopedIdentity(t *testing.T) {
 	require.Equal(t, f.fixMessages[0].ID, f.fixMessages[1].ID)
 	require.Equal(t, "build", f.fixMessages[1].PartitionKey)
 
-	require.NoError(t, f.deliver(f.message("build", false), false))
-	require.Len(t, f.fixMessages, 3)
-	require.Equal(t, f.fixMessages[0].ID, f.fixMessages[2].ID)
+	require.NoError(t, f.deliver(msg, false))
+	require.Len(t, f.fixMessages, 2)
 
 	f.reportIssue("next-build", "new", "unit")
 	require.NoError(t, f.deliver(nextMessage, false))
 	require.Len(t, f.fixMessages, 3)
-	require.False(t, f.progress.HasNewIssues)
-	f.reportIssue("next-build", "another", "unit")
-	require.NoError(t, f.deliver(f.message("next-build", false), false))
-	require.Len(t, f.fixMessages, 4)
-	require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[3].ID)
-	require.Equal(t, "next-build", f.fixMessages[3].PartitionKey)
+	require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[2].ID)
+	require.Equal(t, "next-build", f.fixMessages[2].PartitionKey)
 }
 
 func TestTaskWideFailureStillPublishesRepair(t *testing.T) {
diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go
index 0f85d9c05..a39576d01 100644
--- a/stovepipe/controller/issuetrack/issuetrack.go
+++ b/stovepipe/controller/issuetrack/issuetrack.go
@@ -249,8 +249,8 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage,
 	}
 
 	if len(keys) < limit {
-		if !finalizing && progress.HasNewIssues {
-			if err := fixrequest.Publish(ctx, c.registry, queue, progress.BuildID); err != nil {
+		if !finalizing && progress.CursorTaskID != "" {
+			if err := fixrequest.Publish(ctx, c.registry, queue, progress.BuildID, progress.NotificationID); err != nil {
 				return entity.IssueTrackingState{}, fmt.Errorf("publish build repair request: %w", err)
 			}
 		}
@@ -324,11 +324,6 @@ func (c *Controller) processIssue(ctx context.Context, store storage.Storage, pr
 		}
 	}
 
-	if progress.Phase == entity.IssueTrackingPhaseRefreshing && reported &&
-		(missing || current.Status == entity.IssueStatusClosed || current.FirstSeenBuildID == progress.BuildID) {
-		progress.HasNewIssues = true
-	}
-
 	next := progress
 	next.CursorTaskID, next.CursorUnitID = key.task, key.unit
 	if !missing && current.LastProcessedBuildID == progress.BuildID && current.Status == desired {
diff --git a/stovepipe/core/fixrequest/publish.go b/stovepipe/core/fixrequest/publish.go
index 21f684316..f41a6147f 100644
--- a/stovepipe/core/fixrequest/publish.go
+++ b/stovepipe/core/fixrequest/publish.go
@@ -24,11 +24,11 @@ import (
 	mq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 )
 
-// Publish requires the build and its open issues to be persisted first.
-// Replays after publication errors must call Publish again.
-func Publish(ctx context.Context, registry consumer.TopicRegistry, queueName, buildID string) error {
-	if queueName == "" || buildID == "" {
-		return fmt.Errorf("repair request requires a queue and build ID")
+// Publish requires the opening pass's issue state and history to be durable.
+// The notification ID must be stable across retries and distinct between passes.
+func Publish(ctx context.Context, registry consumer.TopicRegistry, queueName, buildID, notificationID string) error {
+	if queueName == "" || buildID == "" || notificationID == "" {
+		return fmt.Errorf("repair request requires a queue, build ID, and notification ID")
 	}
 
 	payload, err := mq.Marshal(&mq.FixRequest{Id: buildID, QueueName: queueName})
@@ -36,10 +36,9 @@ func Publish(ctx context.Context, registry consumer.TopicRegistry, queueName, bu
 		return fmt.Errorf("failed to serialize repair request: %w", err)
 	}
 
-	// Repair submission is a one-shot handoff for this build.
 	return publish.Message(ctx, registry, mq.TopicKeyFix, publish.MessageParams{
 		Tenant:       queueName,
-		ID:           publish.IntentID(buildID),
+		ID:           publish.IntentID(buildID, notificationID),
 		Payload:      payload,
 		PartitionKey: buildID,
 	})
diff --git a/stovepipe/core/fixrequest/publish_test.go b/stovepipe/core/fixrequest/publish_test.go
index 0b2a600cc..ee5146246 100644
--- a/stovepipe/core/fixrequest/publish_test.go
+++ b/stovepipe/core/fixrequest/publish_test.go
@@ -40,7 +40,7 @@ func newTestRegistry(t *testing.T) (consumer.TopicRegistry, *queuemock.MockPubli
 	return registry, publisher
 }
 
-func TestPublishBuildIdentityAndRetry(t *testing.T) {
+func TestPublishBuildAndPassIdentityAndRetry(t *testing.T) {
 	registry, publisher := newTestRegistry(t)
 	var messages []entityqueue.Message
 	publisher.EXPECT().Publish(gomock.Any(), "test-fix", gomock.Any()).
@@ -50,18 +50,22 @@ func TestPublishBuildIdentityAndRetry(t *testing.T) {
 				return assert.AnError
 			}
 			return nil
-		}).Times(4)
+		}).Times(5)
 
 	ctx := context.Background()
-	require.ErrorIs(t, Publish(ctx, registry, "validation/main", "Build/42 "), assert.AnError)
-	require.NoError(t, Publish(ctx, registry, "validation/main", "Build/42 "))
-	require.NoError(t, Publish(ctx, registry, "validation/main", "Build/43"))
-	require.NoError(t, Publish(ctx, registry, "validation/other", "Build/42 "))
+	require.ErrorIs(t, Publish(ctx, registry, "validation/main", "Build/42 ", "pass/1"), assert.AnError)
+	require.NoError(t, Publish(ctx, registry, "validation/main", "Build/42 ", "pass/1"))
+	require.NoError(t, Publish(ctx, registry, "validation/main", "Build/42 ", "pass/2"))
+	require.NoError(t, Publish(ctx, registry, "validation/main", "Build/43", "pass/1"))
+	require.NoError(t, Publish(ctx, registry, "validation/other", "Build/42 ", "pass/1"))
 
 	assert.Equal(t, messages[0].ID, messages[1].ID)
 	assert.Equal(t, messages[0].PartitionKey, messages[1].PartitionKey)
 	assert.NotEqual(t, messages[1].ID, messages[2].ID)
-	assert.NotEqual(t, messages[1].Tenant, messages[3].Tenant)
+	assert.Equal(t, messages[1].PartitionKey, messages[2].PartitionKey)
+	assert.Equal(t, messages[1].Payload, messages[2].Payload)
+	assert.NotEqual(t, messages[1].ID, messages[3].ID)
+	assert.NotEqual(t, messages[1].Tenant, messages[4].Tenant)
 	for _, message := range messages {
 		payload := &mq.FixRequest{}
 		require.NoError(t, mq.Unmarshal(message.Payload, payload))
@@ -74,22 +78,24 @@ func TestPublishBuildIdentityAndRetry(t *testing.T) {
 
 func TestPublishRejectsMissingIdentity(t *testing.T) {
 	for _, test := range []struct {
-		name  string
-		queue string
-		build string
+		name         string
+		queue        string
+		build        string
+		notification string
 	}{
-		{name: "missing queue", build: "build/42"},
-		{name: "missing build", queue: "validation/main"},
+		{name: "missing queue", build: "build/42", notification: "pass/1"},
+		{name: "missing build", queue: "validation/main", notification: "pass/1"},
+		{name: "missing notification", queue: "validation/main", build: "build/42"},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			registry, _ := newTestRegistry(t)
-			require.Error(t, Publish(context.Background(), registry, test.queue, test.build))
+			require.Error(t, Publish(context.Background(), registry, test.queue, test.build, test.notification))
 		})
 	}
 }
 
 func TestPublishReturnsConfigurationAndSerializationErrors(t *testing.T) {
-	require.Error(t, Publish(context.Background(), consumer.TopicRegistry{}, "validation/main", "build/42"))
+	require.Error(t, Publish(context.Background(), consumer.TopicRegistry{}, "validation/main", "build/42", "pass/1"))
 	registry, _ := newTestRegistry(t)
-	require.Error(t, Publish(context.Background(), registry, "validation/main", string([]byte{0xff})))
+	require.Error(t, Publish(context.Background(), registry, "validation/main", string([]byte{0xff}), "pass/1"))
 }
diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go
index 4b3c37567..664f83264 100644
--- a/stovepipe/entity/issue_tracking_state.go
+++ b/stovepipe/entity/issue_tracking_state.go
@@ -57,8 +57,6 @@ type IssueTrackingState struct {
 	Phase IssueTrackingPhase `json:"phase"`
 	// Version is the positive optimistic-lock version of this state.
 	Version int32 `json:"version"`
-	// HasNewIssues records whether this pass encountered an issue first opened or reopened by its source build.
-	HasNewIssues bool `json:"has_new_issues"`
 }
 
 // Validate verifies the identity, phase, and version of a persisted state.
diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md
index b761f67e8..4e0de28ea 100644
--- a/stovepipe/extension/storage/README.md
+++ b/stovepipe/extension/storage/README.md
@@ -68,8 +68,6 @@ The store does not decide source ordering, apply issue transitions, or advance v
 
 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.
 
-Existing MySQL deployments must append `has_new_issues BOOLEAN NOT NULL DEFAULT FALSE` at the end of `issue_tracking` before deploying the repair handoff. The flag retains openings across batches and retries until the opening pass can publish its build-scoped repair request. `CREATE TABLE IF NOT EXISTS` does not alter existing tables.
-
 ### Applied transition facts
 
 `IssueState.LastProcessedBuildID` identifies the most recent build applied to an issue, including unchanged results. `LastChangedBuildID` and `ChangedAtMs` identify the actual opening or closing and supply its history identity and timestamp. They do not advance when only the processed build changes. `FirstSeenBuildID` remains the opening identity across closure and changes on reopening. These references must remain readable while used for source ordering. The current-issue cursor and the row's processed source provide the storage needed for retry-safe application without retaining a future write payload.
diff --git a/stovepipe/extension/storage/mysql/issue_tracking_store.go b/stovepipe/extension/storage/mysql/issue_tracking_store.go
index 012f5fd6a..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,
-            finalize, has_task_wide_issues, phase, version, has_new_issues
+            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.Finalize, &state.HasTaskWideIssues, &state.Phase, &state.Version, &state.HasNewIssues,
+			&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,
-            finalize, has_task_wide_issues, phase, version, has_new_issues) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+            finalize, has_task_wide_issues, phase, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		s.queue, state.RequestID, state.BuildID, state.NotificationID,
 		state.CursorTaskID, state.CursorUnitID,
-		state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version, state.HasNewIssues,
+		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 = ?,
-            finalize = ?, has_task_wide_issues = ?, phase = ?, version = ?, has_new_issues = ?
+            finalize = ?, has_task_wide_issues = ?, phase = ?, version = ?
          WHERE queue = ? AND version = ?`,
 		state.RequestID, state.BuildID, state.NotificationID,
 		state.CursorTaskID, state.CursorUnitID,
-		state.Finalize, state.HasTaskWideIssues, state.Phase, newVersion, state.HasNewIssues,
+		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 81f97c234..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.IssueTrackingPhaseFinalizing, Finalize: true, Version: 1, HasNewIssues: true}
+	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, finalize, has_task_wide_issues, phase, version, has_new_issues 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", "finalize", "has_task_wide_issues", "phase", "version", "has_new_issues"}).AddRow(request, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version, state.HasNewIssues))
+				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, finalize, has_task_wide_issues, phase, version, has_new_issues) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)).
-				WithArgs("queue", state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, state.Version, state.HasNewIssues)
+			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 = ?, finalize = ?, has_task_wide_issues = ?, phase = ?, version = ?, has_new_issues = ? WHERE queue = ? AND version = ?`)).
-				WithArgs(state.RequestID, state.BuildID, state.NotificationID, state.CursorTaskID, state.CursorUnitID, state.Finalize, state.HasTaskWideIssues, state.Phase, int32(7), state.HasNewIssues, "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 70ad76ca3..9d367d84c 100644
--- a/stovepipe/extension/storage/mysql/schema/issue_tracking.sql
+++ b/stovepipe/extension/storage/mysql/schema/issue_tracking.sql
@@ -12,6 +12,5 @@ CREATE TABLE IF NOT EXISTS issue_tracking (
     has_task_wide_issues BOOLEAN         NOT NULL DEFAULT FALSE,
     phase                VARBINARY(32)   NOT NULL,
     version              INT             NOT NULL,
-    has_new_issues       BOOLEAN         NOT NULL DEFAULT FALSE,
     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 aedb8825a..b8b63bd4f 100644
--- a/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go
+++ b/test/integration/stovepipe/extension/storage/mysql/issue_tracking_test.go
@@ -32,7 +32,7 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F
 	store := bound.GetIssueTrackingStore()
 	_, err = store.Get(ctx)
 	require.ErrorIs(t, err, storage.ErrNotFound)
-	state := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1, HasNewIssues: true}
+	state := entity.IssueTrackingState{RequestID: "10", BuildID: "build/10", NotificationID: "notification/10", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1}
 	require.NoError(t, store.Create(ctx, state))
 	require.ErrorIs(t, store.Create(ctx, state), storage.ErrAlreadyExists)
 	got, err := store.Get(ctx)
@@ -41,7 +41,6 @@ func testIssueTrackingStore(t *testing.T, ctx context.Context, factory storage.F
 
 	state.CursorTaskID, state.CursorUnitID = "task", "unit"
 	state.Phase = entity.IssueTrackingPhaseRefreshed
-	state.HasNewIssues = false
 	require.NoError(t, store.Update(ctx, state, 1, 2))
 	state.Version = 2
 	require.ErrorIs(t, store.Update(ctx, state, 1, 3), storage.ErrVersionMismatch)