feat(stovepipe): Define issue-tracking handoff

stovepipe/core/messagequeue/README.md

@@ -11,3 +11,13 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali
 - **buildsignal** (`TopicKeyBuildSignal`, `BuildSignal`) — build publishes the build id here after triggering; buildsignal holds the delivery between polls until the build reaches a terminal status. Partitioned by build id, so each build's poll loop is an independent partition.
 - **buildanalysis** (`TopicKeyBuildAnalysis`, `BuildAnalysis`) — identifies a persisted build for issue detection, partitioned by build id. Build publishes a start notice and buildsignal publishes a completion notice with a distinct message id. See [build analysis](../../controller/buildanalysis/README.md) for polling and persistence behavior.
 - **record** (`TopicKeyRecord`, `Record`) — buildsignal publishes the request id after persisting the terminal outcome; record reloads the request and records its validation fact. Partitioned by request id.
+
+## Issue tracking handoff
+
+`IssueTrack` binds to `TopicKeyIssueTrack` and identifies a persisted build and its queue. Notifications are partitioned by queue so both running and final issue-state updates share one serialized path, independently of build-partitioned analysis polling. The message carries identifiers rather than a failure collection; per-build observations are read through `BuildIssueStore.List`.
+
+Publish only after detection and observation persistence succeed. With `finalize` unset, the notification permits opening newly detected issues, not resolution by absence. With `finalize` set, it requests final reconciliation after a successful terminal analysis. Publish that notification even when no failures were detected; terminal Build status alone does not replace the handoff. Publication retries must also follow duplicate observation writes so a crash between persistence and publication does not strand work.
+
+Per-build observations accumulate historical evidence; finalization does not turn them into a replacement snapshot or override the detector's absence semantics. The reconciler must establish the authoritative final failure set and applicable validation scope before resolving issues, reject older requests, and ignore running notifications after finalization. Notification identity must distinguish new observations from finalization while remaining stable on publication retries. Business behavior follows the payload, not a parsed message ID.
+
+Only the handoff contract is defined. No publisher, consumer, topic registration, reconciliation progress, or writes to open issues and lifecycle history are enabled.

stovepipe/core/messagequeue/messagequeue.go

@@ -53,6 +53,9 @@ type (
 	// BuildAnalysis identifies the build whose issue observations are polled.
 	BuildAnalysis = protopb.BuildAnalysis
 
+	// IssueTrack identifies persisted build observations for queue-scoped reconciliation.
+	IssueTrack = protopb.IssueTrack
+
 	// Record is the payload buildsignal publishes to the record stage once a
 	// build reaches a terminal status: the build id to record.
 	Record = protopb.Record

stovepipe/core/messagequeue/messagequeue_test.go

@@ -80,7 +80,7 @@ func TestWireFormat(t *testing.T) {
 // no topic_keys option names an unknown key.
 func TestTopicKeysBindEveryTopicKey(t *testing.T) {
 	bound := map[string]int{}
-	for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &BuildAnalysis{}, &Record{}} {
+	for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &BuildAnalysis{}, &IssueTrack{}, &Record{}} {
 		keys := TopicKeys(m)
 		require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option")
 		for _, key := range keys {
@@ -93,6 +93,7 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) {
 		TopicKeyBuild,
 		TopicKeyBuildSignal,
 		TopicKeyBuildAnalysis,
+		TopicKeyIssueTrack,
 		TopicKeyRecord,
 	}
 
@@ -117,3 +118,31 @@ func TestBuildAnalysisRoundTrip(t *testing.T) {
 	assert.True(t, proto.Equal(analysis, got))
 	assert.JSONEq(t, `{"id":"build/42","queue_name":"validation/main"}`, string(data))
 }
+
+func TestIssueTrackRoundTrip(t *testing.T) {
+	for _, tt := range []struct {
+		name     string
+		finalize bool
+		wire     string
+	}{
+		{name: "observations", wire: `{"build_id":"build/42","queue_name":"validation/main"}`},
+		{name: "final analysis", finalize: true, wire: `{"build_id":"build/42","queue_name":"validation/main","finalize":true}`},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			notification := &IssueTrack{BuildId: "build/42", QueueName: "validation/main", Finalize: tt.finalize}
+			data, err := Marshal(notification)
+			require.NoError(t, err)
+			assert.JSONEq(t, tt.wire, string(data))
+
+			got := &IssueTrack{}
+			require.NoError(t, Unmarshal(data, got))
+			assert.True(t, proto.Equal(notification, got))
+		})
+	}
+}
+
+func TestIssueTrackToleratesUnknownFields(t *testing.T) {
+	got := &IssueTrack{}
+	require.NoError(t, Unmarshal([]byte(`{"build_id":"build/42","queue_name":"validation/main","finalize":true,"future_field":42}`), got))
+	assert.True(t, proto.Equal(&IssueTrack{BuildId: "build/42", QueueName: "validation/main", Finalize: true}, got))
+}

stovepipe/core/messagequeue/proto/BUILD.bazel

@@ -3,6 +3,7 @@ exports_files(
         "build.proto",
         "buildanalysis.proto",
         "buildsignal.proto",
+        "issuetrack.proto",
         "process.proto",
         "record.proto",
     ],

stovepipe/core/messagequeue/proto/issuetrack.proto

@@ -0,0 +1,39 @@
+// 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.
+
+syntax = "proto3";
+
+package uber.stovepipe.messagequeue;
+
+import "api/base/messagequeue/proto/messagequeue.proto";
+
+option go_package = "github.com/uber/submitqueue/stovepipe/core/messagequeue/protopb";
+option java_multiple_files = true;
+option java_outer_classname = "IssueTrackProto";
+option java_package = "com.uber.submitqueue.stovepipe.messagequeue";
+
+// IssueTrack requests queue-scoped reconciliation of persisted build observations.
+// Publish only after successful detection and observation persistence; partition by queue.
+message IssueTrack {
+    option (uber.base.messagequeue.topic_keys) = "issuetrack";
+
+    // build_id is the required opaque identifier of the persisted build.
+    string build_id = 1;
+    // queue_name is the required queue namespace of the build.
+    string queue_name = 2;
+    // finalize requests final reconciliation after successful terminal detection
+    // and persistence, including when no failures were detected.
+    // Otherwise the notification permits opening issues, not resolution by absence.
+    bool finalize = 3;
+}

stovepipe/core/messagequeue/protopb/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "build.pb.go",
         "buildanalysis.pb.go",
         "buildsignal.pb.go",
+        "issuetrack.pb.go",
         "process.pb.go",
         "record.pb.go",
     ],

