diff --git a/cmd/ateapi/internal/controlapi/crash.go b/cmd/ateapi/internal/controlapi/crash.go
index 1c371eb69..1ffce559d 100644
--- a/cmd/ateapi/internal/controlapi/crash.go
+++ b/cmd/ateapi/internal/controlapi/crash.go
@@ -21,6 +21,7 @@ import (
 	"log/slog"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/ateerrors"
 	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
@@ -28,16 +29,32 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-// maybeCrashActor inspects err returned by an atelet RPC, it crashes
-// the actor if the err carries the actorCrashed=true metadata directive.
-func maybeCrashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef, err error, wrapMsg string) error {
+// maybeCrashActor inspects err returned by an atelet RPC and crashes the actor
+// if err carries the actorCrashed=true metadata directive.
+//
+// opNames is variadic (...string) to provide backward compatibility for generic callers
+// where the operation context is not explicitly supplied. If provided, opNames[0] is
+// normalized against the bounded operations {resume, suspend, pause, unknown}.
+func maybeCrashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef, err error, wrapMsg string, opNames ...string) error {
 	if err == nil {
 		return nil
 	}
 
 	if ateerrors.ActorCrashRequested(err) {
 		slog.ErrorContext(ctx, "Setting Actor to crashed due to error", slog.Any("error", err))
-		if cerr := crashActor(ctx, st, actorRef); cerr != nil {
+		// Extract AIP-193 ErrorInfo reason enum from the RPC error detail. If unclassified
+		// or unlisted, default to ateattr.ReasonUnknown to protect metric low-cardinality.
+		reason := ateerrors.ExtractReason(err)
+		if reason == "" {
+			reason = ateattr.ReasonUnknown
+		}
+
+		opName := ateattr.OperationNameUnknown
+		if len(opNames) > 0 {
+			opName = ateattr.NormalizeOperationName(opNames[0])
+		}
+
+		if cerr := crashActor(ctx, st, actorRef, opName, reason); cerr != nil {
 			slog.ErrorContext(ctx, "Failed to crash actor", slog.Any("cerr", cerr))
 			return cerr
 		}
@@ -48,17 +65,28 @@ func maybeCrashActor(ctx context.Context, st store.Interface, actorRef resources
 
 // crashActor moves the actor to CRASHED state and frees the worker it was
 // assigned to, if any, so the worker can host other actors.
-func crashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef) error {
+func crashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef, opName, reason string) error {
 	actor, err := st.GetActor(ctx, actorRef)
 	if err != nil {
 		return fmt.Errorf("while loading actor to crash: %w", err)
 	}
 
+	wasAlreadyCrashed := actor.GetStatus() == ateapipb.Actor_STATUS_CRASHED
+	opName = ateattr.NormalizeOperationName(opName)
+	if reason == "" {
+		reason = ateattr.ReasonUnknown
+	}
+
 	var errCollected []error
-	if err := releaseWorker(ctx, st, actor); err != nil {
+	sandboxClass, err := releaseWorker(ctx, st, actor)
+	if err != nil {
 		errCollected = append(errCollected, err)
 	}
 
+	// Snapshot crash attributes before pod and pool pointers are cleared below;
+	// the counter itself is emitted only after the transition commits.
+	crashAttrs := ateattr.ActorMetricAttributes(actor, sandboxClass, opName, reason)
+
 	actor.Status = ateapipb.Actor_STATUS_CRASHED
 
 	// InProgressSnapshot is kept for debugging; failed workflow
@@ -69,15 +97,24 @@ func crashActor(ctx context.Context, st store.Interface, actorRef resources.Acto
 	actor.AteomPodUid = ""
 	actor.WorkerPoolName = ""
 
-	if _, err := st.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()); err != nil {
+	_, err = st.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion())
+	if err != nil {
 		errCollected = append(errCollected, fmt.Errorf("while marking actor crashed: %w", err))
+		return errors.Join(errCollected...)
+	}
+
+	// Increment metric only after a successful UpdateActor, and only if the actor was not already crashed.
+	if !wasAlreadyCrashed {
+		recordActorCrash(ctx, crashAttrs)
 	}
+
 	return errors.Join(errCollected...)
 }
 
 // releaseWorker clears the worker's assignment if it still points at the given
 // actor. A missing worker or an already-cleared assignment is not an error.
-func releaseWorker(ctx context.Context, st store.Interface, actor *ateapipb.Actor) error {
+// It returns the worker's sandboxClass if found.
+func releaseWorker(ctx context.Context, st store.Interface, actor *ateapipb.Actor) (string, error) {
 	podNamespace := actor.GetAteomPodNamespace()
 	podName := actor.GetAteomPodName()
 	podUid := actor.GetAteomPodUid()
@@ -85,32 +122,34 @@ func releaseWorker(ctx context.Context, st store.Interface, actor *ateapipb.Acto
 
 	if podNamespace == "" || podName == "" || poolName == "" {
 		slog.WarnContext(ctx, "Actor's worker assignment is already cleared")
-		return nil
+		return "", nil
 	}
 
 	worker, err := st.GetWorker(ctx, podNamespace, poolName, podName)
 	if errors.Is(err, store.ErrNotFound) {
 		// No need to release if the worker is not found.
 		slog.WarnContext(ctx, "Worker already gone while crashing actor, skipping release", slog.String("worker", podUid))
-		return nil
+		return "", nil
 	}
 	if err != nil {
-		return fmt.Errorf("while getting worker to release: %w", err)
+		return "", fmt.Errorf("while getting worker to release: %w", err)
 	}
+
+	sandboxClass := worker.GetSandboxClass()
 	wass := worker.GetAssignment()
 	if wass == nil {
 		slog.WarnContext(ctx, "Worker's assignment is already nil, skipping release", slog.String("worker", podUid))
-		return nil
+		return sandboxClass, nil
 	}
 	// Only free it if it still belongs to us
 	if resources.ActorRefFromObjectRef(wass.GetActor()) != resources.ActorRefFromActor(actor) {
 		slog.WarnContext(ctx, "Worker already assigned to another Actor", slog.String("worker", podUid))
-		return nil
+		return sandboxClass, nil
 	}
 
 	worker.Assignment = nil
 	if err := st.UpdateWorker(ctx, worker, worker.GetVersion()); err != nil {
-		return fmt.Errorf("while releasing worker: %w", err)
+		return sandboxClass, fmt.Errorf("while releasing worker: %w", err)
 	}
-	return nil
+	return sandboxClass, nil
 }
diff --git a/cmd/ateapi/internal/controlapi/crash_test.go b/cmd/ateapi/internal/controlapi/crash_test.go
index 641e11df2..67e1b7626 100644
--- a/cmd/ateapi/internal/controlapi/crash_test.go
+++ b/cmd/ateapi/internal/controlapi/crash_test.go
@@ -22,9 +22,12 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/ateerrors"
 	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -222,7 +225,8 @@ func TestCrashActor(t *testing.T) {
 				tt.setup(t, ctx, st)
 			}
 
-			err := crashActor(ctx, st, actorRef)
+			err := crashActor(ctx, st, actorRef, ateattr.OperationNameUnknown, ateattr.ReasonUnknown)
+
 			tt.check(t, ctx, st, err)
 		})
 	}
@@ -350,3 +354,95 @@ func TestMaybeCrashActor(t *testing.T) {
 		})
 	}
 }
