Fix integration test build and replay lint errors

Co-authored-by: gh-aw-bot <259018956+gh-aw-bot@users.noreply.github.com>
pelikhan · Oct 9, 2026 · Oct 9, 2026 · Oct 9, 2026 · Oct 9, 2026 · Oct 9, 2026
commit f553cea1f4f81d8dd0e939d7d0ace0db0b790142
diff --git a/pkg/workflow/maintenance_workflow_ledger_compaction_test.go b/pkg/workflow/maintenance_workflow_ledger_compaction_test.go
@@ -319,22 +319,3 @@ func TestComputeEnabledToolNames_LedgerNilSafety(t *testing.T) {
 		assert.Contains(t, names, "ledger_request_compaction")
 	})
 }
-
-// parseMaintenanceJobs splits the generated maintenance workflow into top-level job bodies.
-func parseMaintenanceJobs(t *testing.T, content string) map[string]string {
-	t.Helper()
-	_, jobsSection, found := strings.Cut(content, "\njobs:\n")
-	require.True(t, found)
-	jobs := make(map[string]string)
-	current := ""
-	for line := range strings.SplitSeq(jobsSection, "\n") {
-		if strings.HasPrefix(line, "  ") && !strings.HasPrefix(line, "   ") && strings.HasSuffix(line, ":") {
-			current = strings.TrimSuffix(strings.TrimSpace(line), ":")
-			continue
-		}
-		if current != "" {
-			jobs[current] += line + "\n"
-		}
-	}
-	return jobs
-}
diff --git a/pkg/workflow/maintenance_workflow_work_queue_compaction_test.go b/pkg/workflow/maintenance_workflow_work_queue_compaction_test.go
@@ -43,3 +43,22 @@ func TestMaintenanceGeneratedForWorkQueueOnly(t *testing.T) {
 	assert.Empty(t, jobs["work_queue_compaction_apply"])
 	assert.Empty(t, jobs["ledger_compaction_plan"])
 }
+
+// parseMaintenanceJobs splits the generated maintenance workflow into top-level job bodies.
+func parseMaintenanceJobs(t *testing.T, content string) map[string]string {
+	t.Helper()
+	_, jobsSection, found := strings.Cut(content, "\njobs:\n")
+	require.True(t, found)
+	jobs := make(map[string]string)
+	current := ""
+	for line := range strings.SplitSeq(jobsSection, "\n") {
+		if strings.HasPrefix(line, "  ") && !strings.HasPrefix(line, "   ") && strings.HasSuffix(line, ":") {
+			current = strings.TrimSuffix(strings.TrimSpace(line), ":")
+			continue
+		}
+		if current != "" {
+			jobs[current] += line + "\n"
+		}
+	}
+	return jobs
+}
diff --git a/pkg/workqueue/replay.go b/pkg/workqueue/replay.go
@@ -44,12 +44,16 @@ func Replay(commits []QueueCommit) (Projection, error) {
 	if err != nil {
 		return state, err
 	}
+	firstCommit, firstOperation, foundFirstOperation := firstReplayOperation(ordered)
+	if !foundFirstOperation {
+		return state, queueError("policy_missing", "existing queue has no policy genesis")
+	}
 	epochs := identitySet{}
 	generations := identitySet{state.CredentialGeneration: {}}
 	offset := 0
 	checkpointCount := 0
-	if kind, kindErr := operationKind(ordered[0].Operations[0]); kindErr == nil && kind == "Checkpoint" {
-		state, err = restoreCheckpoint(ordered[0])
+	if kind, kindErr := operationKind(firstOperation); kindErr == nil && kind == "Checkpoint" {
+		state, err = restoreCheckpoint(firstCommit)
 		if err != nil {
 			return state, err
 		}
@@ -196,21 +200,9 @@ func (state *Projection) replayOperation(operation Operation, kind string, commi
 	case "Completion":
 		return state.applyCompletion(operation, commit)
 	case "ClaimCancellation":
-		if err := state.cancelClaim(operation, commit); err != nil {
-			return err
-		}
-		var value ClaimOperation
-		_ = json.Unmarshal(operation, &value)
-		state.Cancellations[value.ClaimID] = operation
-		return nil
+		return state.replayClaimCancellation(operation, commit)
 	case "WorkCancellation":
-		if err := state.cancelWork(operation, commit); err != nil {
-			return err
-		}
-		var value WorkDefinition
-		_ = json.Unmarshal(operation, &value)
-		state.Cancellations[value.WorkID] = operation
-		return nil
+		return state.replayWorkCancellation(operation, commit)
 	case "WorkPriority":
 		return state.reprioritizeWork(operation)
 	case "IssueLink", "IssueComment":
@@ -220,18 +212,58 @@ func (state *Projection) replayOperation(operation Operation, kind string, commi
 	case "Release":
 		return state.release(operation, commit)
 	case "Result", "DeliveryFailure":
-		if err := state.settleResult(operation, commit, kind == "DeliveryFailure"); err != nil {
-			return err
-		}
-		var value WorkDefinition
-		_ = json.Unmarshal(operation, &value)
-		state.TerminalBarriers[value.WorkID] = operation
-		return nil
+		return state.replayResult(operation, kind, commit)
 	default:
 		return queueError("unsupported_protocol", "unknown operation")
 	}
 }
 
+func firstReplayOperation(commits []QueueCommit) (QueueCommit, Operation, bool) {
+	if len(commits) > 0 {
+		first := commits[0]
+		if len(first.Operations) > 0 {
+			return first, first.Operations[0], true
+		}
+	}
+	return QueueCommit{}, nil, false
+}
+
+func (state *Projection) replayClaimCancellation(operation Operation, commit QueueCommit) error {
+	if err := state.cancelClaim(operation, commit); err != nil {
+		return err
+	}
+	var value ClaimOperation
+	if err := json.Unmarshal(operation, &value); err != nil {
+		return err
+	}
+	state.Cancellations[value.ClaimID] = operation
+	return nil
+}
+
+func (state *Projection) replayWorkCancellation(operation Operation, commit QueueCommit) error {
+	if err := state.cancelWork(operation, commit); err != nil {
+		return err
+	}
+	var value WorkDefinition
+	if err := json.Unmarshal(operation, &value); err != nil {
+		return err
+	}
+	state.Cancellations[value.WorkID] = operation
+	return nil
+}
+
+func (state *Projection) replayResult(operation Operation, kind string, commit QueueCommit) error {
+	if err := state.settleResult(operation, commit, kind == "DeliveryFailure"); err != nil {
+		return err
+	}
+	var value WorkDefinition
+	if err := json.Unmarshal(operation, &value); err != nil {
+		return err
+	}
+	state.TerminalBarriers[value.WorkID] = operation
+	return nil
+}
+
 func (state *Projection) installReplayPolicy(op Operation, commit QueueCommit, isPolicy bool, epochs identitySet) error {
 	if !isPolicy || !state.quiescent() {
 		return queueError("policy_not_quiescent", "policy changes require queue-wide drain")