fix(stovepipe): Track validations in order

doc/rfc/stovepipe/steps/build.md

@@ -1,6 +1,6 @@
 # Build stage
 
-Build dispatch retains the request's build association before publishing status and analysis notices. Retries reuse that association instead of triggering again; an uncertain build create is accepted only when its retained request identity matches.
+Build dispatch retains the request's build association before publishing status and analysis notices. Retries reuse that association instead of triggering again; an uncertain build create is accepted only when its retained request identity matches. This supplies the build ID to the request-scoped tracking job already registered at admission.
 
 `build` triggers the build-runner for the scope `process` already decided and hands the resulting build id to `buildsignal`. See [workflow.md](../workflow.md) for where it sits in the pipeline, and [process.md](process.md) for how the scope it reads (`BuildStrategy`, `BaseURI`) is chosen.
 

doc/rfc/stovepipe/steps/buildanalysis.md

@@ -1,6 +1,6 @@
 # Build issue detection
 
-Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success. The reference host consumes issue-tracking notifications and registers their DLQ recovery alongside analysis.
+Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success. The reference host consumes request-scoped issue-tracking jobs and registers their DLQ recovery alongside analysis.
 
 ## Start and completion notices
 
@@ -12,7 +12,7 @@ Publication errors are returned through the producing stage's retry/DLQ policy.
 
 ## Build issues and polling
 
-The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build issue persistence, polling, and the stop condition for Requests that end without a recorded build result. It accumulates per-build issues and publishes queue-partitioned issue-tracking notifications after persistence; it does not maintain open issues itself. Successful terminal analysis requests final reconciliation only after the Request records that Build as its winning outcome. Terminal detection must return the complete cumulative issue set, including issues returned by earlier calls. Analysis persists that set and retains `build_analysis_completed` before publishing final reconciliation. The first retained completion snapshots the issue count and timestamp; retries reuse that snapshot while still persisting detected issues and publishing the handoff. A changed detector count is logged rather than conflicting with history. Nonterminal polls and terminal passes waiting for the recorded winning outcome add no completion event. Issue tracking applies all accumulated build issues before permitting any absence-based closure; a task-wide failure preserves existing issues across all tasks. See the controller documentation for handoff and deduplication rules.
+The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build issue persistence, polling, and the stop condition for Requests that end without a recorded build result. It accumulates per-build issues; it does not maintain open issues itself. Admission registers one tracking job in request order before build dispatch. That job remains current through refreshes and waits for durable terminal readiness, while detection for newer validations continues independently. Successful terminal analysis requests final reconciliation only after the Request records that Build as its winning outcome. Terminal detection must return the complete cumulative issue set, including issues returned by earlier calls. Analysis persists that set and retains `build_analysis_completed` to authorize final reconciliation independently of completion-message delivery. The first retained completion snapshots the issue count and timestamp; retries reuse that snapshot while still persisting detected issues. A changed detector count is logged rather than conflicting with history. Nonterminal polls and terminal passes waiting for the recorded winning outcome add no completion event. Issue tracking applies all accumulated build issues before permitting any absence-based closure; a task-wide failure preserves existing issues across all tasks. See the controller documentation for handoff and deduplication rules.
 
 Status-polling DLQ recovery does not publish a build-completion notice. The analysis polling loop observes the ended Request through storage instead.
 

doc/rfc/stovepipe/steps/process.md

@@ -21,23 +21,27 @@ For a delivery carrying request id `R`:
 2. If R.State is terminal (superseded / succeeded / failed / cancelled):
    - ack and return (idempotent no-op).
 3. If R.State is processing (strategy already recorded):
