feat: wire managers and add cascade delete skeleton

Summary:
Intent:
- Wire TriggerRunManager and PipelineRunManager into the pipeline controller
- Add handleDeletion method with no-children path (removes finalizer immediately)

Changes:
- Add triggerRunManager and pipelineRunManager fields to Reconciler struct
- Construct managers in Register after handler creation
- Replace deletion stub with handleDeletion that lists children and removes finalizer if none
- Register field indexes on fake client for tests
- Add TestCascadeDelete_NoChildren test

Test Plan:
- go test ./components/pipeline/... -v -count=1 (all tests pass)

Revert Plan:
Revert this PR via git revert.

Jira Issues:

Author: ishtmeet.singh

go/components/pipeline/BUILD.bazel

@@ -13,6 +13,8 @@ go_library(
         "//go/api:go_default_library",
         "//go/api/handler:go_default_library",
         "//go/base/env:go_default_library",
+        "//go/components/pipelinerun:go_default_library",
+        "//go/components/triggerrun:go_default_library",
         "//proto/api:go_default_library",
         "//proto/api/v2:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
@@ -35,6 +37,8 @@ go_test(
         "//go/api:go_default_library",
         "//go/api/handler:go_default_library",
         "//go/base/env:go_default_library",
+        "//go/components/pipelinerun:go_default_library",
+        "//go/components/triggerrun:go_default_library",
         "//proto/api:go_default_library",
         "//proto/api/v2:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",

go/components/pipeline/controller.go

@@ -21,6 +21,8 @@ import (
 	"github.com/michelangelo-ai/michelangelo/go/api"
 	apiHandler "github.com/michelangelo-ai/michelangelo/go/api/handler"
 	"github.com/michelangelo-ai/michelangelo/go/base/env"
+	"github.com/michelangelo-ai/michelangelo/go/components/pipelinerun"
+	"github.com/michelangelo-ai/michelangelo/go/components/triggerrun"
 	apipb "github.com/michelangelo-ai/michelangelo/proto-go/api"
 	v2pb "github.com/michelangelo-ai/michelangelo/proto-go/api/v2"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -41,9 +43,11 @@ const (
 // operations and maintains environment context and logging capabilities.
 type Reconciler struct {
 	api.Handler
-	env               env.Context
-	logger            *zap.Logger
-	apiHandlerFactory apiHandler.Factory
+	env                env.Context
+	logger             *zap.Logger
+	apiHandlerFactory  apiHandler.Factory
+	triggerRunManager  triggerrun.Manager
+	pipelineRunManager pipelinerun.Manager
 }
 
 // Reconcile is the main reconciliation loop entry point for Pipeline resources.
@@ -75,14 +79,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 			}
 		}
 	} else {
-		// Pipeline is being deleted — remove finalizer to allow deletion
-		// Cascade delete logic will be added in a subsequent PR
-		logger.Info("Pipeline is being deleted, removing finalizer")
-		controllerutil.RemoveFinalizer(pipeline, api.PipelineFinalizer)
-		if err := r.Update(ctx, pipeline, &metav1.UpdateOptions{}); err != nil {
-			return ctrl.Result{}, fmt.Errorf("remove pipeline finalizer: %w", err)
-		}
-		return ctrl.Result{}, nil
+		return r.handleDeletion(ctx, pipeline, logger)
 	}
 
 	originalPipeline := pipeline.DeepCopy()
@@ -139,6 +136,55 @@ func (r *Reconciler) updatePipelineStatus(ctx context.Context, pipeline *v2pb.Pi
 	return result, nil
 }
 
+func (r *Reconciler) handleDeletion(ctx context.Context, pipeline *v2pb.Pipeline, logger *zap.Logger) (ctrl.Result, error) {
+	// If the finalizer is not present we don't own this deletion; nothing to cascade.
+	// Avoids wasted list/kill/delete work on pipelines that pre-date the finalizer rollout.
+	if !controllerutil.ContainsFinalizer(pipeline, api.PipelineFinalizer) {
+		return ctrl.Result{}, nil
+	}
+	logger.Info("Pipeline is being deleted, starting cascade delete")
+
+	triggerRuns, err := r.triggerRunManager.ListTriggerRunsForPipeline(ctx, pipeline.Namespace, pipeline.Name)
+	if err != nil {
+		logger.Error("Failed to list trigger runs for cascade delete",
+			zap.Error(err),
+			zap.String("operation", "list_trigger_runs"),
+			zap.String("namespace", pipeline.Namespace),
+			zap.String("name", pipeline.Name))
+		return ctrl.Result{}, fmt.Errorf("list trigger runs for pipeline %s/%s: %w", pipeline.Namespace, pipeline.Name, err)
+	}
+
+	pipelineRuns, err := r.pipelineRunManager.ListPipelineRunsForPipeline(ctx, pipeline.Namespace, pipeline.Name)
+	if err != nil {
+		logger.Error("Failed to list pipeline runs for cascade delete",
+			zap.Error(err),
+			zap.String("operation", "list_pipeline_runs"),
+			zap.String("namespace", pipeline.Namespace),
+			zap.String("name", pipeline.Name))
+		return ctrl.Result{}, fmt.Errorf("list pipeline runs for pipeline %s/%s: %w", pipeline.Namespace, pipeline.Name, err)
+	}
+
+	if len(triggerRuns) == 0 && len(pipelineRuns) == 0 {
+		logger.Info("No children found, removing finalizer")
+		controllerutil.RemoveFinalizer(pipeline, api.PipelineFinalizer)
+		if updateErr := r.Update(ctx, pipeline, &metav1.UpdateOptions{}); updateErr != nil {
+			logger.Error("Failed to remove finalizer after cascade delete",
+				zap.Error(updateErr),
+				zap.String("operation", "remove_finalizer"),
+				zap.String("namespace", pipeline.Namespace),
+				zap.String("name", pipeline.Name))
+			return ctrl.Result{}, fmt.Errorf("remove finalizer on pipeline %s/%s: %w", pipeline.Namespace, pipeline.Name, updateErr)
+		}
+		return ctrl.Result{}, nil
+	}
+
+	// Kill and delete steps will be added in subsequent PRs
+	logger.Info("Children found, requeueing for cascade delete",
+		zap.Int("triggerRuns", len(triggerRuns)),
+		zap.Int("pipelineRuns", len(pipelineRuns)))
+	return ctrl.Result{RequeueAfter: reconcileInterval}, nil
+}
+
 // formatRevisionName generates a standardized revision name for a pipeline.
 //
 // The name format is: "pipeline-{lowercase-pipeline-name}-{git-ref-prefix}"
@@ -176,6 +222,8 @@ func (r *Reconciler) Register(mgr ctrl.Manager) error {
 		return err
 	}
 	r.Handler = handler
+	r.triggerRunManager = triggerrun.NewManager(handler, r.logger)
+	r.pipelineRunManager = pipelinerun.NewManager(handler, r.logger)
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v2pb.Pipeline{}).
 		Complete(r)

