feat(stovepipe): define repair coordination contracts and RFC

Makefile

@@ -629,6 +629,7 @@ GO_GENERATE_PACKAGES := \
 	./stovepipe/controller/... \
 	./stovepipe/core/requestlog/... \
 	./stovepipe/extension/buildrunner/... \
+	./stovepipe/extension/fixer/... \
 	./stovepipe/extension/projectstatus/... \
 	./stovepipe/extension/queueconfig/... \
 	./stovepipe/extension/sourcecontrol/... \

doc/rfc/stovepipe/workflow.md

@@ -189,6 +189,8 @@ Per-stage design detail lives under `steps/` so this doc stays a pipeline overvi
 - [buildsignal.md](steps/buildsignal.md) — the poll loop: hold-based re-poll cadence, per-build partitioning, and the fail-closed handoff to record
 - [record.md](steps/record.md) — turning a terminal build outcome into an immutable validation fact, monotonic last-green advancement and ref promotion, and the hook event announcing the outcome. Project analysis already runs inline through `projectstatus.Resolver` and may later use results calculated by build analysis
 
+For repair decisions, execution, and recovery, see the [repair RFC](fixer.md).
+
 ## Dedup, idempotency, and history rewrites
 
 Ingestion is idempotent on `(Queue, head URI)`, so duplicate poller reports — and any future webhook producer racing the poller — converge on one Request. The pipeline persists the Queue's last-green URI and per-URI greenness, so it must tolerate a **history rewrite**: when `SourceControl` reports that the last-green URI is no longer an ancestor of the current head, `process` falls back to a full-monorepo build rather than trusting a baseline that no longer exists on the branch. The system converges to a correct greenness rather than wedging on a stale pointer.

stovepipe/core/messagequeue/README.md

@@ -10,3 +10,12 @@ Payloads are defined in proto3 (`proto/`, generated into `protopb/`) and seriali
 - **build** (`TopicKeyBuild`, `BuildRequest`) — process publishes the request id here once its build scope (`BuildStrategy`/`BaseURI`) is decided; the build controller reloads the `Request`, triggers the build, and persists the resulting `Build`. Partitioned by request id.
 - **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.
 - **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.
+
+## Repair stage contracts
+
+The repair messages and topic keys are defined for the [repair design](../../../doc/rfc/stovepipe/fixer.md). Controller implementations and runtime registration follow in the implementation layers. Both stages use the existing `consumer.Controller` interface.
+
+- **fix** (`TopicKeyFix`, [`FixRequest`](proto/fix.proto)) — `buildanalysis.Controller.Process` persists an episode's failure or recovery evidence, then publishes its queue, check key, and generation. Processing reloads the episode, evaluates Start/Attach/Defer, persists repair membership and episode assignment, and submits new work through `RepairExecutor.Submit` or signals the existing repair. Partitioned by episode identity.
+- **fixsignal** (`TopicKeyFixSignal`, [`FixSignal`](proto/fixsignal.proto)) — fix processing publishes the repair ID after persisting the returned execution ID, or receiving recovery evidence for an assigned episode. Processing reloads the repair and its failure episodes to call `RepairExecutor.Observe` and check recovery. Partitioned by repair ID, with held deliveries between observations.
+
+Each durable episode update has a distinct publication intent; retries reuse that intent. A recovery update re-enters through `fix`, so processing can resolve either a pending request or an already assigned repair using the current stored ownership.

stovepipe/core/messagequeue/messagequeue.go

@@ -53,6 +53,12 @@ type (
 	// 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
+
+	// FixRequest identifies a persisted failure episode whose assignment needs processing.
+	FixRequest = protopb.FixRequest
+
+	// FixSignal identifies a persisted repair whose execution and recovery need processing.
+	FixSignal = protopb.FixSignal
 )
 
 // marshalOpts keeps the JSON field names identical to the proto field names

stovepipe/core/messagequeue/messagequeue_test.go

@@ -66,6 +66,45 @@ func TestRecordRoundTrip(t *testing.T) {
 	assert.True(t, proto.Equal(rec, got), "round-tripped Record should equal the original")
 }
 