-   - re-announce validation start and re-publish R to build (either prior publish may have failed), ack, return.
+   - re-announce validation start and re-publish R to build, ack, return.
 4. R.State is accepted. Load the Queue row Q.
 5. Coalesce: if CompareRequestID(R.Queue, R.ID, Q.latest_request_id) < 0:
    - a newer head exists -> mark R superseded, ack, return. (No slot consumed.)
 6. R is the latest head. Gate: if Q.in_flight_count >= max_concurrent (from queue config; see below):
    - defer (hold the delivery) -> re-check on redelivery until the slot frees (admit) or a newer head supersedes it. See [Waiting for a slot](#waiting-for-a-slot).
 7. Admit R:
    a. Derive build strategy + baseline (see "Build-strategy decision").
-   b. CAS the Request: accepted -> processing, persist build_strategy + base_uri.
+   b. Register the tracking job, then CAS the Queue version to revalidate this head.
+      - version conflict -> hold for immediate redelivery, rechecking the head and gate from step 1.
+   c. CAS the Request: accepted -> processing, persist build_strategy + base_uri.
       - lost to another writer (no longer accepted) -> ack; that writer owns the outcome and nothing is counted.
-   c. CAS the Queue row: in_flight_count += 1, reloading and retrying on a version conflict without re-gating.
-   d. Announce validation start on the hook topic (see "Hooks").
-   e. Publish R to build.
-   f. ack.
+   d. CAS the Queue row: in_flight_count += 1, reloading and retrying on a version conflict without re-gating.
+   e. Announce validation start on the hook topic (see "Hooks").
+   f. Publish R to build.
+   g. ack.
 ```
 
-Step 5 runs regardless of the gate: an intermediate head is superseded on sight (even mid-validation), because superseding consumes no slot.
+Step 5 runs before the gate: an accepted intermediate head is superseded even when all build slots are occupied, because superseding consumes no slot.
+
+Step 7b reserves the request's place in [issue tracking](../../../../stovepipe/controller/issuetrack/README.md#validation-order) before its build can start. The Queue-version CAS checks that no newer head overtook the registration. If the request is superseded before admission, its tracking job is acknowledged without updating issues.
 
 ## Build-strategy decision
 
@@ -149,7 +153,7 @@ The event type is `validation.repository.started`. Its payload names the Queue a
 Published after the admit CAS and before the publish to `build`:
 
 ```
-CAS accepted -> processing → publish HookEvent → publish to build → ack
+register tracking → validate Queue version → CAS accepted -> processing → publish HookEvent → publish to build → ack
 ```
 
 After the CAS because the payload names the Request rather than snapshotting it, and the two facts a start event exists to carry — the scope it chose and the baseline it builds on — are written by that very CAS. A hook that reloads the Request must not find it still `accepted` with neither set. Before the build publish because the announce is the cheaper of the two to retry: a failed announce leaves nothing downstream to undo, whereas announcing after the build publish would make a failed announce force the redelivery to re-publish a build that was already accepted.
@@ -160,8 +164,8 @@ Only an admit announces. A Request that coalescing supersedes never reaches step
 
 Every branch is safe under redelivery:
 
-- **accepted, no strategy** → full admit path. Nothing has been written yet: a crash before the request CAS leaves no trace, and redelivery re-runs the gate and the admit from scratch.
-- **processing** → re-announce the start event, re-publish to `build`, ack. The `build` consumer is keyed on the request id and idempotent, so a duplicate publish is harmless, and the start event's id is derived from the transition rather than the clock, so a re-announce carries the id the first attempt would have and consumers dedupe on it. Re-announcing here is what makes the event at-least-once rather than at-most-once: this is the only branch a redelivery takes once `processing` is durable, so an admit that failed after the state write would otherwise lose the event for good.
+- **accepted, no strategy** → full admit path. A crash before the request CAS may leave a tracking registration, but no admitted request. Redelivery re-runs the gate and admission using the same registration identity.
+- **processing** → re-announce the start event, re-publish to `build`, ack. Tracking registration already succeeded before the processing transition, so it is not repeated. Stable message IDs make repeated hook and build publications idempotent. Re-announcing repairs an admission that persisted the state but failed before publication.
 - **terminal** (superseded / recorded) → ack, no-op.
 - **deferred (waiting for slot)** → no state or count change; pure deferral (re-enters when the held delivery comes due).
 

platform/extension/messagequeue/README.md

@@ -72,6 +72,8 @@ See `subscription_config.go` for all fields and defaults.
 
 `Retry.MaxAttempts` uses zero to mean unlimited attempts. `ExtendedSubscriptionConfig` provides 10 attempts with a longer capped exponential backoff for idempotent stages that should tolerate transient dependency failures before dead-lettering. `DLQSubscriptionConfig` selects unlimited attempts and disables a second-level DLQ so reconciliation messages remain retryable until they converge or an operator removes them.
 
+`SingleMessageSubscriptionConfig` uses extended retries and fetches at most one message per poll. Use it when holding a message must prevent prefetching later messages in its partition.
+
 ## Usage
 
 ```go

platform/extension/messagequeue/subscription_config.go

@@ -113,6 +113,14 @@ func ExtendedSubscriptionConfig(subscriberName, consumerGroup string) Subscripti
 	return config
 }
 