go/components/pipeline/controller_test.go

@@ -3,6 +3,7 @@ package pipeline
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -16,6 +17,8 @@ import (
 	"github.com/michelangelo-ai/michelangelo/go/api"
 	apiHandler "github.com/michelangelo-ai/michelangelo/go/api/handler"
 	"github.com/michelangelo-ai/michelangelo/go/base/env"
+	"github.com/michelangelo-ai/michelangelo/go/components/pipelinerun"
+	"github.com/michelangelo-ai/michelangelo/go/components/triggerrun"
 	apipb "github.com/michelangelo-ai/michelangelo/proto-go/api"
 	v2pb "github.com/michelangelo-ai/michelangelo/proto-go/api/v2"
 	"go.uber.org/zap/zaptest"
@@ -268,11 +271,16 @@ func TestReconcile_RemovesFinalizerOnDelete(t *testing.T) {
 	}
 }
 
-// updateErroringHandler wraps an api.Handler and returns a configured error
-// from Update. Used to exercise finalizer Update error branches.
+// updateErroringHandler wraps an api.Handler and returns configured errors
+// from List/Update. Used to exercise finalizer and cascade-delete error branches.
 type updateErroringHandler struct {
 	api.Handler
-	updateErr error
+	updateErr     error
+	listErr       error
+	// listErrForType, when non-nil, only fails List when the list object is of
+	// the given type (e.g. "*v2pb.TriggerRunList"). This lets us assert the
+	// controller surfaces the exact failure path we expect.
+	listErrForType string
 }
 
 func (u *updateErroringHandler) Update(ctx context.Context, obj client.Object, opts *metav1.UpdateOptions) error {
@@ -282,13 +290,33 @@ func (u *updateErroringHandler) Update(ctx context.Context, obj client.Object, o
 	return u.Handler.Update(ctx, obj, opts)
 }
 
+func (u *updateErroringHandler) List(ctx context.Context, namespace string, opts *metav1.ListOptions, listOptionsExt *apipb.ListOptionsExt, list client.ObjectList) error {
+	if u.listErr != nil && (u.listErrForType == "" || u.listErrForType == fmt.Sprintf("%T", list)) {
+		return u.listErr
+	}
+	return u.Handler.List(ctx, namespace, opts, listOptionsExt, list)
+}
+
 func setUpReconcilerWithUpdateErr(t *testing.T, initialObjects []client.Object, updateErr error) *Reconciler {
+	return setUpReconcilerWithErrors(t, initialObjects, updateErr, nil, "")
+}
+
+func setUpReconcilerWithErrors(t *testing.T, initialObjects []client.Object, updateErr, listErr error, listErrForType string) *Reconciler {
 	scheme := runtime.NewScheme()
 	require.NoError(t, v2pb.AddToScheme(scheme))
 	k8sClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(initialObjects...).WithStatusSubresource(initialObjects...).Build()
+	logger := zaptest.NewLogger(t)
+	handler := &updateErroringHandler{
+		Handler:        apiHandler.NewFakeAPIHandler(k8sClient),
+		updateErr:      updateErr,
+		listErr:        listErr,
+		listErrForType: listErrForType,
+	}
 	return &Reconciler{
-		Handler: &updateErroringHandler{Handler: apiHandler.NewFakeAPIHandler(k8sClient), updateErr: updateErr},
-		logger:  zaptest.NewLogger(t),
+		Handler:            handler,
+		logger:             logger,
+		triggerRunManager:  triggerrun.NewManager(handler, logger),
+		pipelineRunManager: pipelinerun.NewManager(handler, logger),
 	}
 }
 
@@ -313,7 +341,120 @@ func TestReconcile_AddFinalizer_UpdateError(t *testing.T) {
 	require.Contains(t, err.Error(), "add pipeline finalizer")
 }
 
-func TestReconcile_RemoveFinalizer_UpdateError(t *testing.T) {
+// NOTE: Error coverage for the delete path is provided by
+// TestCascadeDelete_RemoveFinalizer_UpdateError below, which exercises the
+// handleDeletion error wrapping introduced in this PR.
+
+func TestCascadeDelete_NoChildren(t *testing.T) {
+	now := metav1.Now()
+	pipeline := &v2pb.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              "test-pipeline",
+			Namespace:         "test-namespace",
+			Finalizers:        []string{api.PipelineFinalizer},
+			DeletionTimestamp: &now,
+		},
+		Spec: v2pb.PipelineSpec{
+			Commit: &v2pb.CommitInfo{GitRef: "abc123", Branch: "main"},
+		},
+	}
+	reconciler := setUpReconciler(t, []client.Object{pipeline}, env.Context{})
+
+	result, err := reconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Name: "test-pipeline", Namespace: "test-namespace"},
+	})
+	require.NoError(t, err)
+	require.Equal(t, ctrl.Result{}, result)
+}
+
+func TestCascadeDelete_FinalizerAbsent(t *testing.T) {
+	// Pipeline with a DeletionTimestamp but not our finalizer must not be
+	// cascaded. handleDeletion returns immediately without listing children.
+	// The fake client requires at least one finalizer when DeletionTimestamp
+	// is set, so we attach an unrelated finalizer.
+	now := metav1.Now()
+	pipeline := &v2pb.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              "test-pipeline",
+			Namespace:         "test-namespace",
+			Finalizers:        []string{"unrelated/finalizer"},
+			DeletionTimestamp: &now,
+		},
+		Spec: v2pb.PipelineSpec{
+			Commit: &v2pb.CommitInfo{GitRef: "abc123", Branch: "main"},
+		},
+	}
+	// Seed a TR that would normally be killed. handleDeletion must not touch it.
+	tr := &v2pb.TriggerRun{
+		ObjectMeta: metav1.ObjectMeta{Name: "tr-running", Namespace: "test-namespace"},
+		Spec: v2pb.TriggerRunSpec{
+			Pipeline: &apipb.ResourceIdentifier{Name: "test-pipeline", Namespace: "test-namespace"},
+		},
+		Status: v2pb.TriggerRunStatus{State: v2pb.TRIGGER_RUN_STATE_RUNNING},
+	}
+	reconciler := setUpReconciler(t, []client.Object{pipeline, tr}, env.Context{})
+
+	result, err := reconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Name: "test-pipeline", Namespace: "test-namespace"},
+	})
+	require.NoError(t, err)
+	require.Equal(t, ctrl.Result{}, result)
+
+	untouched := &v2pb.TriggerRun{}
+	require.NoError(t, reconciler.Get(context.Background(), "test-namespace", "tr-running", &metav1.GetOptions{}, untouched))
+	require.NotEqual(t, v2pb.TRIGGER_RUN_ACTION_KILL, untouched.Spec.Action)
+	require.False(t, untouched.Spec.Kill)
+}
+
+func TestCascadeDelete_ListTriggerRunsError(t *testing.T) {
+	now := metav1.Now()
+	pipeline := &v2pb.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              "test-pipeline",
+			Namespace:         "test-namespace",
+			Finalizers:        []string{api.PipelineFinalizer},
+			DeletionTimestamp: &now,
+		},
+		Spec: v2pb.PipelineSpec{
+			Commit: &v2pb.CommitInfo{GitRef: "abc123", Branch: "main"},
+		},
+	}
+	listErr := errors.New("list tr boom")
+	reconciler := setUpReconcilerWithErrors(t, []client.Object{pipeline}, nil, listErr, "*v2.TriggerRunList")
+
+	_, err := reconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Name: "test-pipeline", Namespace: "test-namespace"},
+	})
+	require.Error(t, err)
+	require.ErrorIs(t, err, listErr)
+	require.Contains(t, err.Error(), "list trigger runs for pipeline test-namespace/test-pipeline")
+}
+
+func TestCascadeDelete_ListPipelineRunsError(t *testing.T) {
+	now := metav1.Now()
+	pipeline := &v2pb.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              "test-pipeline",
+			Namespace:         "test-namespace",
+			Finalizers:        []string{api.PipelineFinalizer},
+			DeletionTimestamp: &now,
+		},
+		Spec: v2pb.PipelineSpec{
+			Commit: &v2pb.CommitInfo{GitRef: "abc123", Branch: "main"},
+		},
+	}
+	listErr := errors.New("list pr boom")
+	reconciler := setUpReconcilerWithErrors(t, []client.Object{pipeline}, nil, listErr, "*v2.PipelineRunList")
+
+	_, err := reconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Name: "test-pipeline", Namespace: "test-namespace"},
+	})
+	require.Error(t, err)
+	require.ErrorIs(t, err, listErr)
+	require.Contains(t, err.Error(), "list pipeline runs for pipeline test-namespace/test-pipeline")
+}
+
+func TestCascadeDelete_RemoveFinalizer_UpdateError(t *testing.T) {
 	now := metav1.Now()
 	pipeline := &v2pb.Pipeline{
 		ObjectMeta: metav1.ObjectMeta{
@@ -327,24 +468,68 @@ func TestReconcile_RemoveFinalizer_UpdateError(t *testing.T) {
 		},
 	}
 	updateErr := errors.New("update boom")
-	reconciler := setUpReconcilerWithUpdateErr(t, []client.Object{pipeline}, updateErr)
+	reconciler := setUpReconcilerWithErrors(t, []client.Object{pipeline}, updateErr, nil, "")
 
 	_, err := reconciler.Reconcile(context.Background(), ctrl.Request{
 		NamespacedName: types.NamespacedName{Name: "test-pipeline", Namespace: "test-namespace"},
 	})
 	require.Error(t, err)
 	require.ErrorIs(t, err, updateErr)
-	require.Contains(t, err.Error(), "remove pipeline finalizer")
+	require.Contains(t, err.Error(), "remove finalizer on pipeline test-namespace/test-pipeline")
+}
+
+func TestCascadeDelete_WithChildrenRequeues(t *testing.T) {
+	// When children exist, handleDeletion does not remove the finalizer; it
+	// requeues after reconcileInterval so a subsequent PR can perform kill/delete.
+	now := metav1.Now()
+	pipeline := &v2pb.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              "test-pipeline",
+			Namespace:         "test-namespace",
+			Finalizers:        []string{api.PipelineFinalizer},
+			DeletionTimestamp: &now,
+		},
+		Spec: v2pb.PipelineSpec{
+			Commit: &v2pb.CommitInfo{GitRef: "abc123", Branch: "main"},
+		},
+	}
+	tr := &v2pb.TriggerRun{
+		ObjectMeta: metav1.ObjectMeta{Name: "tr-running", Namespace: "test-namespace"},
+		Spec: v2pb.TriggerRunSpec{
+			Pipeline: &apipb.ResourceIdentifier{Name: "test-pipeline", Namespace: "test-namespace"},
+		},
+		Status: v2pb.TriggerRunStatus{State: v2pb.TRIGGER_RUN_STATE_RUNNING},
+	}
+	reconciler := setUpReconciler(t, []client.Object{pipeline, tr}, env.Context{})
+
+	result, err := reconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Name: "test-pipeline", Namespace: "test-namespace"},
+	})
+	require.NoError(t, err)
+	require.Equal(t, ctrl.Result{RequeueAfter: reconcileInterval}, result)
+
+	// Finalizer should NOT have been removed yet.
+	updated := &v2pb.Pipeline{}
+	require.NoError(t, reconciler.Get(context.Background(), "test-namespace", "test-pipeline", &metav1.GetOptions{}, updated))
+	require.True(t, controllerutil.ContainsFinalizer(updated, api.PipelineFinalizer))
 }
 
 func setUpReconciler(t *testing.T, initialObjects []client.Object, env env.Context) *Reconciler {
 	scheme := runtime.NewScheme()
 	err := v2pb.AddToScheme(scheme)
 	require.NoError(t, err)
-	k8sClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(initialObjects...).WithStatusSubresource(initialObjects...).Build()
+	k8sClient := fake.NewClientBuilder().
+		WithScheme(scheme).
+		WithObjects(initialObjects...).
+		WithStatusSubresource(initialObjects...).
+		Build()
+	logger := zaptest.NewLogger(t)
+	handler := apiHandler.NewFakeAPIHandler(k8sClient)
 	reconciler := &Reconciler{
-		Handler: apiHandler.NewFakeAPIHandler(k8sClient),
-		logger:  zaptest.NewLogger(t),
+		Handler:            handler,
+		logger:             logger,
+		triggerRunManager:  triggerrun.NewManager(handler, logger),
+		pipelineRunManager: pipelinerun.NewManager(handler, logger),
 	}
 	return reconciler
 }