+func TestFixMessagesRoundTrip(t *testing.T) {
+	for _, tt := range []struct {
+		name    string
+		message proto.Message
+	}{
+		{
+			name: "episode request",
+			message: &FixRequest{
+				QueueName:  "repo/main",
+				Key:        "checks/package:compile",
+				Generation: 9007199254740993,
+			},
+		},
+		{
+			name:    "repair signal",
+			message: &FixSignal{QueueName: "repo/main", Id: "repair-42"},
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			data, err := Marshal(tt.message)
+			require.NoError(t, err)
+			got := tt.message.ProtoReflect().New().Interface()
+			require.NoError(t, Unmarshal(data, got))
+			assert.True(t, proto.Equal(tt.message, got))
+		})
+	}
+}
+
+func TestFixRequestWireFormat(t *testing.T) {
+	message := &FixRequest{QueueName: "repo/main", Key: "check-a", Generation: 9007199254740993}
+	data, err := Marshal(message)
+	require.NoError(t, err)
+	assert.JSONEq(t, `{"queue_name":"repo/main","key":"check-a","generation":"9007199254740993"}`, string(data))
+
+	var got FixRequest
+	require.NoError(t, Unmarshal([]byte(`{"queue_name":"repo/main","key":"check-a","generation":"9007199254740993","future_field":true}`), &got))
+	assert.True(t, proto.Equal(message, &got))
+}
+
 // TestWireFormat locks the protojson encoding decision the contract relies on:
 // snake_case field names (UseProtoNames).
 func TestWireFormat(t *testing.T) {
@@ -80,7 +119,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{}, &Record{}} {
+	for _, m := range []proto.Message{&ProcessRequest{}, &BuildRequest{}, &BuildSignal{}, &Record{}, &FixRequest{}, &FixSignal{}} {
 		keys := TopicKeys(m)
 		require.NotEmpty(t, keys, "message must declare a non-empty topic_keys option")
 		for _, key := range keys {
@@ -93,6 +132,8 @@ func TestTopicKeysBindEveryTopicKey(t *testing.T) {
 		TopicKeyBuild,
 		TopicKeyBuildSignal,
 		TopicKeyRecord,
+		TopicKeyFix,
+		TopicKeyFixSignal,
 	}
 
 	valid := map[string]bool{}

stovepipe/core/messagequeue/proto/BUILD.bazel

@@ -2,6 +2,8 @@ exports_files(
     [
         "build.proto",
         "buildsignal.proto",
+        "fix.proto",
+        "fixsignal.proto",
         "process.proto",
         "record.proto",
     ],

stovepipe/core/messagequeue/proto/fix.proto

@@ -0,0 +1,37 @@
+// 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 = "FixRequestProto";
+option java_package = "com.uber.submitqueue.stovepipe.messagequeue";
+
+// FixRequest references a durable episode request; evidence and assignment are loaded from storage.
+// Publishing requires the episode state to have been persisted first.
+message FixRequest {
+    option (uber.base.messagequeue.topic_keys) = "fix";
+
+    // queue_name identifies the validation queue containing the episode.
+    string queue_name = 1;
+    // key is the opaque check identity within the queue.
+    string key = 2;
+    // generation distinguishes a recurrence after recovery from the previous failure episode.
+    int64 generation = 3;
+}

stovepipe/core/messagequeue/proto/fixsignal.proto

@@ -0,0 +1,35 @@
+// 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 = "FixSignalProto";
+option java_package = "com.uber.submitqueue.stovepipe.messagequeue";
+
+// FixSignal references a durable repair; execution and episode recovery are loaded from storage.
+// Publishing requires the repair and its episode associations to have been persisted first.
+message FixSignal {
+    option (uber.base.messagequeue.topic_keys) = "fixsignal";
+
+    // id is the queue-scoped repair identifier.
+    string id = 1;
+    // queue_name identifies the validation queue containing the repair.
+    string queue_name = 2;
+}

stovepipe/core/messagequeue/protopb/BUILD.bazel

@@ -5,6 +5,8 @@ go_library(
     srcs = [
         "build.pb.go",
         "buildsignal.pb.go",
+        "fix.pb.go",
+        "fixsignal.pb.go",
         "process.pb.go",
         "record.pb.go",
     ],