+// SingleMessageSubscriptionConfig uses extended retries and fetches at most one message per poll.
+// Use it when holding a message must prevent prefetching later messages in its partition.
+func SingleMessageSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig {
+	config := ExtendedSubscriptionConfig(subscriberName, consumerGroup)
+	config.BatchSize = 1
+	return config
+}
+
 // DefaultSubscriptionConfig returns a SubscriptionConfig with sensible defaults.
 func DefaultSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig {
 	return SubscriptionConfig{

platform/extension/messagequeue/subscription_config_test.go

@@ -97,6 +97,13 @@ func TestExtendedSubscriptionConfig(t *testing.T) {
 	assert.True(t, config.DLQ.Enabled)
 }
 
+func TestSingleMessageSubscriptionConfig(t *testing.T) {
+	expected := ExtendedSubscriptionConfig("worker-1", "consumer-1")
+	expected.BatchSize = 1
+
+	assert.Equal(t, expected, SingleMessageSubscriptionConfig("worker-1", "consumer-1"))
+}
+
 func TestSubscriptionConfig_DifferentConsumerGroups(t *testing.T) {
 	// Test that different consumer groups get independent configs
 	tests := []struct {

service/stovepipe/server/main.go

@@ -417,7 +417,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	analysisController := buildanalysis.NewController(logger, scope, store, materializer, detectorFactory, registry, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
+	analysisController := buildanalysis.NewController(logger, scope, store, materializer, detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis")
 	if err := c.Register(analysisController); err != nil {
 		return count, fmt.Errorf("failed to register buildanalysis controller: %w", err)
 	}
@@ -598,7 +598,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 			Key:          stovepipemq.TopicKeyIssueTrack,
 			Name:         "issuetrack",
 			Queue:        q,
-			Subscription: extqueue.ExtendedSubscriptionConfig(subscriberName, "stovepipe-issuetrack"),
+			Subscription: extqueue.SingleMessageSubscriptionConfig(subscriberName, "stovepipe-issuetrack"),
 		},
 		{
 			Key:          dlq.TopicKey(stovepipemq.TopicKeyIssueTrack),

service/stovepipe/server/main_test.go

@@ -112,6 +112,7 @@ func TestIssueTrackingRegistration(t *testing.T) {
 
 	cfg, ok := registry.SubscriptionConfig(stovepipemq.TopicKeyIssueTrack, "stovepipe-issuetrack")
 	require.True(t, ok)
+	assert.Equal(t, 1, cfg.BatchSize)
 	assert.True(t, cfg.DLQ.Enabled)
 	assert.Equal(t, "_dlq", cfg.DLQ.TopicSuffix)
 }

stovepipe/controller/build/build_analysis_test.go

@@ -143,7 +143,7 @@ func TestBuildAnalysisPipeline(t *testing.T) {
 	)
 
 	analysisController := buildanalysis.NewController(zap.NewNop().Sugar(), tally.NoopScope,
-		staticStorageFactory{store: m.store}, requestlog.NewMaterializer(tally.NoopScope), detectorFactory, buildController.registry, stovepipemq.TopicKeyBuildAnalysis, "analysis")
+		staticStorageFactory{store: m.store}, requestlog.NewMaterializer(tally.NoopScope), detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "analysis")
 	statusController := buildsignal.NewController(zap.NewNop().Sugar(), tally.NoopScope,
 		staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, buildController.registry,
 		stovepipemq.TopicKeyBuildSignal, "signal")
@@ -184,19 +184,7 @@ func TestBuildAnalysisPipeline(t *testing.T) {
 	assert.True(t, persisted[b])
 	assert.True(t, persisted[a])
 
-	require.Len(t, published["issuetrack"], 4)
-	for i, msg := range published["issuetrack"] {
-		assert.Equal(t, testQueue, msg.PartitionKey)
-		assert.Equal(t, testQueue, msg.Tenant)
-		notification := &stovepipemq.IssueTrack{}
-		require.NoError(t, stovepipemq.Unmarshal(msg.Payload, notification))
-		assert.Equal(t, testBuildID, notification.GetBuildId())
-		assert.Equal(t, testQueue, notification.GetQueueName())
-		assert.Equal(t, i > 1, notification.GetFinalize())
-	}
-	assert.NotEqual(t, published["issuetrack"][0].ID, published["issuetrack"][1].ID)
-	assert.NotEqual(t, published["issuetrack"][1].ID, published["issuetrack"][2].ID)
-	assert.Equal(t, published["issuetrack"][2].ID, published["issuetrack"][3].ID)
+	require.Empty(t, published["issuetrack"])
 	require.Len(t, history, 1)
 	for _, log := range history {
 		assert.Equal(t, entity.RequestEventBuildAnalysisCompleted, log.Event)

stovepipe/controller/buildanalysis/BUILD.bazel

@@ -12,7 +12,6 @@ go_library(
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
-        "//platform/publish:go_default_library",
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
@@ -38,9 +37,7 @@ go_test(
         "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
         "//platform/errs:go_default_library",
-        "//platform/extension/messagequeue/mock:go_default_library",
         "//platform/metrics:go_default_library",
-        "//platform/publish:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/core/requestlog:go_default_library",
         "//stovepipe/core/requestlog/mock:go_default_library",