feat(stovepipe): Announce validation start events (#652)

## Summary
**What**:
- Announce that validation of a commit has begun as soon as the pipeline
admits it, alongside the existing announcements for validation outcomes.
- Repeat that announcement when an already-admitted request is
redelivered, reusing the original event identity so consumers can
discard the duplicate.

**Why**:
- Let downstream systems track a validation from start to finish,
instead of only learning it happened once a verdict exists.
- Keep the announcement from being lost when the pipeline fails after
durably admitting the request.

## Test Plan
- [x] Add unit tests.

## Revert Plan
- Revert this PR. Nothing consumes the start event yet, so publishing
just stops with no downstream impact.

## Issues
-
[CODEM-466](https://linear.app/uber/issue/CODEM-466/emit-events-when-we-start-a-new-validation-range)

Author: prathshenoy

doc/rfc/stovepipe/steps/process.md

@@ -21,7 +21,7 @@ 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-publish R to build (the prior publish may have failed), ack, return.
+   - re-announce validation start and re-publish R to build (either prior publish may have failed), 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.)
@@ -31,8 +31,9 @@ For a delivery carrying request id `R`:
    a. Derive build strategy + baseline (see "Build-strategy decision").
    b. CAS the Queue row: in_flight_count += 1.
    c. CAS the Request: accepted -> processing, persist build_strategy + base_uri.
-   d. Publish R to build.
-   e. ack.
+   d. Announce validation start on the hook topic (see "Hooks").
+   e. Publish R to build.
+   f. ack.
 ```
 
 Step 5 runs regardless of the gate: an intermediate head is superseded on sight (even mid-validation), because superseding consumes no slot.
@@ -138,12 +139,28 @@ A, D, F each get a full cycle; B, C, E end `superseded`. No intermediate is vali
 - A newer head does **not** preempt an in-flight validation.
 - Deferred messages are **not** failed or dead-lettered — they wait for the gate (see [Waiting for a slot](#waiting-for-a-slot)).
 
+## Hooks
+
+Admitting a request is when the rest of the company can learn "validation of this commit has begun". `process` publishes that as a `HookEvent` on Stovepipe's durable `hook` topic — the same seam `record` uses to announce the outcome. The mechanics (envelope, delivery promise, per-domain dispatcher stage, `hook_dlq`) are settled in [hook-framework.md](../../hook-framework.md); this section covers only what admitting has to decide.
+
+The event type is `validation.repository.started`. Its payload names the Queue and the Request and nothing else, exactly as the terminal events in [record.md](record.md#hooks) do: a hook resolves the commit, the chosen strategy, and the baseline from the request store rather than reading a snapshot off the wire.
+
+Published after the admit CAS and before the publish to `build`:
+
+```
+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.
+
+Only an admit announces. A Request that coalescing supersedes never reaches step 7, so it produces no start event — and a start event is not a promise that a verdict follows, since an admitted Request can still be cancelled or driven to a fail-closed outcome. Consumers pairing a start with an end must tolerate a start that never gets one.
+
 ## Idempotency and at-least-once delivery
 
 Every branch is safe under redelivery:
 
 - **accepted, no strategy** → full admit path. On a crash after incrementing `in_flight_count` but before persisting `processing`, redelivery re-reads `accepted` and re-runs; the increment re-applies only if the count CAS hasn't already moved (see integrity below).
-- **processing** → re-publish to `build` and ack. The `build` consumer is keyed on the request id and idempotent, so a duplicate publish is harmless.
+- **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.
 - **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).
 

doc/rfc/stovepipe/workflow.md

@@ -55,10 +55,10 @@ The ref is a *cache* of the last-green URI, not a second record of greenness. It
 |---|---|
 | **SourceControl** | Resolve a Queue name to its current head URI; answer ancestry/comparison questions between two URIs (is the new head a fast-forward descendant of the last green, or was history rewritten?); enumerate commits in a range; advance the Queue's **promotion ref** to a commit. The sole owner of URI semantics, including which refs a Queue name resolves to. |
 | **build-runner** | Build a scope at a URI (optionally relative to a baseline URI), returning pass/fail and the target graph. See [build-runner.md](../submitqueue/build-runner.md). |
-| **Hooks** | Deliver Stovepipe's greenness events to downstream systems — "this URI / this project is now green (or not green)". Fire-and-forget notification, decoupled so Stovepipe does not know or care who consumes the event. The shared cross-domain hook seam rather than a Stovepipe-specific extension. See [hook-framework.md](../hook-framework.md). |
+| **Hooks** | Deliver Stovepipe's validation events to downstream systems — "validation of this URI has begun", "this URI / this project is now green (or not green)". Fire-and-forget notification, decoupled so Stovepipe does not know or care who consumes the event. The shared cross-domain hook seam rather than a Stovepipe-specific extension. See [hook-framework.md](../hook-framework.md). |
 | **Storage** | Persist Queues (incl. last-green URI), Requests, build records, and per-URI / per-project greenness. Key/value-shaped per the extension-design rules in [AGENTS.md](../../../AGENTS.md). |
 