stovepipe/core/messagequeue/topics.go

@@ -45,6 +45,9 @@ const (
 	// TopicKeyBuildAnalysis carries build IDs for issue detection, partitioned by build ID.
 	TopicKeyBuildAnalysis TopicKey = "buildanalysis"
 
+	// TopicKeyIssueTrack carries persisted observation notifications, partitioned by queue.
+	TopicKeyIssueTrack TopicKey = "issuetrack"
+
 	// TopicKeyRecord carries a request whose build reached a terminal status from
 	// buildsignal to the record stage. The buildsignal controller publishes a
 	// Record (the request id) here once, and only once, a build reaches a

stovepipe/entity/build_issue.go

@@ -16,7 +16,7 @@ package entity
 
 // BuildIssue identifies a validation failure by task and optional unit.
 type BuildIssue struct {
-	// TaskID identifies the logical validation task, stable across builds and retries.
+	// TaskID identifies the logical validation task, stable across builds and retries; it must be nonempty.
 	TaskID string `json:"task_id"`
 	// UnitID identifies the failed unit within the task; empty for a task-wide issue.
 	// Repeated observations of the same condition use the same TaskID and UnitID.

stovepipe/extension/storage/README.md

@@ -28,7 +28,9 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte
 
 ## Build issue observations
 
-`BuildIssueStore` accumulates immutable observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Empty unit IDs represent task-wide observations.
+`BuildIssueStore` accumulates immutable observations identified by build, task, and unit within the bound queue. The build ID is supplied separately from the detected issue. Repeated observations return `ErrAlreadyExists`, which callers may treat as successful persistence. Build and task IDs must be nonempty; empty unit IDs represent task-wide observations.
+
+Listing fixes the queue/build primary-key prefix and returns bounded pages in ascending bytewise task/unit order. A task/unit cursor continues after the last returned key; the zero cursor starts the range, and a unit without a task is invalid. Pages are not a snapshot across calls: concurrent observations can affect membership. Empty pages represent no observations in that range, not build success or completed analysis.
 
 Each observation is stored separately, not in a serialized collection. Observations are historical evidence; they do not track whether an issue is currently open or resolved. No update, deletion, or absence-based recovery is implied by this store.
 

stovepipe/extension/storage/build_issue_store.go

@@ -22,13 +22,35 @@ import (
 	"github.com/uber/submitqueue/stovepipe/entity"
 )
 
+// BuildIssueCursor is an exclusive position in ascending bytewise task/unit order.
+type BuildIssueCursor struct {
+	// TaskID is the task component; empty with an empty UnitID starts the range.
+	TaskID string
+	// UnitID is the unit component; empty identifies a task-wide observation.
+	UnitID string
+}
+
+// BuildIssueRange bounds a primary-key scan within one build.
+type BuildIssueRange struct {
+	// After excludes this key and earlier keys; the zero cursor starts the range.
+	After BuildIssueCursor
+	// Limit is the positive maximum number of observations to return.
+	Limit int
+}
+
 // BuildIssueStore retains immutable observations keyed by build, task, and unit
 // within the bound queue. Presence records an observation, not a currently open issue.
+// Observation keys require nonempty build and task IDs; an empty unit is task-wide.
 type BuildIssueStore interface {
 	// Create returns ErrAlreadyExists for a previously recorded build/task/unit.
 	// Callers may treat that duplicate as success when accumulating observations.
 	Create(ctx context.Context, buildID string, issue entity.BuildIssue) error
 
 	// Get returns ErrNotFound when the build/task/unit has not been recorded.
 	Get(ctx context.Context, buildID, taskID, unitID string) (entity.BuildIssue, error)
+
+	// List returns at most Limit observations in ascending bytewise (task, unit) order.
+	// An empty slice ends the range; invalid bounds or an empty build ID return an error.
+	// Pages are not a snapshot: concurrent observations may change membership.
+	List(ctx context.Context, buildID string, bounds BuildIssueRange) ([]entity.BuildIssue, error)
 }

stovepipe/extension/storage/mysql/BUILD.bazel

@@ -33,6 +33,7 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "build_issue_list_test.go",
         "build_issue_store_test.go",
         "build_store_test.go",
         "issue_log_store_test.go",