From bb096dbd9b697b3f860f532fd5954cc6f78bbd54 Mon Sep 17 00:00:00 2001
From: mnoah1 <mnoah1@uber.com>
Date: Mon, 31 Aug 2026 22:19:04 +0000
Subject: [PATCH 1/5] feat(stovepipe): add request state log recorder
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Summary:
Intent:
- Provide one idempotent path for retaining durable request state transitions.
- Keep the initial rollout limited to state entries while lifecycle events and metadata conventions remain deferred.

Changes:
- Derive stable occurrence identities from the queue, request, and request version.
- Treat identical duplicate writes as success and surface conflicting retained content.
- Report bounded recorder metrics and cover validation, retries, and conflicts.

---

<sub>Generated by the 🪄 [pr-create](https://sg.uberinternal.com/code.uber.internal/uber-code/devexp-agent-marketplace/-/blob/claude-code/plugins/dev/uber-dev/skills/pr-create/SKILL.md) skill in devexp-agent-marketplace</sub>
---
 Makefile                                      |   2 +-
 doc/rfc/stovepipe/request-log.md              |  12 +-
 stovepipe/core/requestlog/BUILD.bazel         |  29 +++
 stovepipe/core/requestlog/mock/BUILD.bazel    |  13 ++
 .../core/requestlog/mock/recorder_mock.go     |  57 ++++++
 stovepipe/core/requestlog/recorder.go         | 140 +++++++++++++
 stovepipe/core/requestlog/recorder_test.go    | 190 ++++++++++++++++++
 stovepipe/entity/request_log.go               |   2 +-
 8 files changed, 437 insertions(+), 8 deletions(-)
 create mode 100644 stovepipe/core/requestlog/BUILD.bazel
 create mode 100644 stovepipe/core/requestlog/mock/BUILD.bazel
 create mode 100644 stovepipe/core/requestlog/mock/recorder_mock.go
 create mode 100644 stovepipe/core/requestlog/recorder.go
 create mode 100644 stovepipe/core/requestlog/recorder_test.go

diff --git a/Makefile b/Makefile
index 8b4f2e2d..25e5097e 100644
--- a/Makefile
+++ b/Makefile
@@ -579,7 +579,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service
 
 mocks: ## Generate mock files using mockgen
 	@echo "Generating mocks..."
-	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
+	@$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./submitqueue/extension/mergechecker/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/...
 	@echo "Mocks generated successfully!"
 
 proto: ## Generate protobuf files from .proto definitions
diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md
index 32f11274..0b41975f 100644
--- a/doc/rfc/stovepipe/request-log.md
+++ b/doc/rfc/stovepipe/request-log.md
@@ -79,7 +79,7 @@ type RequestLog struct {
     Queue string
     // RequestID identifies the request whose log contains this record.
     RequestID string
-    // TimestampMs is the durable occurrence time in Unix milliseconds.
+    // TimestampMs is when the occurrence was first retained, in Unix milliseconds.
     TimestampMs int64
     // State is the durable request state recorded by a state entry. It is unset on an event entry.
     State RequestState
@@ -154,7 +154,7 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
 | Build finished | Request ID, event kind, and build ID |
 | Validation fact recorded | Request ID, event kind, and whole-repository fact identity |
 
-The recorder calls `RequestLogStore.Create`. If the ID already exists, it loads the stored record and compares every semantic field. Identical content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten.
+The controller passes the recorder the `RequestLogStore` from the same queue-scoped storage aggregate used for the source write. The recorder assigns the current time immediately before the first insertion attempt and calls `Create`. If the ID already exists, it loads the stored record and compares every domain field. The first successfully retained timestamp is authoritative and is not compared with a later retry's newly sampled time. Identical domain content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten.
 
 ## Storage Contract
 
@@ -188,11 +188,11 @@ Request-log durability is part of completing a pipeline transition. The source w
 
 For a Request transition, the controller:
 
-1. builds an immutable updated copy with transition context and `StateChangedAtMs`;
+1. builds an immutable updated copy for the state transition;
 2. computes `newVersion = oldVersion + 1`;
 3. calls `RequestStore.Update(updated, oldVersion, newVersion)`;
 4. assigns the in-memory version only after the store succeeds;
-5. asks the recorder to create the log record from durable Request data;
+5. asks the recorder to create the log record from the durable Request and the bounded context still owned by that stage;
 6. publishes the downstream handoff.
 
 Request creation, Build changes, and fact creation use the same source-write, log-write, dependent-publish ordering. A request-log outage can leave a source update visible, but it cannot allow dependent processing to move past an unrecorded transition.
@@ -224,7 +224,7 @@ This is rollout work, not deferred cleanup: a mandatory request log without a du
 
 Rollout therefore:
 
-1. deploys source timestamp and provenance fields, request-log storage, recorder, and readers;
+1. deploys request-log storage, the recorder, and readers;
 2. enables writers and verifies every repair path stage by stage;
 3. enables the public API after every writer and repair path is active.
 
@@ -246,7 +246,7 @@ Identifiers, outcome reasons, and reasonable per-request build counts have expli
 
 Contract tests cover required-field validation, stable IDs, idempotent create/reload, conflict detection, queue binding, deterministic ordering, equal-timestamp tie breaking, and empty histories.
 
-Writer tests cover source success followed by log failure, redelivery with the log record absent or present, CAS loss, conflicting terminal writers, downstream publish failure, stable timestamps, and controller-owned version arithmetic. End-to-end tests cover successful, failed, cancelled, superseded, and fail-closed paths plus idempotent redelivery.
+Writer tests cover source success followed by log failure, redelivery with the log record absent or present, CAS loss, conflicting terminal writers, downstream publish failure, first-insert timestamp reuse, and controller-owned version arithmetic. End-to-end tests cover successful, failed, cancelled, superseded, and fail-closed paths plus idempotent redelivery.
 
 Tests reconstruct the latest Request state from state entries by request version and compare it with `RequestStore.Get`. They separately verify that only a durable fact produces green or broken.
 
diff --git a/stovepipe/core/requestlog/BUILD.bazel b/stovepipe/core/requestlog/BUILD.bazel
new file mode 100644
index 00000000..c4798183
--- /dev/null
+++ b/stovepipe/core/requestlog/BUILD.bazel
@@ -0,0 +1,29 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["recorder.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/core/requestlog",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//platform/metrics:go_default_library",
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["recorder_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "//stovepipe/extension/storage/mock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)
diff --git a/stovepipe/core/requestlog/mock/BUILD.bazel b/stovepipe/core/requestlog/mock/BUILD.bazel
new file mode 100644
index 00000000..7a161d71
--- /dev/null
+++ b/stovepipe/core/requestlog/mock/BUILD.bazel
@@ -0,0 +1,13 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["recorder_mock.go"],
+    importpath = "github.com/uber/submitqueue/stovepipe/core/requestlog/mock",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//stovepipe/entity:go_default_library",
+        "//stovepipe/extension/storage:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)
diff --git a/stovepipe/core/requestlog/mock/recorder_mock.go b/stovepipe/core/requestlog/mock/recorder_mock.go
new file mode 100644
index 00000000..aeab397a
--- /dev/null
+++ b/stovepipe/core/requestlog/mock/recorder_mock.go
@@ -0,0 +1,57 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: recorder.go
+//
+// Generated by this command:
+//
+//	mockgen -source=recorder.go -destination=mock/recorder_mock.go -package=mock
+//
+
+// Package mock is a generated GoMock package.
+package mock
+
+import (
+	context "context"
+	reflect "reflect"
+
+	entity "github.com/uber/submitqueue/stovepipe/entity"
+	storage "github.com/uber/submitqueue/stovepipe/extension/storage"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockRecorder is a mock of Recorder interface.
+type MockRecorder struct {
+	ctrl     *gomock.Controller
+	recorder *MockRecorderMockRecorder
+	isgomock struct{}
+}
+
+// MockRecorderMockRecorder is the mock recorder for MockRecorder.
+type MockRecorderMockRecorder struct {
+	mock *MockRecorder
+}
+
+// NewMockRecorder creates a new mock instance.
+func NewMockRecorder(ctrl *gomock.Controller) *MockRecorder {
+	mock := &MockRecorder{ctrl: ctrl}
+	mock.recorder = &MockRecorderMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockRecorder) EXPECT() *MockRecorderMockRecorder {
+	return m.recorder
+}
+
+// RecordRequestState mocks base method.
+func (m *MockRecorder) RecordRequestState(arg0 context.Context, arg1 storage.RequestLogStore, arg2 entity.Request, arg3 entity.RequestOutcomeReason) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RecordRequestState", arg0, arg1, arg2, arg3)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RecordRequestState indicates an expected call of RecordRequestState.
+func (mr *MockRecorderMockRecorder) RecordRequestState(arg0, arg1, arg2, arg3 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordRequestState", reflect.TypeOf((*MockRecorder)(nil).RecordRequestState), arg0, arg1, arg2, arg3)
+}
diff --git a/stovepipe/core/requestlog/recorder.go b/stovepipe/core/requestlog/recorder.go
new file mode 100644
index 00000000..212b635f
--- /dev/null
+++ b/stovepipe/core/requestlog/recorder.go
@@ -0,0 +1,140 @@
+// Copyright (c) 2025 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 requestlog retains the request occurrences exposed by Stovepipe's history API.
+package requestlog
+
+//go:generate mockgen -source=recorder.go -destination=mock/recorder_mock.go -package=mock
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/binary"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"maps"
+	"reflect"
+	"strconv"
+	"time"
+
+	"github.com/uber-go/tally"
+
+	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+)
+
+const (
+	_occurrenceKindState = "state"
+)
+
+// Recorder retains idempotent request-state occurrences.
+type Recorder interface {
+	// RecordRequestState retains the request's current durable state and version.
+	RecordRequestState(context.Context, storage.RequestLogStore, entity.Request, entity.RequestOutcomeReason) error
+}
+
+type recorder struct {
+	scope tally.Scope
+	now   func() time.Time
+}
+
+// NewRecorder creates a request-log recorder.
+func NewRecorder(scope tally.Scope) Recorder {
+	return &recorder{
+		scope: scope.SubScope("request_log_recorder"),
+		now:   time.Now,
+	}
+}
+
+func (r *recorder) RecordRequestState(ctx context.Context, store storage.RequestLogStore, request entity.Request, outcomeReason entity.RequestOutcomeReason) error {
+	log := entity.RequestLog{
+		ID:             occurrenceID(request.Queue, request.ID, _occurrenceKindState, strconv.FormatInt(int64(request.Version), 10)),
+		Queue:          request.Queue,
+		RequestID:      request.ID,
+		State:          request.State,
+		RequestVersion: request.Version,
+		OutcomeReason:  outcomeReason,
+	}
+	return r.record(ctx, store, log)
+}
+
+func (r *recorder) record(ctx context.Context, store storage.RequestLogStore, log entity.RequestLog) error {
+	log.TimestampMs = r.now().UnixMilli()
+	tag := occurrenceTag(log)
+
+	if err := log.Validate(); err != nil {
+		metrics.NamedCounter(r.scope, "record", "validation_failure", 1, tag)
+		return fmt.Errorf("invalid request log occurrence: %w", err)
+	}
+
+	if err := store.Create(ctx, log); err == nil {
+		metrics.NamedCounter(r.scope, "record", "created", 1, tag)
+		return nil
+	} else if !errors.Is(err, storage.ErrAlreadyExists) {
+		metrics.NamedCounter(r.scope, "record", "storage_failure", 1, tag)
+		return fmt.Errorf("failed to create request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
+	}
+
+	stored, err := store.Get(ctx, log.RequestID, log.ID)
+	if err != nil {
+		metrics.NamedCounter(r.scope, "record", "storage_failure", 1, tag)
+		return fmt.Errorf("failed to reconcile request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
+	}
+	if !sameOccurrence(stored, log) {
+		metrics.NamedCounter(r.scope, "record", "conflict", 1, tag)
+		return fmt.Errorf("request log conflicts with retained occurrence request_id=%q log_id=%q", log.RequestID, log.ID)
+	}
+
+	metrics.NamedCounter(r.scope, "record", "identical_existing", 1, tag)
+	return nil
+}
+
+func occurrenceID(queue, requestID string, identity ...string) string {
+	hash := sha256.New()
+	parts := append([]string{queue, requestID}, identity...)
+	var size [8]byte
+	for _, part := range parts {
+		binary.BigEndian.PutUint64(size[:], uint64(len(part)))
+		_, _ = hash.Write(size[:])
+		_, _ = hash.Write([]byte(part))
+	}
+	return "log/" + hex.EncodeToString(hash.Sum(nil))
+}
+
+func sameOccurrence(stored, candidate entity.RequestLog) bool {
+	// The first successful insert owns display time; retries compare only the occurrence's domain content.
+	storedMetadata := stored.Metadata
+	candidateMetadata := candidate.Metadata
+	stored.Metadata = nil
+	candidate.Metadata = nil
+	stored.TimestampMs = 0
+	candidate.TimestampMs = 0
+	return reflect.DeepEqual(stored, candidate) && maps.Equal(storedMetadata, candidateMetadata)
+}
+
+func occurrenceTag(log entity.RequestLog) metrics.Tag {
+	value := "invalid"
+	switch log.State {
+	case entity.RequestStateAccepted,
+		entity.RequestStateProcessing,
+		entity.RequestStateSuperseded,
+		entity.RequestStateSucceeded,
+		entity.RequestStateFailed,
+		entity.RequestStateCancelled:
+		value = string(log.State)
+	}
+	return metrics.NewTag("occurrence", value)
+}
diff --git a/stovepipe/core/requestlog/recorder_test.go b/stovepipe/core/requestlog/recorder_test.go
new file mode 100644
index 00000000..a9ebe8cb
--- /dev/null
+++ b/stovepipe/core/requestlog/recorder_test.go
@@ -0,0 +1,190 @@
+// Copyright (c) 2025 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 requestlog
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/stovepipe/entity"
+	"github.com/uber/submitqueue/stovepipe/extension/storage"
+	storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock"
+	"go.uber.org/mock/gomock"
+)
+
+const (
+	testQueue     = "monorepo/main"
+	testRequestID = "request/monorepo/main/1"
+	testNowMs     = int64(1735689600000)
+)
+
+func newTestRecorder(t *testing.T) (*recorder, *storagemock.MockRequestLogStore) {
+	t.Helper()
+	ctrl := gomock.NewController(t)
+	return &recorder{
+		scope: tally.NoopScope,
+		now:   func() time.Time { return time.UnixMilli(testNowMs) },
+	}, storagemock.NewMockRequestLogStore(ctrl)
+}
+
+func TestRecorderRecordRequestState(t *testing.T) {
+	tests := []struct {
+		name          string
+		state         entity.RequestState
+		outcomeReason entity.RequestOutcomeReason
+		wantErr       bool
+	}{
+		{name: "accepted", state: entity.RequestStateAccepted},
+		{name: "processing", state: entity.RequestStateProcessing},
+		{
+			name:          "superseded",
+			state:         entity.RequestStateSuperseded,
+			outcomeReason: entity.RequestOutcomeReasonSupersededByNewerHead,
+		},
+		{
+			name:          "succeeded",
+			state:         entity.RequestStateSucceeded,
+			outcomeReason: entity.RequestOutcomeReasonBuildSucceeded,
+		},
+		{name: "terminal reason required", state: entity.RequestStateSucceeded, wantErr: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			recorder, store := newTestRecorder(t)
+			request := entity.Request{
+				ID:      testRequestID,
+				Queue:   testQueue,
+				State:   tt.state,
+				Version: 2,
+			}
+			if !tt.wantErr {
+				store.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, entry entity.RequestLog) error {
+					assert.NotEmpty(t, entry.ID)
+					assert.Equal(t, testNowMs, entry.TimestampMs)
+					assert.Equal(t, testQueue, entry.Queue)
+					assert.Equal(t, testRequestID, entry.RequestID)
+					assert.Equal(t, tt.state, entry.State)
+					assert.Equal(t, int32(2), entry.RequestVersion)
+					assert.Equal(t, tt.outcomeReason, entry.OutcomeReason)
+					require.NoError(t, entry.Validate())
+					return nil
+				})
+			}
+
+			err := recorder.RecordRequestState(context.Background(), store, request, tt.outcomeReason)
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestRecorderExistingIdenticalOccurrenceIsSuccess(t *testing.T) {
+	recorder, store := newTestRecorder(t)
+	request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateAccepted, Version: 1}
+
+	var candidate entity.RequestLog
+	store.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, entry entity.RequestLog) error {
+		candidate = entry
+		return storage.ErrAlreadyExists
+	})
+	store.EXPECT().Get(gomock.Any(), testRequestID, gomock.Any()).DoAndReturn(
+		func(_ context.Context, _ string, entryID string) (entity.RequestLog, error) {
+			stored := candidate
+			stored.ID = entryID
+			stored.TimestampMs = candidate.TimestampMs - 1000
+			stored.Metadata = map[string]string{}
+			return stored, nil
+		},
+	)
+
+	require.NoError(t, recorder.RecordRequestState(context.Background(), store, request, entity.RequestOutcomeReasonUnknown))
+}
+
+func TestRecorderExistingConflictingOccurrenceFails(t *testing.T) {
+	recorder, store := newTestRecorder(t)
+	request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateSucceeded, Version: 3}
+
+	var candidate entity.RequestLog
+	store.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, entry entity.RequestLog) error {
+		candidate = entry
+		return storage.ErrAlreadyExists
+	})
+	store.EXPECT().Get(gomock.Any(), testRequestID, gomock.Any()).DoAndReturn(
+		func(context.Context, string, string) (entity.RequestLog, error) {
+			stored := candidate
+			stored.OutcomeReason = entity.RequestOutcomeReasonBuildFailed
+			return stored, nil
+		},
+	)
+
+	require.Error(t, recorder.RecordRequestState(context.Background(), store, request, entity.RequestOutcomeReasonBuildSucceeded))
+}
+
+func TestRecorderStorageFailures(t *testing.T) {
+	tests := []struct {
+		name  string
+		setup func(*storagemock.MockRequestLogStore)
+	}{
+		{
+			name: "create",
+			setup: func(store *storagemock.MockRequestLogStore) {
+				store.EXPECT().Create(gomock.Any(), gomock.Any()).Return(errors.New("create"))
+			},
+		},
+		{
+			name: "reload duplicate",
+			setup: func(store *storagemock.MockRequestLogStore) {
+				store.EXPECT().Create(gomock.Any(), gomock.Any()).Return(storage.ErrAlreadyExists)
+				store.EXPECT().Get(gomock.Any(), testRequestID, gomock.Any()).Return(entity.RequestLog{}, errors.New("get"))
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			recorder, store := newTestRecorder(t)
+			tt.setup(store)
+			err := recorder.RecordRequestState(context.Background(), store, entity.Request{
+				ID: testRequestID, Queue: testQueue, State: entity.RequestStateAccepted, Version: 1,
+			}, entity.RequestOutcomeReasonUnknown)
+			require.Error(t, err)
+		})
+	}
+}
+
+func TestOccurrenceID(t *testing.T) {
+	assert.Equal(t, occurrenceID("queue", "request", "state", "1"), occurrenceID("queue", "request", "state", "1"))
+	assert.NotEqual(t, occurrenceID("queue", "request", "state", "1"), occurrenceID("queue", "request", "state", "2"))
+	assert.NotEqual(t, occurrenceID("queue", "request", "ab", "c"), occurrenceID("queue", "request", "a", "bc"))
+}
+
+func TestSameOccurrenceMetadata(t *testing.T) {
+	base := entity.RequestLog{ID: "log/1", Queue: testQueue, RequestID: testRequestID, State: entity.RequestStateAccepted, RequestVersion: 1}
+	stored := base
+	stored.Metadata = map[string]string{}
+	assert.True(t, sameOccurrence(stored, base))
+
+	stored.Metadata["source"] = "hook"
+	assert.False(t, sameOccurrence(stored, base))
+}
diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go
index 2da89ed2..2bb06671 100644
--- a/stovepipe/entity/request_log.go
+++ b/stovepipe/entity/request_log.go
@@ -60,7 +60,7 @@ type RequestLog struct {
 	Queue string `json:"queue"`
 	// RequestID identifies the request whose log contains this record.
 	RequestID string `json:"request_id"`
-	// TimestampMs is the occurrence time in Unix milliseconds.
+	// TimestampMs is when the occurrence was first retained, in Unix milliseconds.
 	TimestampMs int64 `json:"timestamp_ms"`
 	// State is the durable request state recorded by a state record and is unset on an event record.
 	State RequestState `json:"state"`

From 72bd8aa56d78b8b8e9ca58312aac3b71c3052a20 Mon Sep 17 00:00:00 2001
From: mnoah1 <mnoah1@uber.com>
Date: Tue, 1 Sep 2026 16:16:47 +0000
Subject: [PATCH 2/5] fix(stovepipe): preserve request log compatibility

---
 doc/rfc/stovepipe/request-log.md           |  2 +-
 stovepipe/core/requestlog/BUILD.bazel      |  1 +
 stovepipe/core/requestlog/recorder.go      | 48 ++++++++++++++--------
 stovepipe/core/requestlog/recorder_test.go | 32 ++++++++++++++-
 4 files changed, 64 insertions(+), 19 deletions(-)

diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md
index 0b41975f..11153291 100644
--- a/doc/rfc/stovepipe/request-log.md
+++ b/doc/rfc/stovepipe/request-log.md
@@ -154,7 +154,7 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
 | Build finished | Request ID, event kind, and build ID |
 | Validation fact recorded | Request ID, event kind, and whole-repository fact identity |
 
-The controller passes the recorder the `RequestLogStore` from the same queue-scoped storage aggregate used for the source write. The recorder assigns the current time immediately before the first insertion attempt and calls `Create`. If the ID already exists, it loads the stored record and compares every domain field. The first successfully retained timestamp is authoritative and is not compared with a later retry's newly sampled time. Identical domain content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten.
+The controller passes the recorder the `RequestLogStore` from the same queue-scoped storage aggregate used for the source write. The recorder assigns the current time immediately before the first insertion attempt and calls `Create`. If the ID already exists, it loads the stored record and compares the explicitly designated stable semantic fields. The first successfully retained timestamp is authoritative and is not compared with a later retry's newly sampled time. Metadata keys emitted by both records must agree, while a key present on only one record remains compatible so an additive metadata rollout does not turn retries of older occurrences into conflicts. Compatible content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten or enriched.
 
 ## Storage Contract
 
diff --git a/stovepipe/core/requestlog/BUILD.bazel b/stovepipe/core/requestlog/BUILD.bazel
index c4798183..a70b6847 100644
--- a/stovepipe/core/requestlog/BUILD.bazel
+++ b/stovepipe/core/requestlog/BUILD.bazel
@@ -18,6 +18,7 @@ go_test(
     srcs = ["recorder_test.go"],
     embed = [":go_default_library"],
     deps = [
+        "//platform/metrics:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "//stovepipe/extension/storage/mock:go_default_library",
diff --git a/stovepipe/core/requestlog/recorder.go b/stovepipe/core/requestlog/recorder.go
index 212b635f..14355dd4 100644
--- a/stovepipe/core/requestlog/recorder.go
+++ b/stovepipe/core/requestlog/recorder.go
@@ -24,8 +24,6 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
-	"maps"
-	"reflect"
 	"strconv"
 	"time"
 
@@ -73,35 +71,38 @@ func (r *recorder) RecordRequestState(ctx context.Context, store storage.Request
 
 func (r *recorder) record(ctx context.Context, store storage.RequestLogStore, log entity.RequestLog) error {
 	log.TimestampMs = r.now().UnixMilli()
-	tag := occurrenceTag(log)
 
 	if err := log.Validate(); err != nil {
-		metrics.NamedCounter(r.scope, "record", "validation_failure", 1, tag)
+		r.count(ctx, "validation_failure", log)
 		return fmt.Errorf("invalid request log occurrence: %w", err)
 	}
 
 	if err := store.Create(ctx, log); err == nil {
-		metrics.NamedCounter(r.scope, "record", "created", 1, tag)
+		r.count(ctx, "created", log)
 		return nil
 	} else if !errors.Is(err, storage.ErrAlreadyExists) {
-		metrics.NamedCounter(r.scope, "record", "storage_failure", 1, tag)
+		r.count(ctx, "storage_failure", log)
 		return fmt.Errorf("failed to create request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
 	}
 
 	stored, err := store.Get(ctx, log.RequestID, log.ID)
 	if err != nil {
-		metrics.NamedCounter(r.scope, "record", "storage_failure", 1, tag)
+		r.count(ctx, "storage_failure", log)
 		return fmt.Errorf("failed to reconcile request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
 	}
 	if !sameOccurrence(stored, log) {
-		metrics.NamedCounter(r.scope, "record", "conflict", 1, tag)
+		r.count(ctx, "conflict", log)
 		return fmt.Errorf("request log conflicts with retained occurrence request_id=%q log_id=%q", log.RequestID, log.ID)
 	}
 
-	metrics.NamedCounter(r.scope, "record", "identical_existing", 1, tag)
+	r.count(ctx, "identical_existing", log)
 	return nil
 }
 
+func (r *recorder) count(ctx context.Context, counter string, log entity.RequestLog) {
+	metrics.NamedCounter(r.scope, "record", counter, 1, metrics.TagsFromContext(ctx, occurrenceTag(log))...)
+}
+
 func occurrenceID(queue, requestID string, identity ...string) string {
 	hash := sha256.New()
 	parts := append([]string{queue, requestID}, identity...)
@@ -115,14 +116,27 @@ func occurrenceID(queue, requestID string, identity ...string) string {
 }
 
 func sameOccurrence(stored, candidate entity.RequestLog) bool {
-	// The first successful insert owns display time; retries compare only the occurrence's domain content.
-	storedMetadata := stored.Metadata
-	candidateMetadata := candidate.Metadata
-	stored.Metadata = nil
-	candidate.Metadata = nil
-	stored.TimestampMs = 0
-	candidate.TimestampMs = 0
-	return reflect.DeepEqual(stored, candidate) && maps.Equal(storedMetadata, candidateMetadata)
+	// This list is the compatibility boundary for duplicate reconciliation. New entity fields do not
+	// become conflict-sensitive until they are deliberately added here.
+	return stored.ID == candidate.ID &&
+		stored.Queue == candidate.Queue &&
+		stored.RequestID == candidate.RequestID &&
+		stored.State == candidate.State &&
+		stored.Event == candidate.Event &&
+		stored.RequestVersion == candidate.RequestVersion &&
+		stored.OutcomeReason == candidate.OutcomeReason &&
+		metadataCompatible(stored.Metadata, candidate.Metadata)
+}
+
+func metadataCompatible(stored, candidate map[string]string) bool {
+	// One-sided keys permit additive metadata rollout without making a retry conflict with an older
+	// immutable row. Values emitted by both versions must still agree.
+	for key, storedValue := range stored {
+		if candidateValue, ok := candidate[key]; ok && candidateValue != storedValue {
+			return false
+		}
+	}
+	return true
 }
 
 func occurrenceTag(log entity.RequestLog) metrics.Tag {
diff --git a/stovepipe/core/requestlog/recorder_test.go b/stovepipe/core/requestlog/recorder_test.go
index a9ebe8cb..6fe516df 100644
--- a/stovepipe/core/requestlog/recorder_test.go
+++ b/stovepipe/core/requestlog/recorder_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/metrics"
 	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 	storagemock "github.com/uber/submitqueue/stovepipe/extension/storage/mock"
@@ -186,5 +187,34 @@ func TestSameOccurrenceMetadata(t *testing.T) {
 	assert.True(t, sameOccurrence(stored, base))
 
 	stored.Metadata["source"] = "hook"
-	assert.False(t, sameOccurrence(stored, base))
+	assert.True(t, sameOccurrence(stored, base))
+
+	candidate := base
+	candidate.Metadata = map[string]string{"source": "ingest"}
+	assert.False(t, sameOccurrence(stored, candidate))
+
+	candidate.Metadata["source"] = "hook"
+	candidate.Metadata["new_key"] = "new_value"
+	stored.Metadata["new_key"] = "new_value"
+	assert.True(t, sameOccurrence(stored, candidate))
+}
+
+func TestRecorderMetricsIncludeContextTags(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	scope := tally.NewTestScope("test", nil)
+	recorder := &recorder{
+		scope: scope,
+		now:   func() time.Time { return time.UnixMilli(testNowMs) },
+	}
+	store := storagemock.NewMockRequestLogStore(ctrl)
+	store.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
+	ctx := metrics.WithContextTags(context.Background(), metrics.NewTag("queue", testQueue))
+
+	require.NoError(t, recorder.RecordRequestState(ctx, store, entity.Request{
+		ID: testRequestID, Queue: testQueue, State: entity.RequestStateAccepted, Version: 1,
+	}, entity.RequestOutcomeReasonUnknown))
+
+	counter, ok := scope.Snapshot().Counters()["test.record.created+occurrence=accepted,queue=monorepo/main"]
+	require.True(t, ok)
+	assert.EqualValues(t, 1, counter.Value())
 }

From e41cc1d7cf377997dee71fc88935a2884a92199b Mon Sep 17 00:00:00 2001
From: mnoah1 <mnoah1@uber.com>
Date: Tue, 1 Sep 2026 16:30:34 +0000
Subject: [PATCH 3/5] refactor(stovepipe): materialize request logs directly

---
 doc/rfc/stovepipe/request-history-api.md      |  4 +-
 doc/rfc/stovepipe/request-log.md              | 12 ++--
 stovepipe/core/requestlog/BUILD.bazel         |  4 +-
 .../{recorder.go => materializer.go}          | 51 +++++++++-------
 ...{recorder_test.go => materializer_test.go} | 60 ++++++++++++-------
 stovepipe/core/requestlog/mock/BUILD.bazel    |  2 +-
 .../core/requestlog/mock/materializer_mock.go | 57 ++++++++++++++++++
 .../core/requestlog/mock/recorder_mock.go     | 57 ------------------
 8 files changed, 137 insertions(+), 110 deletions(-)
 rename stovepipe/core/requestlog/{recorder.go => materializer.go} (72%)
 rename stovepipe/core/requestlog/{recorder_test.go => materializer_test.go} (74%)
 create mode 100644 stovepipe/core/requestlog/mock/materializer_mock.go
 delete mode 100644 stovepipe/core/requestlog/mock/recorder_mock.go

diff --git a/doc/rfc/stovepipe/request-history-api.md b/doc/rfc/stovepipe/request-history-api.md
index b297eea1..79b5b33b 100644
--- a/doc/rfc/stovepipe/request-history-api.md
+++ b/doc/rfc/stovepipe/request-history-api.md
@@ -107,7 +107,9 @@ Event rows remain in SubmitQueue history but never participate in current-status
 
 Stovepipe has no equivalent ownership gap. The same service owns the queue-scoped `Request`, `ValidationFact`, request-URI mapping, and request-log store. Operational reads use their owning entities, while request history reads retained log records directly.
 
-Stovepipe therefore does not add `RequestSummary`, replay history to determine current state, or materialize another history table. The controller performs only an in-memory wire projection from stored log records to protobuf messages. This avoids a second winner-selection algorithm competing with Request CAS state.
+Stovepipe calls `requestlog.Materializer.PersistLog` directly, without an intermediate topic. Its initial materializer appends only the request log: it does not add `RequestSummary`, replay history to determine current state, or materialize another history table. The controller performs only an in-memory wire projection from stored log records to protobuf messages. This avoids a second winner-selection algorithm competing with Request CAS state.
+
+`PersistLog` receives the whole queue-scoped storage aggregate so a future current-status or queue-list API can add SubmitQueue-style summary and index projections behind the same write boundary without changing request-log producers. Such projections remain deferred until a concrete read path needs them; `Request` remains authoritative in the initial implementation.
 
 ## Ordering and Consistency
 
diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md
index 11153291..3b553168 100644
--- a/doc/rfc/stovepipe/request-log.md
+++ b/doc/rfc/stovepipe/request-log.md
@@ -2,7 +2,7 @@
 
 ## Summary
 
-Stovepipe retains an append-only request log for each validation request. Its internal `RequestLog` is the counterpart of SubmitQueue's `RequestLog`: both retain request status changes and explanatory lifecycle events, while Stovepipe persists records directly instead of sending them through a cross-service log topic and materializer. The public API presents these records as request history. The log records every durable `Request.State` transition plus three asynchronous milestones needed to explain those transitions and the public verdict:
+Stovepipe retains an append-only request log for each validation request. Its internal `RequestLog` is the counterpart of SubmitQueue's `RequestLog`: both retain request status changes and explanatory lifecycle events, while Stovepipe calls its materializer directly instead of sending records through a cross-service log topic. The public API presents these records as request history. The log records every durable `Request.State` transition plus three asynchronous milestones needed to explain those transitions and the public verdict:
 
 - `build_triggered`;
 - `build_finished`;
@@ -145,7 +145,7 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
 
 ## Stable IDs and Idempotency
 
-`stovepipe/core/requestlog.Recorder` constructs opaque IDs from durable identities:
+`stovepipe/core/requestlog` constructs opaque IDs from durable identities before passing each record to `Materializer.PersistLog`:
 
 | Entry | Stable identity inputs |
 |---|---|
@@ -154,7 +154,7 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
 | Build finished | Request ID, event kind, and build ID |
 | Validation fact recorded | Request ID, event kind, and whole-repository fact identity |
 
-The controller passes the recorder the `RequestLogStore` from the same queue-scoped storage aggregate used for the source write. The recorder assigns the current time immediately before the first insertion attempt and calls `Create`. If the ID already exists, it loads the stored record and compares the explicitly designated stable semantic fields. The first successfully retained timestamp is authoritative and is not compared with a later retry's newly sampled time. Metadata keys emitted by both records must agree, while a key present on only one record remains compatible so an additive metadata rollout does not turn retries of older occurrences into conflicts. Compatible content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten or enriched.
+The controller passes the materializer the same queue-scoped storage aggregate used for the source write. The materializer preserves a supplied occurrence time or assigns the current time immediately before the first insertion attempt, then calls `RequestLogStore.Create`. If the ID already exists, it loads the stored record and compares the explicitly designated stable semantic fields. The first successfully retained timestamp is authoritative and is not compared with a later retry's newly sampled time. Metadata keys emitted by both records must agree, while a key present on only one record remains compatible so an additive metadata rollout does not turn retries of older occurrences into conflicts. Compatible content is idempotent success; conflicting content is an internal consistency error, and the stored record is never overwritten or enriched.
 
 ## Storage Contract
 
@@ -192,7 +192,7 @@ For a Request transition, the controller:
 2. computes `newVersion = oldVersion + 1`;
 3. calls `RequestStore.Update(updated, oldVersion, newVersion)`;
 4. assigns the in-memory version only after the store succeeds;
-5. asks the recorder to create the log record from the durable Request and the bounded context still owned by that stage;
+5. constructs the log record from the durable Request and bounded context still owned by that stage, then calls `Materializer.PersistLog`;
 6. publishes the downstream handoff.
 
 Request creation, Build changes, and fact creation use the same source-write, log-write, dependent-publish ordering. A request-log outage can leave a source update visible, but it cannot allow dependent processing to move past an unrecorded transition.
@@ -224,7 +224,7 @@ This is rollout work, not deferred cleanup: a mandatory request log without a du
 
 Rollout therefore:
 
-1. deploys request-log storage, the recorder, and readers;
+1. deploys request-log storage, the materializer, and readers;
 2. enables writers and verifies every repair path stage by stage;
 3. enables the public API after every writer and repair path is active.
 
@@ -250,7 +250,7 @@ Writer tests cover source success followed by log failure, redelivery with the l
 
 Tests reconstruct the latest Request state from state entries by request version and compare it with `RequestStore.Get`. They separately verify that only a durable fact produces green or broken.
 
-The recorder reports create, identical-existing, conflict, validation failure, and storage failure counters tagged only by bounded state or event. IDs, URIs, and errors remain structured log fields rather than metric tags. Alerts cover sustained repair gaps and content conflicts.
+The materializer reports create, identical-existing, conflict, validation failure, and storage failure counters tagged only by bounded state or event plus tags carried by the context. IDs, URIs, and errors remain structured log fields rather than metric tags. Alerts cover sustained repair gaps and content conflicts.
 
 ## Alternatives Considered
 
diff --git a/stovepipe/core/requestlog/BUILD.bazel b/stovepipe/core/requestlog/BUILD.bazel
index a70b6847..da18171c 100644
--- a/stovepipe/core/requestlog/BUILD.bazel
+++ b/stovepipe/core/requestlog/BUILD.bazel
@@ -2,7 +2,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["recorder.go"],
+    srcs = ["materializer.go"],
     importpath = "github.com/uber/submitqueue/stovepipe/core/requestlog",
     visibility = ["//visibility:public"],
     deps = [
@@ -15,7 +15,7 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["recorder_test.go"],
+    srcs = ["materializer_test.go"],
     embed = [":go_default_library"],
     deps = [
         "//platform/metrics:go_default_library",
diff --git a/stovepipe/core/requestlog/recorder.go b/stovepipe/core/requestlog/materializer.go
similarity index 72%
rename from stovepipe/core/requestlog/recorder.go
rename to stovepipe/core/requestlog/materializer.go
index 14355dd4..0fa1e587 100644
--- a/stovepipe/core/requestlog/recorder.go
+++ b/stovepipe/core/requestlog/materializer.go
@@ -15,7 +15,7 @@
 // Package requestlog retains the request occurrences exposed by Stovepipe's history API.
 package requestlog
 
-//go:generate mockgen -source=recorder.go -destination=mock/recorder_mock.go -package=mock
+//go:generate mockgen -source=materializer.go -destination=mock/materializer_mock.go -package=mock
 
 import (
 	"context"
@@ -38,27 +38,30 @@ const (
 	_occurrenceKindState = "state"
 )
 
-// Recorder retains idempotent request-state occurrences.
-type Recorder interface {
-	// RecordRequestState retains the request's current durable state and version.
-	RecordRequestState(context.Context, storage.RequestLogStore, entity.Request, entity.RequestOutcomeReason) error
+// Materializer persists request-log occurrences into their queue-scoped read model.
+type Materializer interface {
+	// PersistLog retains one request-log occurrence idempotently.
+	PersistLog(context.Context, storage.Storage, entity.RequestLog) error
 }
 
-type recorder struct {
+type materializer struct {
 	scope tally.Scope
 	now   func() time.Time
 }
 
-// NewRecorder creates a request-log recorder.
-func NewRecorder(scope tally.Scope) Recorder {
-	return &recorder{
-		scope: scope.SubScope("request_log_recorder"),
+var _ Materializer = (*materializer)(nil)
+
+// NewMaterializer creates a request-log materializer.
+func NewMaterializer(scope tally.Scope) Materializer {
+	return &materializer{
+		scope: scope.SubScope("request_log_materializer"),
 		now:   time.Now,
 	}
 }
 
-func (r *recorder) RecordRequestState(ctx context.Context, store storage.RequestLogStore, request entity.Request, outcomeReason entity.RequestOutcomeReason) error {
-	log := entity.RequestLog{
+// NewRequestStateLog constructs the occurrence representing the request's current durable state.
+func NewRequestStateLog(request entity.Request, outcomeReason entity.RequestOutcomeReason) entity.RequestLog {
+	return entity.RequestLog{
 		ID:             occurrenceID(request.Queue, request.ID, _occurrenceKindState, strconv.FormatInt(int64(request.Version), 10)),
 		Queue:          request.Queue,
 		RequestID:      request.ID,
@@ -66,41 +69,43 @@ func (r *recorder) RecordRequestState(ctx context.Context, store storage.Request
 		RequestVersion: request.Version,
 		OutcomeReason:  outcomeReason,
 	}
-	return r.record(ctx, store, log)
 }
 
-func (r *recorder) record(ctx context.Context, store storage.RequestLogStore, log entity.RequestLog) error {
-	log.TimestampMs = r.now().UnixMilli()
+func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, log entity.RequestLog) error {
+	if log.TimestampMs == 0 {
+		log.TimestampMs = m.now().UnixMilli()
+	}
 
 	if err := log.Validate(); err != nil {
-		r.count(ctx, "validation_failure", log)
+		m.count(ctx, "validation_failure", log)
 		return fmt.Errorf("invalid request log occurrence: %w", err)
 	}
 
+	store := stores.GetRequestLogStore()
 	if err := store.Create(ctx, log); err == nil {
-		r.count(ctx, "created", log)
+		m.count(ctx, "created", log)
 		return nil
 	} else if !errors.Is(err, storage.ErrAlreadyExists) {
-		r.count(ctx, "storage_failure", log)
+		m.count(ctx, "storage_failure", log)
 		return fmt.Errorf("failed to create request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
 	}
 
 	stored, err := store.Get(ctx, log.RequestID, log.ID)
 	if err != nil {
-		r.count(ctx, "storage_failure", log)
+		m.count(ctx, "storage_failure", log)
 		return fmt.Errorf("failed to reconcile request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
 	}
 	if !sameOccurrence(stored, log) {
-		r.count(ctx, "conflict", log)
+		m.count(ctx, "conflict", log)
 		return fmt.Errorf("request log conflicts with retained occurrence request_id=%q log_id=%q", log.RequestID, log.ID)
 	}
 
-	r.count(ctx, "identical_existing", log)
+	m.count(ctx, "identical_existing", log)
 	return nil
 }
 
-func (r *recorder) count(ctx context.Context, counter string, log entity.RequestLog) {
-	metrics.NamedCounter(r.scope, "record", counter, 1, metrics.TagsFromContext(ctx, occurrenceTag(log))...)
+func (m *materializer) count(ctx context.Context, counter string, log entity.RequestLog) {
+	metrics.NamedCounter(m.scope, "persist", counter, 1, metrics.TagsFromContext(ctx, occurrenceTag(log))...)
 }
 
 func occurrenceID(queue, requestID string, identity ...string) string {
diff --git a/stovepipe/core/requestlog/recorder_test.go b/stovepipe/core/requestlog/materializer_test.go
similarity index 74%
rename from stovepipe/core/requestlog/recorder_test.go
rename to stovepipe/core/requestlog/materializer_test.go
index 6fe516df..4fc320df 100644
--- a/stovepipe/core/requestlog/recorder_test.go
+++ b/stovepipe/core/requestlog/materializer_test.go
@@ -36,16 +36,19 @@ const (
 	testNowMs     = int64(1735689600000)
 )
 
-func newTestRecorder(t *testing.T) (*recorder, *storagemock.MockRequestLogStore) {
+func newTestMaterializer(t *testing.T) (*materializer, *storagemock.MockStorage, *storagemock.MockRequestLogStore) {
 	t.Helper()
 	ctrl := gomock.NewController(t)
-	return &recorder{
+	stores := storagemock.NewMockStorage(ctrl)
+	store := storagemock.NewMockRequestLogStore(ctrl)
+	stores.EXPECT().GetRequestLogStore().Return(store).AnyTimes()
+	return &materializer{
 		scope: tally.NoopScope,
 		now:   func() time.Time { return time.UnixMilli(testNowMs) },
-	}, storagemock.NewMockRequestLogStore(ctrl)
+	}, stores, store
 }
 
-func TestRecorderRecordRequestState(t *testing.T) {
+func TestMaterializerPersistRequestStateLog(t *testing.T) {
 	tests := []struct {
 		name          string
 		state         entity.RequestState
@@ -69,7 +72,7 @@ func TestRecorderRecordRequestState(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			recorder, store := newTestRecorder(t)
+			materializer, stores, store := newTestMaterializer(t)
 			request := entity.Request{
 				ID:      testRequestID,
 				Queue:   testQueue,
@@ -90,7 +93,8 @@ func TestRecorderRecordRequestState(t *testing.T) {
 				})
 			}
 
-			err := recorder.RecordRequestState(context.Background(), store, request, tt.outcomeReason)
+			log := NewRequestStateLog(request, tt.outcomeReason)
+			err := materializer.PersistLog(context.Background(), stores, log)
 			if tt.wantErr {
 				require.Error(t, err)
 			} else {
@@ -100,8 +104,8 @@ func TestRecorderRecordRequestState(t *testing.T) {
 	}
 }
 
-func TestRecorderExistingIdenticalOccurrenceIsSuccess(t *testing.T) {
-	recorder, store := newTestRecorder(t)
+func TestMaterializerExistingIdenticalOccurrenceIsSuccess(t *testing.T) {
+	materializer, stores, store := newTestMaterializer(t)
 	request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateAccepted, Version: 1}
 
 	var candidate entity.RequestLog
@@ -119,11 +123,23 @@ func TestRecorderExistingIdenticalOccurrenceIsSuccess(t *testing.T) {
 		},
 	)
 
-	require.NoError(t, recorder.RecordRequestState(context.Background(), store, request, entity.RequestOutcomeReasonUnknown))
+	require.NoError(t, materializer.PersistLog(context.Background(), stores, NewRequestStateLog(request, entity.RequestOutcomeReasonUnknown)))
+}
+
+func TestMaterializerPreservesSuppliedTimestamp(t *testing.T) {
+	materializer, stores, store := newTestMaterializer(t)
+	log := NewRequestStateLog(
+		entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateAccepted, Version: 1},
+		entity.RequestOutcomeReasonUnknown,
+	)
+	log.TimestampMs = testNowMs - 1000
+	store.EXPECT().Create(gomock.Any(), log).Return(nil)
+
+	require.NoError(t, materializer.PersistLog(context.Background(), stores, log))
 }
 
-func TestRecorderExistingConflictingOccurrenceFails(t *testing.T) {
-	recorder, store := newTestRecorder(t)
+func TestMaterializerExistingConflictingOccurrenceFails(t *testing.T) {
+	materializer, stores, store := newTestMaterializer(t)
 	request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateSucceeded, Version: 3}
 
 	var candidate entity.RequestLog
@@ -139,10 +155,10 @@ func TestRecorderExistingConflictingOccurrenceFails(t *testing.T) {
 		},
 	)
 
-	require.Error(t, recorder.RecordRequestState(context.Background(), store, request, entity.RequestOutcomeReasonBuildSucceeded))
+	require.Error(t, materializer.PersistLog(context.Background(), stores, NewRequestStateLog(request, entity.RequestOutcomeReasonBuildSucceeded)))
 }
 
-func TestRecorderStorageFailures(t *testing.T) {
+func TestMaterializerStorageFailures(t *testing.T) {
 	tests := []struct {
 		name  string
 		setup func(*storagemock.MockRequestLogStore)
@@ -164,11 +180,12 @@ func TestRecorderStorageFailures(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			recorder, store := newTestRecorder(t)
+			materializer, stores, store := newTestMaterializer(t)
 			tt.setup(store)
-			err := recorder.RecordRequestState(context.Background(), store, entity.Request{
+			log := NewRequestStateLog(entity.Request{
 				ID: testRequestID, Queue: testQueue, State: entity.RequestStateAccepted, Version: 1,
 			}, entity.RequestOutcomeReasonUnknown)
+			err := materializer.PersistLog(context.Background(), stores, log)
 			require.Error(t, err)
 		})
 	}
@@ -199,22 +216,25 @@ func TestSameOccurrenceMetadata(t *testing.T) {
 	assert.True(t, sameOccurrence(stored, candidate))
 }
 
-func TestRecorderMetricsIncludeContextTags(t *testing.T) {
+func TestMaterializerMetricsIncludeContextTags(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	scope := tally.NewTestScope("test", nil)
-	recorder := &recorder{
+	materializer := &materializer{
 		scope: scope,
 		now:   func() time.Time { return time.UnixMilli(testNowMs) },
 	}
+	stores := storagemock.NewMockStorage(ctrl)
 	store := storagemock.NewMockRequestLogStore(ctrl)
+	stores.EXPECT().GetRequestLogStore().Return(store)
 	store.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
 	ctx := metrics.WithContextTags(context.Background(), metrics.NewTag("queue", testQueue))
 
-	require.NoError(t, recorder.RecordRequestState(ctx, store, entity.Request{
+	log := NewRequestStateLog(entity.Request{
 		ID: testRequestID, Queue: testQueue, State: entity.RequestStateAccepted, Version: 1,
-	}, entity.RequestOutcomeReasonUnknown))
+	}, entity.RequestOutcomeReasonUnknown)
+	require.NoError(t, materializer.PersistLog(ctx, stores, log))
 
-	counter, ok := scope.Snapshot().Counters()["test.record.created+occurrence=accepted,queue=monorepo/main"]
+	counter, ok := scope.Snapshot().Counters()["test.persist.created+occurrence=accepted,queue=monorepo/main"]
 	require.True(t, ok)
 	assert.EqualValues(t, 1, counter.Value())
 }
diff --git a/stovepipe/core/requestlog/mock/BUILD.bazel b/stovepipe/core/requestlog/mock/BUILD.bazel
index 7a161d71..be4d1714 100644
--- a/stovepipe/core/requestlog/mock/BUILD.bazel
+++ b/stovepipe/core/requestlog/mock/BUILD.bazel
@@ -2,7 +2,7 @@ load("@rules_go//go:def.bzl", "go_library")
 
 go_library(
     name = "go_default_library",
-    srcs = ["recorder_mock.go"],
+    srcs = ["materializer_mock.go"],
     importpath = "github.com/uber/submitqueue/stovepipe/core/requestlog/mock",
     visibility = ["//visibility:public"],
     deps = [
diff --git a/stovepipe/core/requestlog/mock/materializer_mock.go b/stovepipe/core/requestlog/mock/materializer_mock.go
new file mode 100644
index 00000000..4ac18268
--- /dev/null
+++ b/stovepipe/core/requestlog/mock/materializer_mock.go
@@ -0,0 +1,57 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: materializer.go
+//
+// Generated by this command:
+//
+//	mockgen -source=materializer.go -destination=mock/materializer_mock.go -package=mock
+//
+
+// Package mock is a generated GoMock package.
+package mock
+
+import (
+	context "context"
+	reflect "reflect"
+
+	entity "github.com/uber/submitqueue/stovepipe/entity"
+	storage "github.com/uber/submitqueue/stovepipe/extension/storage"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockMaterializer is a mock of Materializer interface.
+type MockMaterializer struct {
+	ctrl     *gomock.Controller
+	recorder *MockMaterializerMockRecorder
+	isgomock struct{}
+}
+
+// MockMaterializerMockRecorder is the mock recorder for MockMaterializer.
+type MockMaterializerMockRecorder struct {
+	mock *MockMaterializer
+}
+
+// NewMockMaterializer creates a new mock instance.
+func NewMockMaterializer(ctrl *gomock.Controller) *MockMaterializer {
+	mock := &MockMaterializer{ctrl: ctrl}
+	mock.recorder = &MockMaterializerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockMaterializer) EXPECT() *MockMaterializerMockRecorder {
+	return m.recorder
+}
+
+// PersistLog mocks base method.
+func (m *MockMaterializer) PersistLog(arg0 context.Context, arg1 storage.Storage, arg2 entity.RequestLog) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "PersistLog", arg0, arg1, arg2)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// PersistLog indicates an expected call of PersistLog.
+func (mr *MockMaterializerMockRecorder) PersistLog(arg0, arg1, arg2 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PersistLog", reflect.TypeOf((*MockMaterializer)(nil).PersistLog), arg0, arg1, arg2)
+}
diff --git a/stovepipe/core/requestlog/mock/recorder_mock.go b/stovepipe/core/requestlog/mock/recorder_mock.go
deleted file mode 100644
index aeab397a..00000000
--- a/stovepipe/core/requestlog/mock/recorder_mock.go
+++ /dev/null
@@ -1,57 +0,0 @@
-// Code generated by MockGen. DO NOT EDIT.
-// Source: recorder.go
-//
-// Generated by this command:
-//
-//	mockgen -source=recorder.go -destination=mock/recorder_mock.go -package=mock
-//
-
-// Package mock is a generated GoMock package.
-package mock
-
-import (
-	context "context"
-	reflect "reflect"
-
-	entity "github.com/uber/submitqueue/stovepipe/entity"
-	storage "github.com/uber/submitqueue/stovepipe/extension/storage"
-	gomock "go.uber.org/mock/gomock"
-)
-
-// MockRecorder is a mock of Recorder interface.
-type MockRecorder struct {
-	ctrl     *gomock.Controller
-	recorder *MockRecorderMockRecorder
-	isgomock struct{}
-}
-
-// MockRecorderMockRecorder is the mock recorder for MockRecorder.
-type MockRecorderMockRecorder struct {
-	mock *MockRecorder
-}
-
-// NewMockRecorder creates a new mock instance.
-func NewMockRecorder(ctrl *gomock.Controller) *MockRecorder {
-	mock := &MockRecorder{ctrl: ctrl}
-	mock.recorder = &MockRecorderMockRecorder{mock}
-	return mock
-}
-
-// EXPECT returns an object that allows the caller to indicate expected use.
-func (m *MockRecorder) EXPECT() *MockRecorderMockRecorder {
-	return m.recorder
-}
-
-// RecordRequestState mocks base method.
-func (m *MockRecorder) RecordRequestState(arg0 context.Context, arg1 storage.RequestLogStore, arg2 entity.Request, arg3 entity.RequestOutcomeReason) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "RecordRequestState", arg0, arg1, arg2, arg3)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// RecordRequestState indicates an expected call of RecordRequestState.
-func (mr *MockRecorderMockRecorder) RecordRequestState(arg0, arg1, arg2, arg3 any) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordRequestState", reflect.TypeOf((*MockRecorder)(nil).RecordRequestState), arg0, arg1, arg2, arg3)
-}

From 3e2af244452977185e417820c6860ccf592effe3 Mon Sep 17 00:00:00 2001
From: mnoah1 <mnoah1@uber.com>
Date: Tue, 1 Sep 2026 17:22:23 +0000
Subject: [PATCH 4/5] refactor(stovepipe): simplify request log identity

---
 doc/rfc/stovepipe/request-log.md              |  4 +-
 stovepipe/core/requestlog/BUILD.bazel         |  1 +
 stovepipe/core/requestlog/materializer.go     | 52 +++++--------------
 .../core/requestlog/materializer_test.go      | 26 ++++++----
 stovepipe/entity/request_log.go               |  2 +-
 5 files changed, 32 insertions(+), 53 deletions(-)

diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md
index 3b553168..0422c78b 100644
--- a/doc/rfc/stovepipe/request-log.md
+++ b/doc/rfc/stovepipe/request-log.md
@@ -73,7 +73,7 @@ The retained unit is `entity.RequestLog`:
 
 ```go
 type RequestLog struct {
-    // ID is the stable identity of one logical occurrence within the request. It is opaque, stable across redelivery, and never derived from time or randomness.
+    // ID is the stable identity of one logical occurrence within the request. It is stable across redelivery and never derived from time or randomness.
     ID string
     // Queue is the logical queue containing the request and scopes RequestID.
     Queue string
@@ -145,7 +145,7 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
 
 ## Stable IDs and Idempotency
 
-`stovepipe/core/requestlog` constructs opaque IDs from durable identities before passing each record to `Materializer.PersistLog`:
+`stovepipe/core/requestlog` composes readable IDs from durable identities with the same `publish.IntentID` convention used by SubmitQueue before passing each record to `Materializer.PersistLog`. The surrounding `(queue, request_id)` storage key scopes the ID to one request, so a request state entry uses `state/<request-version>` without repeating the request identity:
 
 | Entry | Stable identity inputs |
 |---|---|
diff --git a/stovepipe/core/requestlog/BUILD.bazel b/stovepipe/core/requestlog/BUILD.bazel
index da18171c..c6a7e17f 100644
--- a/stovepipe/core/requestlog/BUILD.bazel
+++ b/stovepipe/core/requestlog/BUILD.bazel
@@ -7,6 +7,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//platform/metrics:go_default_library",
+        "//platform/publish:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go
index 0fa1e587..8741c756 100644
--- a/stovepipe/core/requestlog/materializer.go
+++ b/stovepipe/core/requestlog/materializer.go
@@ -19,9 +19,6 @@ package requestlog
 
 import (
 	"context"
-	"crypto/sha256"
-	"encoding/binary"
-	"encoding/hex"
 	"errors"
 	"fmt"
 	"strconv"
@@ -30,6 +27,7 @@ import (
 	"github.com/uber-go/tally"
 
 	"github.com/uber/submitqueue/platform/metrics"
+	"github.com/uber/submitqueue/platform/publish"
 	"github.com/uber/submitqueue/stovepipe/entity"
 	"github.com/uber/submitqueue/stovepipe/extension/storage"
 )
@@ -62,7 +60,7 @@ func NewMaterializer(scope tally.Scope) Materializer {
 // NewRequestStateLog constructs the occurrence representing the request's current durable state.
 func NewRequestStateLog(request entity.Request, outcomeReason entity.RequestOutcomeReason) entity.RequestLog {
 	return entity.RequestLog{
-		ID:             occurrenceID(request.Queue, request.ID, _occurrenceKindState, strconv.FormatInt(int64(request.Version), 10)),
+		ID:             publish.IntentID(_occurrenceKindState, strconv.FormatInt(int64(request.Version), 10)),
 		Queue:          request.Queue,
 		RequestID:      request.ID,
 		State:          request.State,
@@ -77,50 +75,38 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l
 	}
 
 	if err := log.Validate(); err != nil {
-		m.count(ctx, "validation_failure", log)
+		m.count(ctx, "validation_failure")
 		return fmt.Errorf("invalid request log occurrence: %w", err)
 	}
 
 	store := stores.GetRequestLogStore()
 	if err := store.Create(ctx, log); err == nil {
-		m.count(ctx, "created", log)
+		m.count(ctx, "created")
 		return nil
 	} else if !errors.Is(err, storage.ErrAlreadyExists) {
-		m.count(ctx, "storage_failure", log)
+		m.count(ctx, "storage_failure")
 		return fmt.Errorf("failed to create request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
 	}
 
 	stored, err := store.Get(ctx, log.RequestID, log.ID)
 	if err != nil {
-		m.count(ctx, "storage_failure", log)
+		m.count(ctx, "storage_failure")
 		return fmt.Errorf("failed to reconcile request log request_id=%q log_id=%q: %w", log.RequestID, log.ID, err)
 	}
-	if !sameOccurrence(stored, log) {
-		m.count(ctx, "conflict", log)
+	if !sameSemanticOccurrence(stored, log) {
+		m.count(ctx, "conflict")
 		return fmt.Errorf("request log conflicts with retained occurrence request_id=%q log_id=%q", log.RequestID, log.ID)
 	}
 
-	m.count(ctx, "identical_existing", log)
+	m.count(ctx, "identical_existing")
 	return nil
 }
 
-func (m *materializer) count(ctx context.Context, counter string, log entity.RequestLog) {
-	metrics.NamedCounter(m.scope, "persist", counter, 1, metrics.TagsFromContext(ctx, occurrenceTag(log))...)
+func (m *materializer) count(ctx context.Context, counter string) {
+	metrics.NamedCounter(m.scope, "persist", counter, 1, metrics.TagsFromContext(ctx)...)
 }
 
-func occurrenceID(queue, requestID string, identity ...string) string {
-	hash := sha256.New()
-	parts := append([]string{queue, requestID}, identity...)
-	var size [8]byte
-	for _, part := range parts {
-		binary.BigEndian.PutUint64(size[:], uint64(len(part)))
-		_, _ = hash.Write(size[:])
-		_, _ = hash.Write([]byte(part))
-	}
-	return "log/" + hex.EncodeToString(hash.Sum(nil))
-}
-
-func sameOccurrence(stored, candidate entity.RequestLog) bool {
+func sameSemanticOccurrence(stored, candidate entity.RequestLog) bool {
 	// This list is the compatibility boundary for duplicate reconciliation. New entity fields do not
 	// become conflict-sensitive until they are deliberately added here.
 	return stored.ID == candidate.ID &&
@@ -143,17 +129,3 @@ func metadataCompatible(stored, candidate map[string]string) bool {
 	}
 	return true
 }
-
-func occurrenceTag(log entity.RequestLog) metrics.Tag {
-	value := "invalid"
-	switch log.State {
-	case entity.RequestStateAccepted,
-		entity.RequestStateProcessing,
-		entity.RequestStateSuperseded,
-		entity.RequestStateSucceeded,
-		entity.RequestStateFailed,
-		entity.RequestStateCancelled:
-		value = string(log.State)
-	}
-	return metrics.NewTag("occurrence", value)
-}
diff --git a/stovepipe/core/requestlog/materializer_test.go b/stovepipe/core/requestlog/materializer_test.go
index 4fc320df..f0a191da 100644
--- a/stovepipe/core/requestlog/materializer_test.go
+++ b/stovepipe/core/requestlog/materializer_test.go
@@ -191,29 +191,35 @@ func TestMaterializerStorageFailures(t *testing.T) {
 	}
 }
 
-func TestOccurrenceID(t *testing.T) {
-	assert.Equal(t, occurrenceID("queue", "request", "state", "1"), occurrenceID("queue", "request", "state", "1"))
-	assert.NotEqual(t, occurrenceID("queue", "request", "state", "1"), occurrenceID("queue", "request", "state", "2"))
-	assert.NotEqual(t, occurrenceID("queue", "request", "ab", "c"), occurrenceID("queue", "request", "a", "bc"))
+func TestNewRequestStateLogStableID(t *testing.T) {
+	request := entity.Request{ID: testRequestID, Queue: testQueue, State: entity.RequestStateAccepted, Version: 1}
+	first := NewRequestStateLog(request, entity.RequestOutcomeReasonUnknown)
+	retry := NewRequestStateLog(request, entity.RequestOutcomeReasonUnknown)
+	request.Version++
+	next := NewRequestStateLog(request, entity.RequestOutcomeReasonUnknown)
+
+	assert.Equal(t, "state/1", first.ID)
+	assert.Equal(t, first.ID, retry.ID)
+	assert.NotEqual(t, first.ID, next.ID)
 }
 
-func TestSameOccurrenceMetadata(t *testing.T) {
+func TestSameSemanticOccurrenceMetadata(t *testing.T) {
 	base := entity.RequestLog{ID: "log/1", Queue: testQueue, RequestID: testRequestID, State: entity.RequestStateAccepted, RequestVersion: 1}
 	stored := base
 	stored.Metadata = map[string]string{}
-	assert.True(t, sameOccurrence(stored, base))
+	assert.True(t, sameSemanticOccurrence(stored, base))
 
 	stored.Metadata["source"] = "hook"
-	assert.True(t, sameOccurrence(stored, base))
+	assert.True(t, sameSemanticOccurrence(stored, base))
 
 	candidate := base
 	candidate.Metadata = map[string]string{"source": "ingest"}
-	assert.False(t, sameOccurrence(stored, candidate))
+	assert.False(t, sameSemanticOccurrence(stored, candidate))
 
 	candidate.Metadata["source"] = "hook"
 	candidate.Metadata["new_key"] = "new_value"
 	stored.Metadata["new_key"] = "new_value"
-	assert.True(t, sameOccurrence(stored, candidate))
+	assert.True(t, sameSemanticOccurrence(stored, candidate))
 }
 
 func TestMaterializerMetricsIncludeContextTags(t *testing.T) {
@@ -234,7 +240,7 @@ func TestMaterializerMetricsIncludeContextTags(t *testing.T) {
 	}, entity.RequestOutcomeReasonUnknown)
 	require.NoError(t, materializer.PersistLog(ctx, stores, log))
 
-	counter, ok := scope.Snapshot().Counters()["test.persist.created+occurrence=accepted,queue=monorepo/main"]
+	counter, ok := scope.Snapshot().Counters()["test.persist.created+queue=monorepo/main"]
 	require.True(t, ok)
 	assert.EqualValues(t, 1, counter.Value())
 }
diff --git a/stovepipe/entity/request_log.go b/stovepipe/entity/request_log.go
index 2bb06671..fe47a9aa 100644
--- a/stovepipe/entity/request_log.go
+++ b/stovepipe/entity/request_log.go
@@ -54,7 +54,7 @@ const (
 
 // RequestLog is one immutable request state or explanatory lifecycle occurrence.
 type RequestLog struct {
-	// ID is the stable opaque identity of the occurrence within the request.
+	// ID is the stable identity of the occurrence within the request.
 	ID string `json:"id"`
 	// Queue is the logical queue containing the request and scopes RequestID.
 	Queue string `json:"queue"`

From 2f7ff911f7e69b736ee9e09f8b5f8cbd02a90bdf Mon Sep 17 00:00:00 2001
From: mnoah1 <mnoah1@uber.com>
Date: Tue, 1 Sep 2026 17:31:44 +0000
Subject: [PATCH 5/5] docs(stovepipe): explain retained log deduplication

---
 doc/rfc/stovepipe/request-log.md          | 2 ++
 stovepipe/core/requestlog/materializer.go | 2 ++
 2 files changed, 4 insertions(+)

diff --git a/doc/rfc/stovepipe/request-log.md b/doc/rfc/stovepipe/request-log.md
index 0422c78b..d50eee30 100644
--- a/doc/rfc/stovepipe/request-log.md
+++ b/doc/rfc/stovepipe/request-log.md
@@ -147,6 +147,8 @@ Terminal entries retain domain reasons rather than transport mechanisms. Initial
 
 `stovepipe/core/requestlog` composes readable IDs from durable identities with the same `publish.IntentID` convention used by SubmitQueue before passing each record to `Materializer.PersistLog`. The surrounding `(queue, request_id)` storage key scopes the ID to one request, so a request state entry uses `state/<request-version>` without repeating the request identity:
 
+SubmitQueue applies that identity to the message carrying a log to its materializer, while its log store remains append-only and may retain a duplicate after a later materialization step fails. Stovepipe has no intermediate log topic, so it applies the identity to the retained row itself: retrying the direct call reloads the existing occurrence and succeeds only when its semantic content is compatible.
+
 | Entry | Stable identity inputs |
 |---|---|
 | Request state transition | Request ID and request version |
diff --git a/stovepipe/core/requestlog/materializer.go b/stovepipe/core/requestlog/materializer.go
index 8741c756..187526bc 100644
--- a/stovepipe/core/requestlog/materializer.go
+++ b/stovepipe/core/requestlog/materializer.go
@@ -80,6 +80,8 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l
 	}
 
 	store := stores.GetRequestLogStore()
+	// SubmitQueue deduplicates the message that hands a log to its materializer. Stovepipe has no
+	// log topic, so the retained occurrence ID is the retry boundary instead.
 	if err := store.Create(ctx, log); err == nil {
 		m.count(ctx, "created")
 		return nil