-Hooks are the notification boundary. When a validation fact is recorded — whole-repo green/not-green, or later a project green/not-green — the event reaches deployment systems, dashboards, and developer tooling without any of them polling Stovepipe's store, and each environment can route it to its own downstream (a deploy gate, a Slack notifier, an event bus) without changing the pipeline. The mechanism is the cross-domain hook framework rather than a call out of the recording stage: `record` publishes a `HookEvent` to Stovepipe's `hook` topic, and a dispatcher stage consumes it and invokes the wired hooks, so a slow or failing downstream cannot add latency to the pipeline. Both halves exist; what a deployment supplies is the hooks themselves, since the example server resolves every event to `noop`. See [record.md](steps/record.md#hooks) for the fact-to-event mapping.
+Hooks are the notification boundary. When validation of a commit begins, and when a validation fact is recorded — whole-repo green/not-green, or later a project green/not-green — the event reaches deployment systems, dashboards, and developer tooling without any of them polling Stovepipe's store, and each environment can route it to its own downstream (a deploy gate, a Slack notifier, an event bus) without changing the pipeline. The mechanism is the cross-domain hook framework rather than a call out of the pipeline stages: `process` and `record` publish a `HookEvent` to Stovepipe's `hook` topic, and a dispatcher stage consumes it and invokes the wired hooks, so a slow or failing downstream cannot add latency to the pipeline. Both halves exist; what a deployment supplies is the hooks themselves, since the example server resolves every event to `noop`. See [process.md](steps/process.md#hooks) for the start event and [record.md](steps/record.md#hooks) for the fact-to-event mapping.
 
 ## Workflow
 
@@ -74,9 +74,9 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe
                                    └───────────────┬──────────────┘
                                                    │ RequestID
                                                    ▼
-                                   ┌──────────────────────────────┐
-                                   │ process                      │
-                                   │ Ask SourceControl: is head a │
+                                   ┌──────────────────────────────┐   Hooks
+                                   │ process                      │┄┄┄┄┄►  "validation
+                                   │ Ask SourceControl: is head a │        started"
                                    │ descendant of last-green?    │
                                    │  → incremental since green   │
                                    │ else (history rewrite)       │
@@ -132,7 +132,7 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe
 ### Phase 1 — whole-repo greenness
 
 1. **ingest** — invoked by the external poller with a **Queue name**. It asks `SourceControl` for that Queue's current head URI, mints a Request namespaced by the Queue, persists it with no recorded greenness yet, and dedups on `(Queue, head URI)` so a re-reported head is processed once. It publishes the RequestID onward.
-2. **process** — decides build strategy (incremental since last-green vs full monorepo), gates concurrent work per Queue, coalesces backlog to the latest head, and publishes to `build`. See [process.md](steps/process.md).
+2. **process** — decides build strategy (incremental since last-green vs full monorepo), gates concurrent work per Queue, coalesces backlog to the latest head, publishes a **hook event** announcing that validation of the commit has begun, and publishes to `build`. See [process.md](steps/process.md).
 3. **build** — runs the build-runner for the chosen scope. A flag derived from `process` decides whether to build relative to the last-green **baseline URI** (incremental) or from scratch (full). It records a build and publishes the BuildID.
 4. **buildsignal** — records the build's status and target graph when the build completes, then releases the Queue's `in_flight_count` slot, projects the terminal status onto the Request (`succeeded` / `failed` / `cancelled`), and publishes the RequestID to `record`.
 5. **record** — writes the whole-repo greenness for the head URI (`0` green / `1` broken to start), derived from the Request's build outcome. On green it advances the Queue's **last-green URI** so the next `process` can build incrementally from here, and asks `SourceControl` to advance the Queue's **promotion ref** to the same commit (see [Promotion ref](#promotion-ref--the-last-green-commit-by-name)). It publishes a **hook event** for the green/not-green transition, then fans out into Phase 2. The Queue's `in_flight_count` was already released by `buildsignal` when the build went terminal.
@@ -150,7 +150,7 @@ The pipeline runs in two phases against the same Request. **Phase 1** establishe
 | Controller | In | Out | One-line role |
 |---|---|---|---|
 | **ingest** | Queue name (from poller) | process | Resolve head URI via SourceControl, mint Request, persist (no greenness), dedup on `(Queue, head URI)` |
-| **process** | RequestID | build | Build strategy, concurrency gate, backlog coalescing → [process.md](steps/process.md) |
+| **process** | RequestID | build, hook topic | Build strategy, concurrency gate, backlog coalescing; announce validation start on admit → [process.md](steps/process.md) |
 | **build** | RequestID | buildsignal | Run the build-runner for the chosen scope; baseline = last-green URI iff incremental |
 | **buildsignal** | BuildID | record (P1), record (P2) | Record build status + target graph; release `in_flight_count`; project the outcome onto the Request; signal completion |
 | **record** | RequestID | analyze (P1→P2), hook topic | Write greenness; on whole-repo green advance last-green URI and the promotion ref; publish the hook event |

stovepipe/controller/process/BUILD.bazel

@@ -6,10 +6,13 @@ go_library(
     importpath = "github.com/uber/submitqueue/stovepipe/controller/process",
     visibility = ["//visibility:public"],
     deps = [
+        "//api/base/hook:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/errs:go_default_library",
+        "//platform/hook:go_default_library",
         "//platform/metrics:go_default_library",
         "//platform/publish:go_default_library",
+        "//stovepipe/core/hookevent:go_default_library",
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
@@ -26,12 +29,14 @@ go_test(
     srcs = ["process_test.go"],
     embed = [":go_default_library"],
     deps = [
+        "//api/base/hook:go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//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",
+        "//stovepipe/core/hookevent:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/queueconfig/default:go_default_library",

stovepipe/controller/process/process.go

@@ -24,10 +24,13 @@ import (
 	"fmt"
 
 	"github.com/uber-go/tally"
+	basehook "github.com/uber/submitqueue/api/base/hook"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/errs"
+	platformhook "github.com/uber/submitqueue/platform/hook"
 	"github.com/uber/submitqueue/platform/metrics"
 	"github.com/uber/submitqueue/platform/publish"
+	"github.com/uber/submitqueue/stovepipe/core/hookevent"
 	"github.com/uber/submitqueue/stovepipe/core/loader"
 	stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
 	"github.com/uber/submitqueue/stovepipe/entity"
@@ -113,6 +116,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 
 	switch request.State {
 	case entity.RequestStateProcessing:
+		// Announce here as well as at admit: this is the only path a redelivery
+		// takes once the transition is durable, so an admit that failed after
+		// persisting would otherwise lose the start event for good. The event id
+		// is derived from the transition, so a repeat carries the id the first
+		// attempt would have and a consumer deduplicates on it.
+		if err := c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryStarted(request)); err != nil {
+			return err
+		}
 		if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil {
 			metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1, metrics.TagsFromContext(ctx)...)
 			return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err)
@@ -252,6 +263,10 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage,
 		return nil
 	}
 
+	if err := c.publishHookEvent(ctx, request, hookevent.NewValidationRepositoryStarted(request)); err != nil {
+		return err
+	}
+
 	if err := c.publishBuild(ctx, request.ID, request.Queue); err != nil {
 		metrics.NamedCounter(c.metricsScope, _opName, "publish_errors", 1, metrics.TagsFromContext(ctx)...)
 		return fmt.Errorf("failed to publish request %s to build: %w", request.ID, err)
@@ -474,6 +489,33 @@ func (c *Controller) publishBuild(ctx context.Context, id, queue string) error {
 	return nil
 }
 
+// publishHookEvent announces a lifecycle transition on the hook topic.
+//
+// Published only once the transition is durable: the payload names the request
+// rather than snapshotting it, so a hook that reloads it must not find a request
+// whose strategy and baseline are still unwritten.
+//
+// Partitioning by request id matches the process topic's own, carrying
+// per-request ordering across the seam.
+func (c *Controller) publishHookEvent(ctx context.Context, request entity.Request, event *basehook.HookEvent) error {
+	if err := platformhook.Publish(ctx, c.registry, event, request.ID); err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "hook_errors", 1, metrics.TagsFromContext(ctx)...)
+		return fmt.Errorf("failed to announce %s for request %s: %w", event.GetType(), request.ID, err)
+	}
+
+	metrics.NamedCounter(c.metricsScope, _opName, "hook_events_published", 1,
+		metrics.TagsFromContext(ctx, metrics.NewTag("event_type", event.GetType()))...,
+	)
+	c.logger.Debugw("announced validation event",
+		"queue", request.Queue,
+		"request_id", request.ID,
+		"uri", request.URI,
+		"event_type", event.GetType(),
+		"event_id", event.GetId(),
+	)
+	return nil
+}
+
 // Name returns the controller name for logging and metrics.
 func (c *Controller) Name() string {
 	return "process"