+
+func TestCrashActor_Metrics(t *testing.T) {
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	if err := RegisterActorCrashes(mp.Meter("ateapi")); err != nil {
+		t.Fatalf("RegisterActorCrashes: %v", err)
+	}
+
+	ctx := context.Background()
+	st, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+
+	actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-crash-test"}
+	worker := &ateapipb.Worker{
+		WorkerNamespace: "demo-ns",
+		WorkerPool:      "pool-1",
+		WorkerPod:       "pod-1",
+		SandboxClass:    "gvisor",
+		Assignment: &ateapipb.Assignment{
+			Actor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name},
+		},
+	}
+	if err := st.CreateWorker(ctx, worker); err != nil {
+		t.Fatalf("CreateWorker: %v", err)
+	}
+
+	actor := &ateapipb.Actor{
+		Metadata: &ateapipb.ResourceMetadata{
+			Atespace: actorRef.Atespace,
+			Name:     actorRef.Name,
+			Uid:      "uid-123",
+		},
+		ActorTemplateNamespace: "demo-ns",
+		ActorTemplateName:      "counter-template",
+		WorkerPoolName:         "pool-1",
+		AteomPodNamespace:      "demo-ns",
+		AteomPodName:           "pod-1",
+		AteomPodUid:            "pod-uid-1",
+		Status:                 ateapipb.Actor_STATUS_RUNNING,
+	}
+	if _, err := st.CreateActor(ctx, actor); err != nil {
+		t.Fatalf("CreateActor: %v", err)
+	}
+
+	if err := crashActor(ctx, st, actorRef, ateattr.OperationNameResume, ateattr.ReasonCorruptedAssignment); err != nil {
+		t.Fatalf("crashActor: %v", err)
+	}
+
+	assertCrashMetricDatapoint(t, reader, ateattr.OperationNameResume, ateattr.ReasonCorruptedAssignment, "demo-ns", "counter-template", "pool-1", "gvisor", 1)
+}
+
+func assertCrashMetricDatapoint(t *testing.T, reader *sdkmetric.ManualReader, wantOpName, wantReason, wantTmplNS, wantTmplName, wantWorkerPool, wantSandboxClass string, wantValue int64) {
+	t.Helper()
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(context.Background(), &rm); err != nil {
+		t.Fatalf("Collect: %v", err)
+	}
+
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name != "ate.actor.crashes" {
+				continue
+			}
+			sum, ok := m.Data.(metricdata.Sum[int64])
+			if !ok {
+				continue
+			}
+			for _, dp := range sum.DataPoints {
+				op, _ := dp.Attributes.Value(ateattr.OperationNameKey)
+				r, _ := dp.Attributes.Value(ateattr.FailureReasonKey)
+				tNS, _ := dp.Attributes.Value(ateattr.TemplateNamespaceKey)
+				tName, _ := dp.Attributes.Value(ateattr.TemplateNameKey)
+				wp, _ := dp.Attributes.Value(ateattr.WorkerPoolNameKey)
+				sc, _ := dp.Attributes.Value(ateattr.SandboxClassKey)
+
+				if op.AsString() == wantOpName &&
+					r.AsString() == wantReason &&
+					tNS.AsString() == wantTmplNS &&
+					tName.AsString() == wantTmplName &&
+					wp.AsString() == wantWorkerPool &&
+					sc.AsString() == wantSandboxClass {
+					if dp.Value != wantValue {
+						t.Errorf("metric value = %d, want %d", dp.Value, wantValue)
+					}
+					return
+				}
+			}
+		}
+	}
+	t.Errorf("did not find ate.actor.crashes metric with attrs: opName=%q, reason=%q, tmplNS=%q, tmplName=%q, workerPool=%q, sandboxClass=%q",
+		wantOpName, wantReason, wantTmplNS, wantTmplName, wantWorkerPool, wantSandboxClass)
+}
diff --git a/cmd/ateapi/internal/controlapi/metrics.go b/cmd/ateapi/internal/controlapi/metrics.go
index 3a21e8837..762572ddf 100644
--- a/cmd/ateapi/internal/controlapi/metrics.go
+++ b/cmd/ateapi/internal/controlapi/metrics.go
@@ -21,11 +21,39 @@ import (
 	"github.com/agent-substrate/substrate/internal/ateattr"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"k8s.io/apimachinery/pkg/labels"
 )
 
-const workerpoolWorkersMetric = "ate.workerpool.workers"
+const (
+	workerpoolWorkersMetric = "ate.workerpool.workers"
+	actorCrashesMetric      = "ate.actor.crashes"
+)
+
+var actorCrashesCounter metric.Int64Counter
+
+// RegisterActorCrashes initializes the ate.actor.crashes counter instrument.
+func RegisterActorCrashes(meter metric.Meter) error {
+	counter, err := meter.Int64Counter(
+		actorCrashesMetric,
+		metric.WithUnit("{actor}"),
+		metric.WithDescription("Total number of actors that transitioned to STATUS_CRASHED."),
+	)
+	if err != nil {
+		return fmt.Errorf("create %s counter: %w", actorCrashesMetric, err)
+	}
+	actorCrashesCounter = counter
+	return nil
+}
+
+// recordActorCrash records a crash event on ate.actor.crashes with low-cardinality attributes.
+func recordActorCrash(ctx context.Context, attrs []attribute.KeyValue) {
+	if actorCrashesCounter == nil || len(attrs) == 0 {
+		return
+	}
+	actorCrashesCounter.Add(ctx, 1, metric.WithAttributes(attrs...))
+}
 
 // RegisterWorkerCount wires the ate.workerpool.workers observable against workers
 // (workercache.Cache.Workers in prod) and listPools (a WorkerPool lister's List,
