feat(stovepipe): hand off newly opened issues for repair

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",

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)
 

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.

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",

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.
 
+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:
 
 - 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
 

stovepipe/controller/issuetrack/fix_handoff_test.go

@@ -0,0 +1,197 @@
+// 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 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 tt.reportExisting {
+				f.reportIssue("build", "existing", "unit")
+			}
+			require.NoError(t, f.deliver(f.message("build", false), false))
+			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, 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) {
+	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)
+				}
+			}
+			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 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
+	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))
+	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(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.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[2].ID)
+	require.Equal(t, "next-build", f.fixMessages[2].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)
+}

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
 }

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.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)
+			}
+		}
 		next := progress
 		switch {
 		case finalizing:

stovepipe/core/fixrequest/publish.go

@@ -24,22 +24,21 @@ 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})
 	if err != nil {
 		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,
 	})