diff --git a/cmd/ateapi/internal/controlapi/syncer.go b/cmd/ateapi/internal/controlapi/syncer.go
index 5e1d58961..d728cab1f 100644
--- a/cmd/ateapi/internal/controlapi/syncer.go
+++ b/cmd/ateapi/internal/controlapi/syncer.go
@@ -21,6 +21,7 @@ import (
 	"maps"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/resources"
 	listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
@@ -308,6 +309,20 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
 	if actor.Status == ateapipb.Actor_STATUS_SUSPENDED {
 		return nil
 	}
+	opName := ateattr.OperationNameUnknown
+	switch actor.Status {
+	case ateapipb.Actor_STATUS_RESUMING:
+		opName = ateattr.OperationNameResume
+	case ateapipb.Actor_STATUS_SUSPENDING:
+		opName = ateattr.OperationNameSuspend
+	case ateapipb.Actor_STATUS_PAUSING:
+		opName = ateattr.OperationNamePause
+	}
+
+	wasAlreadyCrashed := actor.GetStatus() == ateapipb.Actor_STATUS_CRASHED
+
+	// Snapshot crash attributes before pod and pool pointers are cleared on actor.
+	crashAttrs := ateattr.ActorMetricAttributes(actor, worker.GetSandboxClass(), opName, ateattr.ReasonWorkerPodGone)
 
 	actor.Status = ateapipb.Actor_STATUS_CRASHED
 	actor.AteomPodNamespace = ""
@@ -318,5 +333,10 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
 	actor.WorkerPoolName = ""
 
 	_, err = s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion())
+
+	if err == nil && !wasAlreadyCrashed {
+		recordActorCrash(ctx, crashAttrs)
+	}
 	return err
+
 }
diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go
index f2050ae4a..3a431ac11 100644
--- a/cmd/ateapi/internal/controlapi/syncer_test.go
+++ b/cmd/ateapi/internal/controlapi/syncer_test.go
@@ -24,11 +24,14 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/resources"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
 	atefake "github.com/agent-substrate/substrate/pkg/client/clientset/versioned/fake"
 	"github.com/agent-substrate/substrate/pkg/client/informers/externalversions"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -635,15 +638,27 @@ func TestSyncer_ReconcileOrphanedWorkers(t *testing.T) {
 func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) {
 	ns, pool, pod, ip := "ns-status", "pool1", "worker-status", "10.0.0.3"
 	tests := []struct {
-		name  string
-		start ateapipb.Actor_Status
-		want  ateapipb.Actor_Status
+		name       string
+		start      ateapipb.Actor_Status
+		wantStatus ateapipb.Actor_Status
+		wantOp     string
+		wantMetric bool
 	}{
-		{name: "running becomes crashed", start: ateapipb.Actor_STATUS_RUNNING, want: ateapipb.Actor_STATUS_CRASHED},
-		{name: "suspended stays suspended", start: ateapipb.Actor_STATUS_SUSPENDED, want: ateapipb.Actor_STATUS_SUSPENDED},
+		{name: "running becomes crashed", start: ateapipb.Actor_STATUS_RUNNING, wantStatus: ateapipb.Actor_STATUS_CRASHED, wantOp: ateattr.OperationNameUnknown, wantMetric: true},
+		{name: "resuming becomes crashed", start: ateapipb.Actor_STATUS_RESUMING, wantStatus: ateapipb.Actor_STATUS_CRASHED, wantOp: ateattr.OperationNameResume, wantMetric: true},
+		{name: "suspending becomes crashed", start: ateapipb.Actor_STATUS_SUSPENDING, wantStatus: ateapipb.Actor_STATUS_CRASHED, wantOp: ateattr.OperationNameSuspend, wantMetric: true},
+		{name: "pausing becomes crashed", start: ateapipb.Actor_STATUS_PAUSING, wantStatus: ateapipb.Actor_STATUS_CRASHED, wantOp: ateattr.OperationNamePause, wantMetric: true},
+		{name: "suspended stays suspended", start: ateapipb.Actor_STATUS_SUSPENDED, wantStatus: ateapipb.Actor_STATUS_SUSPENDED, wantMetric: false},
 	}
+
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
+			reader := sdkmetric.NewManualReader()
+			mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+			if err := RegisterActorCrashes(mp.Meter("ateapi")); err != nil {
+				t.Fatalf("RegisterActorCrashes: %v", err)
+			}
+
 			ctx := context.Background()
 			persistence, cleanup := storetest.SetupTestStore(t)
 			defer cleanup()
@@ -652,6 +667,7 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) {
 			atespace, actorID := "team-status", "actor-status"
 			if _, err := persistence.CreateActor(ctx, &ateapipb.Actor{
 				Metadata: &ateapipb.ResourceMetadata{Name: actorID, Atespace: atespace}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
+				WorkerPoolName:    pool,
 				Status:            tc.start,
 				AteomPodNamespace: ns, AteomPodName: pod, AteomPodIp: ip, AteomPodUid: "uid",
 			}); err != nil {
@@ -660,7 +676,8 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) {
 			if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
 				WorkerNamespace: ns, WorkerPool: pool, WorkerPod: pod, Ip: ip,
 				WorkerPodUid: "08675309-4a65-6e6e-7973-6e756d626572", NodeName: "node1",
-				State: ateapipb.Worker_STATE_ACTIVE,
+				SandboxClass: "gvisor",
+				State:        ateapipb.Worker_STATE_ACTIVE,
 				Assignment: &ateapipb.Assignment{
 					ActorTemplate: &ateapipb.KubeNamespacedObjectRef{Namespace: ns, Name: "tmpl"},
 					Actor:         &ateapipb.ObjectRef{Name: actorID, Atespace: atespace},
@@ -677,12 +694,17 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) {
 			if err != nil {
 				t.Fatalf("get actor: %v", err)
 			}
-			if got.GetStatus() != tc.want {
-				t.Errorf("actor status = %v, want %v", got.GetStatus(), tc.want)
+			if got.GetStatus() != tc.wantStatus {
+				t.Errorf("actor status = %v, want %v", got.GetStatus(), tc.wantStatus)
 			}
-			if tc.want == ateapipb.Actor_STATUS_CRASHED && got.GetAteomPodUid() != "" {
+			if tc.wantStatus == ateapipb.Actor_STATUS_CRASHED && got.GetAteomPodUid() != "" {
 				t.Errorf("crashed actor AteomPodUid = %q, want cleared", got.GetAteomPodUid())
 			}
+
+			if tc.wantMetric {
+				assertCrashMetricDatapoint(t, reader, tc.wantOp, ateattr.ReasonWorkerPodGone, ns, "tmpl", pool, "gvisor", 1)
+			}
 		})
 	}
+
 }
diff --git a/cmd/ateapi/internal/controlapi/workflow_pause.go b/cmd/ateapi/internal/controlapi/workflow_pause.go
index c162117f5..283294857 100644
--- a/cmd/ateapi/internal/controlapi/workflow_pause.go
+++ b/cmd/ateapi/internal/controlapi/workflow_pause.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
 	"github.com/agent-substrate/substrate/internal/resources"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
@@ -119,7 +120,8 @@ func (s *CallAteletPauseStep) CheckPrerequisite(ctx context.Context, input *Paus
 		return status.Errorf(codes.FailedPrecondition, "CallAteletPauseStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, state.Actor.GetStatus(), ateapipb.Actor_STATUS_PAUSING)
 	}
 	if state.Actor.GetAteomPodNamespace() == "" || state.Actor.GetAteomPodName() == "" {
-		if err := crashActor(ctx, s.store, input.ActorRef); err != nil {
+		// Missing active worker pod reference in PAUSING state indicates corrupted store state.
+		if err := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNamePause, ateattr.ReasonCorruptedAssignment); err != nil {
 			slog.ErrorContext(ctx, "Failed to crash actor", slog.String("err", err.Error()))
 		}
 		return status.Errorf(codes.FailedPrecondition, "CallAteletPauseStep prerequisite not met for Actor: %s. AteomPodNamespace: %s, GetAteomPodName %s", input.ActorRef, state.Actor.GetAteomPodNamespace(), state.Actor.GetAteomPodName())
@@ -132,7 +134,7 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
 	if err != nil {
 		if errors.Is(err, ErrWorkerPodNotFound) {
 			slog.ErrorContext(ctx, "Worker pod gone before checkpoint, crashing actor", "namespace", state.Actor.GetAteomPodNamespace(), "pod", state.Actor.GetAteomPodName(), "in_progress_snapshot", state.Actor.GetInProgressSnapshot())
-			if err := crashActor(ctx, s.store, input.ActorRef); err != nil {
+			if err := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNamePause, ateattr.ReasonWorkerPodGone); err != nil {
 				slog.ErrorContext(ctx, "Failed to crash actor", slog.String("err", err.Error()))
 			}
 			return fmt.Errorf("actor is CRASHED because its worker pod is gone and no snapshot was written")
@@ -167,7 +169,7 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
 	}
 
 	_, err = client.Checkpoint(ctx, req)
-	return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while checkpointing workload")
+	return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while checkpointing workload", ateattr.OperationNamePause)
 }
 
 func (s *CallAteletPauseStep) RetryBackoff() *wait.Backoff { return nil }
@@ -190,7 +192,7 @@ func (s *DetachVolumesForPauseStep) CheckPrerequisite(ctx context.Context, input
 }
 
 func (s *DetachVolumesForPauseStep) Execute(ctx context.Context, input *PauseInput, state *PauseState) error {
-	return detachActorVolumes(ctx, s.store, state.Actor, state.ActorTemplate, "pause")
+	return detachActorVolumes(ctx, s.store, state.Actor, state.ActorTemplate, ateattr.OperationNamePause)
 }
 
 func (s *DetachVolumesForPauseStep) RetryBackoff() *wait.Backoff { return nil }
@@ -255,6 +257,7 @@ func (s *FinalizePausedStep) Execute(ctx context.Context, input *PauseInput, sta
 		if err != nil {
 			return err
 		}
+		wasAlreadyCrashed := latestActor.GetStatus() == ateapipb.Actor_STATUS_CRASHED
 		latestActor.Status = ateapipb.Actor_STATUS_PAUSED
 		if nodeName == "" {
 			// Without a node name we cannot record where the local snapshot lives,
@@ -275,15 +278,26 @@ func (s *FinalizePausedStep) Execute(ctx context.Context, input *PauseInput, sta
 			latestActor.LocalSnapshotInfo = localInfo
 			latestActor.InProgressSnapshot = ""
 		}
+		sandboxClass := ""
+		if worker != nil {
+			sandboxClass = worker.GetSandboxClass()
+		}
+		// Snapshot crash attributes before pod and pool pointers are cleared below.
+		crashAttrs := ateattr.ActorMetricAttributes(latestActor, sandboxClass, ateattr.OperationNamePause, ateattr.ReasonCorruptedAssignment)
+
 		latestActor.AteomPodNamespace = ""
 		latestActor.AteomPodName = ""
 		latestActor.AteomPodIp = ""
 		latestActor.WorkerPoolName = ""
 		updatedActor, err := s.store.UpdateActor(ctx, latestActor, latestActor.GetMetadata().GetVersion())
+		if err == nil && updatedActor.GetStatus() == ateapipb.Actor_STATUS_CRASHED && !wasAlreadyCrashed {
+			recordActorCrash(ctx, crashAttrs)
+		}
 		if err != nil {
 			return err
 		}
 		latestActor = updatedActor
+
 	}
 
 	state.Actor = latestActor
diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go
index 09a3f134a..b27e30998 100644
--- a/cmd/ateapi/internal/controlapi/workflow_resume.go
+++ b/cmd/ateapi/internal/controlapi/workflow_resume.go
@@ -24,6 +24,7 @@ import (
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/scheduling"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
 	"github.com/agent-substrate/substrate/internal/resources"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
@@ -116,7 +117,7 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput
 				slog.String("AteomPodName", actor.AteomPodName))
 
 			// Crash the actor if its worker assignment is corrupted. We should never be in this state.
-			if cerr := crashActor(ctx, s.store, input.ActorRef); cerr != nil {
+			if cerr := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNameResume, ateattr.ReasonCorruptedAssignment); cerr != nil {
 				return cerr
 			}
 			return status.Errorf(codes.Aborted, "actor %s crashed", input.ActorRef)
@@ -126,7 +127,7 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput
 		if err != nil {
 			// Crash the actor if it was assigned to a deleted pod.
 			if errors.Is(err, store.ErrNotFound) {
-				if cerr := crashActor(ctx, s.store, input.ActorRef); cerr != nil {
+				if cerr := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNameResume, ateattr.ReasonWorkerPodGone); cerr != nil {
 					return cerr
 				}
 				return status.Errorf(codes.Aborted, "actor %s crashed", input.ActorRef)
@@ -137,7 +138,8 @@ func (s *LoadActorForResumeStep) Execute(ctx context.Context, input *ResumeInput
 			slog.InfoContext(ctx, "Assigned worker is draining; crashing actor",
 				slog.String("actor", input.ActorRef.String()),
 				slog.String("worker", wk.GetWorkerNamespace()+"/"+wk.GetWorkerPod()))
-			if cerr := crashActor(ctx, s.store, input.ActorRef); cerr != nil {
+			if cerr := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNameResume, ateattr.ReasonWorkerReassigned); cerr != nil {
+
 				return cerr
 			}
 			return status.Errorf(codes.Aborted, "actor %s crashed", input.ActorRef.String())
@@ -419,7 +421,7 @@ func (s *CallAteletRestoreStep) CheckPrerequisite(ctx context.Context, input *Re
 		slog.ErrorContext(ctx, "crashing actor because its assigned worker no longer belongs to it",
 			slog.String("worker", state.Worker.GetWorkerPod()),
 			slog.Any("assignment", state.Worker.GetAssignment()))
-		if cerr := crashActor(ctx, s.store, input.ActorRef); cerr != nil {
+		if cerr := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNameResume, ateattr.ReasonWorkerReassigned); cerr != nil {
 			return fmt.Errorf("while crashing actor: %w", cerr)
 		}
 		return status.Errorf(codes.Aborted, "actor %s crashed", input.ActorRef)
@@ -439,7 +441,7 @@ func (s *CallAteletRestoreStep) CheckPrerequisite(ctx context.Context, input *Re
 		if err := s.store.UpdateWorker(ctx, release, release.Version); err != nil {
 			return fmt.Errorf("while releasing stale worker assignment: %w", err)
 		}
-		if cerr := crashActor(ctx, s.store, input.ActorRef); cerr != nil {
+		if cerr := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNameResume, ateattr.ReasonCorruptedAssignment); cerr != nil {
 			return fmt.Errorf("while crashing actor: %w", cerr)
 		}
 		return status.Errorf(codes.Aborted, "actor %s crashed", input.ActorRef)
@@ -477,7 +479,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 		req.Scope = toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause)
 
 		_, err = client.Restore(ctx, req)
-		return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while restoring workload")
+		return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while restoring workload", ateattr.OperationNameResume)
 	} else if state.SnapshotLocation != "" {
 		slog.InfoContext(ctx, "Actor has durable snapshot; Restoring from snapshot")
 
@@ -498,7 +500,8 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 			ActorUid: state.Actor.GetMetadata().Uid,
 		}
 		_, err = client.Restore(ctx, req)
-		return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while restoring durable snapshot")
+		return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while restoring durable snapshot", ateattr.OperationNameResume)
+
 	} else {
 		slog.InfoContext(ctx, "Actor has no snapshot; ActorTemplate has no golden snapshot; Booting from ActorTemplate spec")
 
@@ -521,7 +524,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 			ActorUid:               state.Actor.GetMetadata().Uid,
 		}
 		_, err = client.Run(ctx, req)
-		return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while creating workload from spec")
+		return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while creating workload from spec", ateattr.OperationNameResume)
 	}
 	// Unreachable
 }
diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend.go b/cmd/ateapi/internal/controlapi/workflow_suspend.go
index 28b9bf49b..93c5c92cd 100644
--- a/cmd/ateapi/internal/controlapi/workflow_suspend.go
+++ b/cmd/ateapi/internal/controlapi/workflow_suspend.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
 	"github.com/agent-substrate/substrate/internal/resources"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
@@ -126,7 +127,8 @@ func (s *CallAteletSuspendStep) CheckPrerequisite(ctx context.Context, input *Su
 		return status.Errorf(codes.FailedPrecondition, "CallAteletSuspendStep prerequisite not met for Actor: %s (got: %v, want %s)", input.ActorRef, state.Actor.GetStatus(), ateapipb.Actor_STATUS_SUSPENDING)
 	}
 	if state.Actor.GetAteomPodNamespace() == "" || state.Actor.GetAteomPodName() == "" {
-		if err := crashActor(ctx, s.store, input.ActorRef); err != nil {
+		// Missing active worker pod reference in SUSPENDING state indicates corrupted store state.
+		if err := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNameSuspend, ateattr.ReasonCorruptedAssignment); err != nil {
 			slog.ErrorContext(ctx, "Failed to crash actor", slog.String("err", err.Error()))
 		}
 		return fmt.Errorf("actor is CRASHED because it was in SUSPENDING state but has no active worker")
@@ -138,7 +140,7 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
 	if err != nil {
 		if errors.Is(err, ErrWorkerPodNotFound) {
 			slog.ErrorContext(ctx, "Worker pod gone before checkpoint, crashing actor", "namespace", state.Actor.GetAteomPodNamespace(), "pod", state.Actor.GetAteomPodName(), "in_progress_snapshot", state.Actor.GetInProgressSnapshot())
-			if err := crashActor(ctx, s.store, input.ActorRef); err != nil {
+			if err := crashActor(ctx, s.store, input.ActorRef, ateattr.OperationNameSuspend, ateattr.ReasonWorkerPodGone); err != nil {
 				slog.ErrorContext(ctx, "Failed to crash actor", slog.String("err", err.Error()))
 			}
 			return fmt.Errorf("actor is CRASHED because its worker pod is gone and no snapshot was written")
@@ -173,7 +175,7 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
 	}
 
 	_, err = client.Checkpoint(ctx, req)
-	return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while checkpointing workload")
+	return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while checkpointing workload", ateattr.OperationNameSuspend)
 }
 
 func (s *CallAteletSuspendStep) RetryBackoff() *wait.Backoff { return nil }
@@ -194,7 +196,7 @@ func (s *DetachVolumesStep) CheckPrerequisite(ctx context.Context, input *Suspen
 }
 
 func (s *DetachVolumesStep) Execute(ctx context.Context, input *SuspendInput, state *SuspendState) error {
-	return detachActorVolumes(ctx, s.store, state.Actor, state.ActorTemplate, "suspend")
+	return detachActorVolumes(ctx, s.store, state.Actor, state.ActorTemplate, ateattr.OperationNameSuspend)
 }
 
 func (s *DetachVolumesStep) RetryBackoff() *wait.Backoff { return nil }
diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go
index 0db329e1f..1545ab55f 100644
--- a/cmd/ateapi/main.go
+++ b/cmd/ateapi/main.go
@@ -165,6 +165,9 @@ func main() {
 	if err := controlapi.RegisterWorkerCount(otel.Meter("ateapi"), workerCache.Workers, workerPoolLister.List); err != nil {
 		serverboot.Fatal(ctx, "Failed to register worker-count metric", err)
 	}
+	if err := controlapi.RegisterActorCrashes(otel.Meter("ateapi")); err != nil {
+		serverboot.Fatal(ctx, "Failed to register actor-crashes metric", err)
+	}
 
 	ateletDialer := controlapi.NewAteletDialer(workerPodInformer.GetIndexer(), ateletPodInformer.GetIndexer(), *ateletClientCredBundle, *podIdentityCACerts)
 	sm := controlapi.NewService(redisPersistence, workerCache, actorTemplateLister, workerPoolLister, sandboxConfigLister, ateletDialer, clientset)
diff --git a/docs/observability.md b/docs/observability.md
index 78ab8e922..379833f6a 100644
--- a/docs/observability.md
+++ b/docs/observability.md
@@ -110,6 +110,7 @@ Agent Substrate emits foundational OpenTelemetry system and server metrics to mo
 | Metric | Emitted by | Type | Measures |
 |--------|------------|------|----------|
 | `rpc.server.call.duration` | ateapi & atelet (gRPC servers, via `otelgrpc`) | histogram | per-method gRPC latency, request rate, and errors (labels `rpc.method`, `rpc.response.status_code`) |
+| `ate.actor.crashes` | ateapi | counter | Number of times actors transitioned to `STATUS_CRASHED` (labels `ate.operation.name`, `ate.failure.reason`, `ate.template.namespace`, `ate.template.name`, `ate.workerpool.name`, `ate.sandbox.class`) |
 | `atenet.router.route.duration` | atenet-router | histogram | Substrate E2E — Envoy receiving a request to Envoy forwarding it to the resolved worker, excluding actor compute and the response (labels `ate.template.namespace`, `ate.template.name`, `ate.router.outcome`, `ate.router.resume`) |
 | `atelet.snapshot.size` | atelet | histogram | uncompressed size in bytes of each gVisor snapshot image written during checkpoint (labels `kind`, `actor_template_namespace`, `actor_template_name`) |
 
diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go
index e316384b1..8d3a12469 100644
--- a/internal/ateattr/ateattr.go
+++ b/internal/ateattr/ateattr.go
@@ -21,7 +21,9 @@ package ateattr
 import (
 	"go.opentelemetry.io/otel/attribute"
 
+	"github.com/agent-substrate/substrate/internal/ateerrors"
 	"github.com/agent-substrate/substrate/internal/resources"
+
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 )
 
@@ -56,8 +58,18 @@ const (
 	WorkerPoolNameKey      = attribute.Key("ate.workerpool.name")
 	WorkerStateKey         = attribute.Key("ate.worker.state")
 	SandboxClassKey        = attribute.Key("ate.sandbox.class")
-	RouterResumeKey        = attribute.Key("ate.router.resume")
-	RouterOutcomeKey       = attribute.Key("ate.router.outcome")
+	OperationNameKey       = attribute.Key("ate.operation.name")
+	FailureReasonKey       = attribute.Key("ate.failure.reason")
+)
+
+// Control-plane failure reasons for ate.actor.crashes metric.
+const (
+	ReasonCorruptedAssignment = string(ateerrors.ReasonCorruptedAssignment)
+	ReasonWorkerReassigned    = string(ateerrors.ReasonWorkerReassigned)
+	ReasonWorkerPodGone       = string(ateerrors.ReasonWorkerPodGone)
+	ReasonUnknown             = string(ateerrors.ReasonUnknown)
+	RouterResumeKey           = attribute.Key("ate.router.resume")
+	RouterOutcomeKey          = attribute.Key("ate.router.outcome")
 )
 
 // Values for RouterResumeKey.
@@ -77,6 +89,25 @@ const (
 	WorkerStateAssigned = "assigned"
 )
 
+// Values for OperationNameKey representing actor lifecycle operations.
+const (
+	OperationNameResume  = "resume"
+	OperationNameSuspend = "suspend"
+	OperationNamePause   = "pause"
+	OperationNameUnknown = "unknown"
+)
+
+// NormalizeOperationName ensures op is one of the bounded lifecycle operations
+// {resume, suspend, pause, unknown}. Any other value maps to OperationNameUnknown.
+func NormalizeOperationName(op string) string {
+	switch op {
+	case OperationNameResume, OperationNameSuspend, OperationNamePause:
+		return op
+	default:
+		return OperationNameUnknown
+	}
+}
+
 // ActorRefAttributes returns the subset knowable before the Actor record
 // resolves: only the (atespace, name) the request addresses. The uid and version
 // are server-assigned and unknown until the record loads, so they are omitted.
@@ -98,3 +129,26 @@ func ActorAttributes(a *ateapipb.Actor) []attribute.KeyValue {
 		ActorVersionKey.Int64(a.GetMetadata().GetVersion()),
 	}
 }
+
+// ActorMetricAttributes returns the metric labels for an Actor.
+// High-cardinality attributes (atespace, actor name, actor uid) are omitted.
+func ActorMetricAttributes(a *ateapipb.Actor, sandboxClass, operationName, reason string) []attribute.KeyValue {
+	if a == nil {
+		return nil
+	}
+
+	// Default values for unknown/unset attributes.
+	if reason == "" {
+		reason = ReasonUnknown
+	}
+	operationName = NormalizeOperationName(operationName)
+
+	return []attribute.KeyValue{
+		TemplateNamespaceKey.String(a.GetActorTemplateNamespace()),
+		TemplateNameKey.String(a.GetActorTemplateName()),
+		WorkerPoolNameKey.String(a.GetWorkerPoolName()),
+		SandboxClassKey.String(sandboxClass),
+		OperationNameKey.String(operationName),
+		FailureReasonKey.String(reason),
+	}
+}
diff --git a/internal/ateattr/ateattr_test.go b/internal/ateattr/ateattr_test.go
index 8964c248c..eba8331d5 100644
--- a/internal/ateattr/ateattr_test.go
+++ b/internal/ateattr/ateattr_test.go
@@ -152,6 +152,9 @@ func TestKeySpellings(t *testing.T) {
 		{WorkerPoolNameKey, "ate.workerpool.name"},
 		{WorkerStateKey, "ate.worker.state"},
 		{SandboxClassKey, "ate.sandbox.class"},
+		{OperationNameKey, "ate.operation.name"},
+
+		{FailureReasonKey, "ate.failure.reason"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.want, func(t *testing.T) {
@@ -161,3 +164,72 @@ func TestKeySpellings(t *testing.T) {
 		})
 	}
 }
+
+func TestActorMetricAttributes(t *testing.T) {
+	actor := &ateapipb.Actor{
+		ActorTemplateNamespace: "default",
+		ActorTemplateName:      "counter-template",
+		WorkerPoolName:         "default-pool",
+	}
+
+	t.Run("explicit operation and reason", func(t *testing.T) {
+		got := toMap(ActorMetricAttributes(actor, "gvisor", OperationNameResume, ReasonCorruptedAssignment))
+		want := map[attribute.Key]any{
+			TemplateNamespaceKey: "default",
+			TemplateNameKey:      "counter-template",
+			WorkerPoolNameKey:    "default-pool",
+			SandboxClassKey:      "gvisor",
+			OperationNameKey:     OperationNameResume,
+			FailureReasonKey:     ReasonCorruptedAssignment,
+		}
+
+		assertAttrs(t, got, want)
+	})
+
+	t.Run("default unknown values", func(t *testing.T) {
+		got := toMap(ActorMetricAttributes(actor, "gvisor", "", ""))
+		want := map[attribute.Key]any{
+			TemplateNamespaceKey: "default",
+			TemplateNameKey:      "counter-template",
+			WorkerPoolNameKey:    "default-pool",
+			SandboxClassKey:      "gvisor",
+			OperationNameKey:     OperationNameUnknown,
+			FailureReasonKey:     ReasonUnknown,
+		}
+
+		assertAttrs(t, got, want)
+	})
+
+	t.Run("out of range operation name is normalized to unknown", func(t *testing.T) {
+		got := toMap(ActorMetricAttributes(actor, "gvisor", "invalid_op", ""))
+		want := map[attribute.Key]any{
+			TemplateNamespaceKey: "default",
+			TemplateNameKey:      "counter-template",
+			WorkerPoolNameKey:    "default-pool",
+			SandboxClassKey:      "gvisor",
+			OperationNameKey:     OperationNameUnknown,
+			FailureReasonKey:     ReasonUnknown,
+		}
+
+		assertAttrs(t, got, want)
+	})
+}
+
+func TestNormalizeOperationName(t *testing.T) {
+	tests := []struct {
+		op   string
+		want string
+	}{
+		{OperationNameResume, OperationNameResume},
+		{OperationNameSuspend, OperationNameSuspend},
+		{OperationNamePause, OperationNamePause},
+		{"", OperationNameUnknown},
+		{"invalid", OperationNameUnknown},
+		{"crash", OperationNameUnknown},
+	}
+	for _, tt := range tests {
+		if got := NormalizeOperationName(tt.op); got != tt.want {
+			t.Errorf("NormalizeOperationName(%q) = %q, want %q", tt.op, got, tt.want)
+		}
+	}
+}
diff --git a/internal/ateerrors/ateerrors.go b/internal/ateerrors/ateerrors.go
index a1c38bfe7..7aa1c384a 100644
--- a/internal/ateerrors/ateerrors.go
+++ b/internal/ateerrors/ateerrors.go
@@ -22,6 +22,7 @@ import (
 	"slices"
 
 	epb "google.golang.org/genproto/googleapis/rpc/errdetails"
+
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -41,6 +42,7 @@ type Reason string
 // Error makes a Reason wrappable with %w and matchable with errors.Is/As.
 func (r Reason) Error() string { return string(r) }
 
+// NOTE: When adding a Reason constant below, also add it to AllReasons.
 const (
 	ReasonTerminalFileSystemError Reason = "TERMINAL_FILE_SYSTEM_ERROR"
 	ReasonInvalidSandboxAsset     Reason = "INVALID_SANDBOX_ASSET"
@@ -52,8 +54,29 @@ const (
 	// produce a runnable process (e.g. the resolved argv is empty because the
 	// image defines no ENTRYPOINT/CMD and the ActorTemplate sets no command/args).
 	ReasonInvalidContainerConfig Reason = "INVALID_CONTAINER_CONFIG"
+
+	// Control-plane failure reasons for ate.actor.crashes metric.
+	ReasonCorruptedAssignment Reason = "CORRUPTED_ASSIGNMENT"
+	ReasonWorkerReassigned    Reason = "WORKER_REASSIGNED"
+	ReasonWorkerPodGone       Reason = "WORKER_POD_GONE"
+	ReasonUnknown             Reason = "UNKNOWN"
 )
 
+// AllReasons contains all valid Reason constants for validation. Keep in sync with const block above.
+var AllReasons = []Reason{
+	ReasonTerminalFileSystemError,
+	ReasonInvalidSandboxAsset,
+	ReasonInvalidCheckpointResult,
+	ReasonFaileSaveSnapshot,
+	ReasonInvalidObjectURL,
+	ReasonFailedGetExternalObject,
+	ReasonInvalidContainerConfig,
+	ReasonCorruptedAssignment,
+	ReasonWorkerReassigned,
+	ReasonWorkerPodGone,
+	ReasonUnknown,
+}
+
 // MetadataKeyActorCrashed marks (in ErrorInfo.Metadata) a failure that requires
 // the control plane to crash the actor.
 const MetadataKeyActorCrashed = "actorCrashed"
@@ -127,3 +150,31 @@ func ActorCrashRequested(err error) bool {
 	}
 	return false
 }
+
+// IsValidReason reports whether a string matches a known ateerrors.Reason enum.
+func IsValidReason(s string) bool {
+	return slices.Contains(AllReasons, Reason(s))
+}
+
+// ExtractReason returns the validated enum reason string from an error's AIP-193 ErrorInfo detail
+// or wrapped ateerrors.Reason, or empty string if unclassified.
+func ExtractReason(err error) string {
+	if err == nil {
+		return ""
+	}
+	var r Reason
+	if errors.As(err, &r) && IsValidReason(string(r)) {
+		return string(r)
+	}
+	st, ok := status.FromError(err)
+	if ok {
+		for _, d := range st.Details() {
+			if info, ok := d.(*epb.ErrorInfo); ok {
+				if rStr := info.GetReason(); rStr != "" && IsValidReason(rStr) {
+					return rStr
+				}
+			}
+		}
+	}
+	return ""
+}
diff --git a/internal/ateerrors/ateerrors_test.go b/internal/ateerrors/ateerrors_test.go
index 7e8fc9d25..ab3de6154 100644
--- a/internal/ateerrors/ateerrors_test.go
+++ b/internal/ateerrors/ateerrors_test.go
@@ -287,3 +287,35 @@ func TestActorCrashRequested(t *testing.T) {
 		})
 	}
 }
+
+func TestExtractReason_EnforcesAllowedEnumValuesOnly(t *testing.T) {
+	t.Run("valid enum reason returned", func(t *testing.T) {
+		err := NewGRPCError(context.Background(), codes.DataLoss, ReasonFaileSaveSnapshot, nil, errors.New("boom"))
+		if got := ExtractReason(err); got != "FAILED_SAVE_SNAPSHOT" {
+			t.Errorf("ExtractReason(%v) = %q, want %q", err, got, "FAILED_SAVE_SNAPSHOT")
+		}
+	})
+
+	t.Run("unlisted dynamic reason rejected to prevent metric high cardinality", func(t *testing.T) {
+		err := NewGRPCError(context.Background(), codes.DataLoss, Reason("UNLISTED_DYNAMIC_ERROR_STRING"), nil, errors.New("boom"))
+		if got := ExtractReason(err); got != "" {
+			t.Errorf("ExtractReason(%v) = %q, want %q (empty string)", err, got, "")
+		}
+	})
+}
+
+func TestAllReasonsRegistered(t *testing.T) {
+	if len(AllReasons) == 0 {
+		t.Fatal("AllReasons slice is empty")
+	}
+
+	for _, r := range AllReasons {
+		if !IsValidReason(string(r)) {
+			t.Errorf("IsValidReason(%q) = false, want true", r)
+		}
+		err := NewGRPCError(context.Background(), codes.DataLoss, r, nil, errors.New("boom"))
+		if got := ExtractReason(err); got != string(r) {
+			t.Errorf("ExtractReason for %q = %q, want %q", r, got, r)
+		}
+	}
+}
diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go
index c6dbd9034..1d46719cd 100644
--- a/internal/e2e/collector_metrics.go
+++ b/internal/e2e/collector_metrics.go
@@ -41,6 +41,7 @@ const (
 // it pins the worker-count instrument introduced alongside this harness.
 var PlatformMetricPrefixes = []string{
 	"ate_workerpool_workers",
+	"ate_actor_crashes",
 	"atenet_router_route_duration",
 }
 
diff --git a/internal/e2e/suites/metrics/metrics_test.go b/internal/e2e/suites/metrics/metrics_test.go
index 46a912eee..4bd54062c 100644
--- a/internal/e2e/suites/metrics/metrics_test.go
+++ b/internal/e2e/suites/metrics/metrics_test.go
@@ -22,13 +22,19 @@ package metrics
 
 import (
 	"context"
+	"fmt"
 	"os"
+
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/agent-substrate/substrate/internal/ateattr"
+	"github.com/agent-substrate/substrate/internal/ateerrors"
 	"github.com/agent-substrate/substrate/internal/e2e"
 	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 const metricsAtespace = "ate-metrics-e2e"
@@ -48,7 +54,7 @@ func TestPlatformMetricsEmitted(t *testing.T) {
 	ctx := context.Background()
 	clients := e2e.GetClients()
 	tmplNS, tmplName := templateRef()
-	actorID := "metrics-probe"
+	actorID := fmt.Sprintf("metrics-probe-%d", time.Now().UnixNano())
 
 	// CreateActor requires the atespace to exist first; ignore AlreadyExists.
 	_, _ = clients.SubstrateAPI.CreateAtespace(ctx, &ateapipb.CreateAtespaceRequest{
@@ -84,6 +90,9 @@ func TestPlatformMetricsEmitted(t *testing.T) {
 	}
 	_ = resp.Body.Close()
 
+	// Trigger an actor crash to verify ate_actor_crashes counter emission.
+	triggerActorCrash(t, ctx, clients, actorID)
+
 	deadline := time.Now().Add(2 * time.Minute)
 	var missing []string
 	var ateomSeen bool
@@ -95,13 +104,68 @@ func TestPlatformMetricsEmitted(t *testing.T) {
 		missing = e2e.MissingPlatformMetrics(scrape, e2e.PlatformMetricPrefixes)
 		ateomSeen = e2e.CollectorHasService(scrape, "ateom-gvisor", "ateom-microvm")
 		if len(missing) == 0 && ateomSeen {
-			return
+			// Verify ate_actor_crashes metric carries valid, non-empty low-cardinality labels for all attributes.
+			for _, line := range strings.Split(scrape, "\n") {
+				if strings.HasPrefix(line, "ate_actor_crashes") &&
+					strings.Contains(line, `ate_operation_name=`) &&
+					strings.Contains(line, `ate_failure_reason=`) &&
+					strings.Contains(line, `ate_template_namespace=`) &&
+					strings.Contains(line, `ate_template_name=`) &&
+					strings.Contains(line, `ate_workerpool_name=`) &&
+					strings.Contains(line, `ate_sandbox_class=`) {
+
+					opVal := extractPrometheusLabelValue(line, "ate_operation_name")
+					reasonVal := extractPrometheusLabelValue(line, "ate_failure_reason")
+					tmplNSVal := extractPrometheusLabelValue(line, "ate_template_namespace")
+					tmplNameVal := extractPrometheusLabelValue(line, "ate_template_name")
+					workerPoolVal := extractPrometheusLabelValue(line, "ate_workerpool_name")
+					sandboxVal := extractPrometheusLabelValue(line, "ate_sandbox_class")
+
+					if ateattr.NormalizeOperationName(opVal) == opVal &&
+						ateerrors.IsValidReason(reasonVal) &&
+						tmplNSVal != "" &&
+						tmplNameVal != "" &&
+						workerPoolVal != "" &&
+						sandboxVal != "" {
+						return
+					}
+				}
+			}
 		}
+
 		time.Sleep(3 * time.Second)
 	}
+
 	t.Fatalf("platform telemetry never reached the collector: missing metrics %v, ateom pushed=%v", missing, ateomSeen)
 }
 
+func triggerActorCrash(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string) {
+	t.Helper()
+
+	// Get running actor to find its assigned worker pod.
+	actor, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{
+		Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID},
+	})
+	if err != nil {
+		t.Fatalf("GetActor failed: %v", err)
+	}
+
+	// Delete the assigned worker pod directly.
+	// The WorkerPoolSyncer will detect the pod is gone, crash the actor via syncer.go,
+	// and emit the ate_actor_crashes counter.
+	if podName := actor.GetAteomPodName(); podName != "" {
+		podNS := actor.GetAteomPodNamespace()
+		if err := clients.K8s.CoreV1().Pods(podNS).Delete(ctx, podName, metav1.DeleteOptions{}); err != nil {
+
+			t.Fatalf("Delete worker pod failed: %v", err)
+		}
+	} else {
+		t.Fatalf("Actor %s has no assigned worker pod to delete", actorID)
+	}
+
+	waitForStatus(t, ctx, clients, actorID, ateapipb.Actor_STATUS_CRASHED)
+}
+
 func resume(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string) {
 	t.Helper()
 	if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
@@ -126,3 +190,17 @@ func waitForStatus(t *testing.T, ctx context.Context, clients *e2e.Clients, acto
 	}
 	t.Fatalf("actor %q never reached %v", actorID, want)
 }
+
+func extractPrometheusLabelValue(line, labelName string) string {
+	key := labelName + `="`
+	idx := strings.Index(line, key)
+	if idx == -1 {
+		return ""
+	}
+	start := idx + len(key)
+	end := strings.IndexByte(line[start:], '"')
+	if end == -1 {
+		return ""
+	}
+	return line[start : start+end]
+}