diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go
index 55429e6aa..dcceb82d8 100644
--- a/cmd/ateapi/internal/controlapi/functional_test.go
+++ b/cmd/ateapi/internal/controlapi/functional_test.go
@@ -293,7 +293,7 @@ func setupTest(t *testing.T, ns string) *testContext {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	syncer := NewWorkerPoolSyncer(persistence, workerInformer, workerPoolLister)
+	syncer := NewWorkerPoolSyncer(persistence, k8sClient, workerInformer, workerPoolLister)
 	syncer.Start(ctx)
 
 	workerFactory.Start(ctx.Done())
@@ -2446,6 +2446,15 @@ func TestResumeActor_DanglingWorker(t *testing.T) {
 
 	deleteWorkerPod(t, tc, ns, "worker-a")
 
+	// 5. Wait for the syncer to remove worker A from the store. The actor is
+	// RESUMING, not RUNNING, so the syncer leaves it bound to the dead worker.
+	if err := wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) {
+		_, gerr := tc.persistence.GetWorker(ctx, ns, "pool1", "worker-a")
+		return gerr != nil, nil
+	}); err != nil {
+		t.Fatalf("worker-a not removed from store: %v", err)
+	}
+
 	// 6. Create Worker Pod B
 	createWorkerPod(t, tc, ns, "worker-b", "node1", "pool1")
 
@@ -2453,28 +2462,30 @@ func TestResumeActor_DanglingWorker(t *testing.T) {
 	tc.fakeAtelet.FailRestore = nil
 	tc.fakeAtelet.RestoreCalled = false // reset
 
-	// 8. Call ResumeActor again -> Expect success and picking Worker B!
+	// 8. Call ResumeActor again -> Expect failure: the resume workflow refuses
+	// to reassign a RESUMING actor whose recorded worker is gone, even though
+	// a free worker is available.
 	_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
 		Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name},
 	})
-	if err != nil {
-		t.Fatalf("ResumeActor failed on retry: %v", err)
+	if err == nil {
+		t.Fatalf("expected ResumeActor retry to fail for actor bound to a deleted worker")
 	}
 
-	if !tc.fakeAtelet.RestoreCalled {
-		t.Errorf("expected Restore to be called on retry")
+	if tc.fakeAtelet.RestoreCalled {
+		t.Errorf("expected Restore not to be called on retry")
 	}
 
-	// Verify actor state is RUNNING with worker B assigned
+	// Verify the actor is still RESUMING and pinned to the dead worker A.
 	actor, err = tc.persistence.GetActor(context.Background(), testAtespace, name)
 	if err != nil {
 		t.Fatalf("failed to get actor from store: %v", err)
 	}
-	if actor.GetStatus() != ateapipb.Actor_STATUS_RUNNING {
-		t.Errorf("expected status RUNNING, got %v", actor.GetStatus())
+	if actor.GetStatus() != ateapipb.Actor_STATUS_RESUMING {
+		t.Errorf("expected status RESUMING, got %v", actor.GetStatus())
 	}
-	if actor.GetAteomPodName() != "worker-b" {
-		t.Errorf("expected worker-b assigned, got %v", actor.GetAteomPodName())
+	if actor.GetAteomPodName() != "worker-a" {
+		t.Errorf("expected worker-a still assigned, got %v", actor.GetAteomPodName())
 	}
 }
 
@@ -2508,29 +2519,34 @@ func TestSuspendActor_DanglingWorker(t *testing.T) {
 
 	deleteWorkerPod(t, tc, ns, "worker-1")
 
-	// 3. Call SuspendActor -> Should succeed (our fix skips missing pod execution)
-	actors, _, _ := tc.persistence.ListActors(context.Background(), testAtespace, maxPageSize, "")
-	t.Logf("Actors in Redis before Suspend: %d", len(actors))
-	for _, a := range actors {
-		t.Logf("  Actor: %s/%s/%s", a.GetActorTemplateNamespace(), a.GetActorTemplateName(), a.GetMetadata().GetName())
+	// 3. The syncer crashes a RUNNING actor whose worker pod vanished.
+	if err := wait.PollUntilContextTimeout(context.Background(), 100*time.Millisecond, 5*time.Second, true, func(ctx context.Context) (bool, error) {
+		a, gerr := tc.persistence.GetActor(ctx, testAtespace, name)
+		if gerr != nil {
+			return false, gerr
+		}
+		return a.GetStatus() == ateapipb.Actor_STATUS_CRASHED, nil
+	}); err != nil {
+		t.Fatalf("actor not marked CRASHED after worker deletion: %v", err)
 	}
 
+	// 4. SuspendActor now fails: there is no live state left to snapshot.
 	_, err = tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{
 		Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name},
 	})
-	if err != nil {
-		t.Fatalf("SuspendActor failed: %v", err)
+	if err == nil {
+		t.Fatalf("expected SuspendActor to fail for a crashed actor")
 	}
 
-	// 4. Verify it becomes SUSPENDED in Redis
+	// 5. Verify the crash stuck and the dead worker binding is cleared.
 	getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
 		Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name},
 	})
 	if err != nil {
 		t.Fatalf("GetActor failed: %v", err)
 	}
-	if getResp.GetStatus() != ateapipb.Actor_STATUS_SUSPENDED {
-		t.Errorf("expected status SUSPENDED, got %v", getResp.GetStatus())
+	if getResp.GetStatus() != ateapipb.Actor_STATUS_CRASHED {
+		t.Errorf("expected status CRASHED, got %v", getResp.GetStatus())
 	}
 	if getResp.GetAteomPodNamespace() != "" {
 		t.Errorf("expected ateom_pod_namespace to be empty, got %v", getResp.GetAteomPodNamespace())
diff --git a/cmd/ateapi/internal/controlapi/syncer.go b/cmd/ateapi/internal/controlapi/syncer.go
index 82854f826..8a2315b2e 100644
--- a/cmd/ateapi/internal/controlapi/syncer.go
+++ b/cmd/ateapi/internal/controlapi/syncer.go
@@ -27,6 +27,9 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 )
 
@@ -34,14 +37,16 @@ import (
 // into the store.
 type WorkerPoolSyncer struct {
 	persistence      store.Interface
+	kubeClient       kubernetes.Interface
 	workerInformer   cache.SharedIndexInformer
 	workerPoolLister listersv1alpha1.WorkerPoolLister
 }
 
 // NewWorkerPoolSyncer creates a new WorkerPoolSyncer.
-func NewWorkerPoolSyncer(persistence store.Interface, workerInformer cache.SharedIndexInformer, workerPoolLister listersv1alpha1.WorkerPoolLister) *WorkerPoolSyncer {
+func NewWorkerPoolSyncer(persistence store.Interface, kubeClient kubernetes.Interface, workerInformer cache.SharedIndexInformer, workerPoolLister listersv1alpha1.WorkerPoolLister) *WorkerPoolSyncer {
 	return &WorkerPoolSyncer{
 		persistence:      persistence,
+		kubeClient:       kubeClient,
 		workerInformer:   workerInformer,
 		workerPoolLister: workerPoolLister,
 	}
@@ -75,8 +80,8 @@ func (s *WorkerPoolSyncer) Start(ctx context.Context) {
 				return
 			}
 			slog.InfoContext(ctx, "Syncer: removing worker from store", slog.String("worker", pod.Namespace+"/"+pod.Name))
-			if err := s.releaseActorOnDeadWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil {
-				slog.ErrorContext(ctx, "Failed to release actor bound to deleted worker", slog.Any("err", err))
+			if err := s.markActorCrashed(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil {
+				slog.ErrorContext(ctx, "Failed to crash actor bound to deleted worker", slog.Any("err", err))
 			}
 			err := s.persistence.DeleteWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name)
 			if err != nil {
@@ -107,7 +112,7 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po
 
 	if pod.DeletionTimestamp != nil {
 		slog.InfoContext(ctx, "Syncer: removing worker from store (pod deleting)", slog.String("worker", pod.Namespace+"/"+pod.Name))
-		if err := s.releaseActorOnDeadWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil {
+		if err := s.markActorCrashed(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil {
 			slog.ErrorContext(ctx, "Failed to release actor bound to soft-deleting worker", slog.Any("err", err))
 		}
 		err := s.persistence.DeleteWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name)
@@ -117,6 +122,12 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po
 		return
 	}
 
+	// If either the ateom server itself or the actor inside it terminated, crash the actor running on the worker.
+	if ateomTerminated(pod) {
+		s.handleTerminatedAteom(ctx, pod)
+		return
+	}
+
 	poolName := pod.Labels[workerPodLabel]
 	pool, err := s.workerPoolLister.WorkerPools(pod.Namespace).Get(poolName)
 	if err != nil {
@@ -185,21 +196,58 @@ func isWorkerEligible(pod *corev1.Pod) bool {
 	return pod.Status.PodIP != ""
 }
 
-// releaseActorOnDeadWorker resets the actor bound to a vanishing worker
-// pod back to STATUS_SUSPENDED so the next request reassigns it.
-//
-// UpdateActor uses optimistic version checking. A concurrent SuspendActor
-// or ResumeActor wins; we drop this attempt silently.
-//
-// Best-effort only. The caller always proceeds to DeleteWorker after this
-// returns, so any non-contention failure leaves the actor stranded
-// (STATUS_RUNNING, pointer at a pod that no longer exists). Recovery
-// then needs a manual SuspendActor.
-//
-// The long-term fix is a finalizer-based controller that holds the pod
-// in Terminating state until the actor is gracefully suspended. Tracked
-// in https://github.com/agent-substrate/substrate/issues/23.
-func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespace, pool, podName string) error {
+const ateomContainerName = "ateom"
+
+func ateomTerminated(pod *corev1.Pod) bool {
+	for i := range pod.Status.ContainerStatuses {
+		cs := &pod.Status.ContainerStatuses[i]
+		if cs.Name != "ateom" {
+			continue
+		}
+		if cs.LastTerminationState.Terminated != nil || cs.State.Terminated != nil {
+			return true
+		}
+	}
+	return false
+}
+
+func (s *WorkerPoolSyncer) handleTerminatedAteom(ctx context.Context, pod *corev1.Pod) {
+	slog.InfoContext(ctx, "Syncer: worker ateom terminated; crashing actor and recycling pod",
+		slog.String("worker", pod.Namespace+"/"+pod.Name))
+	pool := pod.Labels[workerPodLabel]
+	worker, err := s.persistence.GetWorker(ctx, pod.Namespace, pool, pod.Name)
+	switch {
+	case errors.Is(err, store.ErrNotFound):
+		// Worker already deleted from store, nothing to do.
+		slog.WarnContext(ctx, "worker is already deleted from store")
+	case err != nil:
+		slog.ErrorContext(ctx, "Failed to get worker for terminated ateom; keeping pod so the resync retries", slog.Any("err", err))
+		return
+	case worker.Assignment == nil:
+		// If the worker is not assigned to any actor, delete the worker from pool.
+		if err := s.persistence.DeleteWorker(ctx, pod.Namespace, pool, pod.Name); err != nil {
+			slog.ErrorContext(ctx, "Failed to delete idle crashed worker from store; keeping pod so the resync retries", slog.Any("err", err))
+			return
+		}
+	default:
+		// Crash the actor first, if the ateapi server is restarted. If we deleted the worker first, and ate apiserver restarted, we wouldn't be able to crash the Actor anymore.
+		if err := s.markActorCrashed(ctx, pod.Namespace, pool, pod.Name); err != nil {
+			slog.ErrorContext(ctx, "Failed to crash actor bound to crashed worker; keeping pod so the resync retries", slog.Any("err", err))
+			return
+		}
+		if err := s.persistence.DeleteWorker(ctx, pod.Namespace, pool, pod.Name); err != nil {
+			slog.ErrorContext(ctx, "Failed to delete crashed worker from store", slog.Any("err", err))
+		}
+	}
+	// Deleting the pod, because the actor network may not have been cleaned up from the Ateom Pod's net NS.
+	if err := s.kubeClient.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{
+		Preconditions: metav1.NewUIDPreconditions(string(pod.UID)),
+	}); err != nil && !apierrors.IsNotFound(err) {
+		slog.ErrorContext(ctx, "Failed to delete crashed worker pod", slog.Any("err", err))
+	}
+}
+
+func (s *WorkerPoolSyncer) markActorCrashed(ctx context.Context, namespace, pool, podName string) error {
 	worker, err := s.persistence.GetWorker(ctx, namespace, pool, podName)
 	if err != nil {
 		if errors.Is(err, store.ErrNotFound) {
@@ -210,24 +258,36 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
 	if worker.Assignment == nil {
 		return nil
 	}
-	actor, err := s.persistence.GetActor(ctx, worker.Assignment.Actor.Atespace, worker.Assignment.Actor.Name)
+	atespace := worker.GetAssignment().GetActor().GetAtespace()
+	actorName := worker.GetAssignment().GetActor().GetName()
+	if atespace == "" || actorName == "" {
+		slog.WarnContext(ctx, "cannot release actor on worker, since either atespace or actor name is empty", slog.String("atespace", atespace), slog.String("actor name", actorName))
+		return nil
+	}
+	actor, err := s.persistence.GetActor(ctx, atespace, actorName)
 	if err != nil {
 		if errors.Is(err, store.ErrNotFound) {
 			return nil
 		}
 		return err
 	}
-	// Skip if a concurrent SuspendActor already cleared the pointer.
-	if actor.GetAteomPodNamespace() != namespace || actor.GetAteomPodName() != podName {
+	// Only crash the actor if it's still assigned to the same worker.
+	if actor.GetAteomPodNamespace() == "" || actor.GetAteomPodName() == "" || actor.GetAteomPodUid() == "" {
+		slog.WarnContext(ctx, "cannot release actor on worker", slog.String("ateom pod namesace", actor.GetAteomPodNamespace()), slog.String("ateom name", actor.GetAteomPodName()), slog.String("ateom pod uid", actor.GetAteomPodUid()))
 		return nil
 	}
-	if actor.Status != ateapipb.Actor_STATUS_CRASHED {
-		actor.Status = ateapipb.Actor_STATUS_SUSPENDED
+	if actor.GetStatus() != ateapipb.Actor_STATUS_RUNNING {
+		slog.WarnContext(ctx, "actor is not in running state")
+		return nil
+	}
+	if actor.GetAteomPodNamespace() != worker.GetWorkerNamespace() || actor.GetAteomPodName() != worker.GetWorkerPod() || actor.GetAteomPodUid() != worker.GetWorkerPodUid() {
+		return nil
 	}
+	actor.Status = ateapipb.Actor_STATUS_CRASHED
 	actor.AteomPodNamespace = ""
 	actor.AteomPodName = ""
 	actor.AteomPodIp = ""
-	actor.InProgressSnapshot = ""
+	actor.AteomPodUid = ""
 	actor.WorkerPoolName = ""
 	if _, err := s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()); err != nil && !errors.Is(err, store.ErrPersistenceRetry) {
 		return err
diff --git a/cmd/ateapi/internal/controlapi/syncer_test.go b/cmd/ateapi/internal/controlapi/syncer_test.go
index 6452bfaf3..7c088c2ac 100644
--- a/cmd/ateapi/internal/controlapi/syncer_test.go
+++ b/cmd/ateapi/internal/controlapi/syncer_test.go
@@ -29,12 +29,18 @@ import (
 	"github.com/agent-substrate/substrate/pkg/client/informers/externalversions"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes/fake"
+	k8stesting "k8s.io/client-go/testing"
 )
 
+// testPodUID is the UID shared by the worker pods created in these tests and
+// the actor/worker rows bound to them.
+const testPodUID = "08675309-4a65-6e6e-7973-6e756d626572"
+
 // setupSyncerTest sets up a real store with fake Redis and a fake K8s client with informer.
 func setupSyncerTest(t *testing.T, ctx context.Context, initPools ...*atev1alpha1.WorkerPool) (store.Interface, *fake.Clientset, *atefake.Clientset, func()) {
 	t.Helper()
@@ -53,7 +59,7 @@ func setupSyncerTest(t *testing.T, ctx context.Context, initPools ...*atev1alpha
 	ateInformerFactory := externalversions.NewSharedInformerFactory(fakeAte, 0)
 	workerPoolLister := ateInformerFactory.Api().V1alpha1().WorkerPools().Lister()
 
-	syncer := NewWorkerPoolSyncer(persistence, workerInformer, workerPoolLister)
+	syncer := NewWorkerPoolSyncer(persistence, fakeK8s, workerInformer, workerPoolLister)
 	syncer.Start(ctx)
 
 	workerFactory.Start(ctx.Done())
@@ -101,7 +107,7 @@ func TestSyncer_Lifecycle(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      podName,
 			Namespace: ns,
-			UID:       "08675309-4a65-6e6e-7973-6e756d626572",
+			UID:       testPodUID,
 			Labels: map[string]string{
 				workerPodLabel: poolName,
 			},
@@ -215,7 +221,7 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      pod,
 				Namespace: ns,
-				UID:       "08675309-4a65-6e6e-7973-6e756d626572",
+				UID:       testPodUID,
 				Labels:    map[string]string{workerPodLabel: pool},
 			},
 			Spec: corev1.PodSpec{
@@ -239,7 +245,7 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
 	if _, err := persistence.CreateActor(ctx, &ateapipb.Actor{
 		Metadata: &ateapipb.ResourceMetadata{Name: actorName, Atespace: "team-orphan"}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
 		Status:            ateapipb.Actor_STATUS_RUNNING,
-		AteomPodNamespace: ns, AteomPodName: pod, AteomPodIp: ip,
+		AteomPodNamespace: ns, AteomPodName: pod, AteomPodIp: ip, AteomPodUid: testPodUID,
 		InProgressSnapshot: "gs://snapshots/partial",
 		LatestSnapshotInfo: &ateapipb.SnapshotInfo{
 			Data: &ateapipb.SnapshotInfo_External{
@@ -269,6 +275,8 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
 	if err := fakeK8s.CoreV1().Pods(ns).Delete(ctx, pod, metav1.DeleteOptions{}); err != nil {
 		t.Fatalf("delete pod: %v", err)
 	}
+	// The actor was RUNNING when its worker vanished, so state since the
+	// last snapshot is lost: the release must surface that as CRASHED.
 	var got *ateapipb.Actor
 	if err := wait.PollUntilContextTimeout(ctx, 50*time.Millisecond, 2*time.Second, true, func(c context.Context) (bool, error) {
 		a, gerr := persistence.GetActor(c, "team-orphan", actorName)
@@ -276,11 +284,11 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
 			return false, gerr
 		}
 		got = a
-		return a.GetStatus() == ateapipb.Actor_STATUS_SUSPENDED, nil
+		return a.GetStatus() == ateapipb.Actor_STATUS_CRASHED, nil
 	}); err != nil {
-		t.Fatalf("actor not reset to SUSPENDED: %v", err)
+		t.Fatalf("actor not marked CRASHED: %v", err)
 	}
-	if got.AteomPodName != "" || got.AteomPodNamespace != "" || got.AteomPodIp != "" || got.InProgressSnapshot != "" {
+	if got.AteomPodName != "" || got.AteomPodNamespace != "" || got.AteomPodIp != "" {
 		t.Errorf("bind fields not cleared: %+v", got)
 	}
 	if got.GetLatestSnapshotInfo().GetExternal().SnapshotUriPrefix == "" {
@@ -288,6 +296,422 @@ func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
 	}
 }
 
+func TestAteomTerminated(t *testing.T) {
+	terminated := func(exitCode int32) corev1.ContainerState {
+		return corev1.ContainerState{
+			Terminated: &corev1.ContainerStateTerminated{ExitCode: exitCode},
+		}
+	}
+	tests := []struct {
+		name     string
+		statuses []corev1.ContainerStatus
+		want     bool
+	}{
+		{name: "no statuses", statuses: nil, want: false},
+		{name: "running", statuses: []corev1.ContainerStatus{
+			{Name: ateomContainerName, State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}},
+		}, want: false},
+		{name: "wrong container", statuses: []corev1.ContainerStatus{
+			{Name: "other", LastTerminationState: terminated(1)},
+		}, want: false},
+		{name: "exit 0 in last state", statuses: []corev1.ContainerStatus{
+			{Name: ateomContainerName, LastTerminationState: terminated(0)},
+		}, want: true},
+		{name: "exit 1 in last state", statuses: []corev1.ContainerStatus{
+			{Name: ateomContainerName, LastTerminationState: terminated(1)},
+		}, want: true},
+		{name: "exit 1 in current state", statuses: []corev1.ContainerStatus{
+			{Name: ateomContainerName, State: terminated(1)},
+		}, want: true},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			pod := &corev1.Pod{Status: corev1.PodStatus{ContainerStatuses: tc.statuses}}
+			if got := ateomTerminated(pod); got != tc.want {
+				t.Errorf("ateomTerminated() = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestSyncer_AteomTerminated covers the crash reaction end to end through
+// the informer: any termination of the ateom container — whatever the exit
+// code — marks the assigned actor CRASHED and recycles the doomed pod.
+func TestSyncer_AteomTerminated(t *testing.T) {
+	tests := []struct {
+		name       string
+		exitCode   int32
+		wantStatus ateapipb.Actor_Status
+	}{
+		{
+			name:       "nonzero exit crashes actor",
+			exitCode:   1,
+			wantStatus: ateapipb.Actor_STATUS_CRASHED,
+		},
+		{
+			name:       "zero exit crashes actor",
+			exitCode:   0,
+			wantStatus: ateapipb.Actor_STATUS_CRASHED,
+		},
+	}
+	for i, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+
+			ns, pool, pod, ip := fmt.Sprintf("ns-crash-%d", i), "pool1", "worker-crash", "10.0.0.2"
+			workerPool := &atev1alpha1.WorkerPool{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      pool,
+					Namespace: ns,
+				},
+				Spec: atev1alpha1.WorkerPoolSpec{
+					SandboxClass: "gvisor",
+				},
+			}
+
+			persistence, fakeK8s, _, cleanup := setupSyncerTest(t, ctx, workerPool)
+			defer cleanup()
+			created, err := fakeK8s.CoreV1().Pods(ns).Create(ctx,
+				&corev1.Pod{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      pod,
+						Namespace: ns,
+						UID:       testPodUID,
+						Labels:    map[string]string{workerPodLabel: pool},
+					},
+					Spec: corev1.PodSpec{
+						NodeName:   "node1",
+						Containers: []corev1.Container{{Name: ateomContainerName, Image: "ateom"}},
+					},
+					Status: corev1.PodStatus{
+						Phase: corev1.PodRunning, PodIP: ip,
+						PodIPs: []corev1.PodIP{{IP: ip}},
+					},
+				}, metav1.CreateOptions{})
+			if err != nil {
+				t.Fatalf("create pod: %v", err)
+			}
+			if err := wait.PollUntilContextTimeout(ctx, 50*time.Millisecond, 2*time.Second, true, func(c context.Context) (bool, error) {
+				_, gerr := persistence.GetWorker(c, ns, pool, pod)
+				return gerr == nil, nil
+			}); err != nil {
+				t.Fatalf("worker row not materialised: %v", err)
+			}
+			actorName := "actor-crash"
+			_, err = persistence.CreateActor(ctx, &ateapipb.Actor{
+				Metadata: &ateapipb.ResourceMetadata{Name: actorName, Atespace: "team-crash"}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
+				Status:            ateapipb.Actor_STATUS_RUNNING,
+				AteomPodNamespace: ns, AteomPodName: pod, AteomPodIp: ip, AteomPodUid: string(created.UID),
+				LatestSnapshotInfo: &ateapipb.SnapshotInfo{
+					Data: &ateapipb.SnapshotInfo_External{
+						External: &ateapipb.ExternalSnapshotInfo{
+							SnapshotUriPrefix: "gs://snapshots/last",
+						},
+					},
+				},
+			})
+			if err != nil {
+				t.Fatalf("create actor: %v", err)
+			}
+			w, _ := persistence.GetWorker(ctx, ns, pool, pod)
+			w.Assignment = &ateapipb.Assignment{
+				ActorTemplate: &ateapipb.KubeNamespacedObjectRef{
+					Namespace: ns,
+					Name:      "tmpl",
+				},
+				Actor: &ateapipb.ObjectRef{
+					Name:     actorName,
+					Atespace: "team-crash",
+				},
+			}
+			if err := persistence.UpdateWorker(ctx, w, w.Version); err != nil {
+				t.Fatalf("update worker: %v", err)
+			}
+
+			// ateom exits and kubelet restarts it in place: the pod stays
+			// Running and the termination record lands in lastState.
+			crashed := created.DeepCopy()
+			crashed.Status.ContainerStatuses = []corev1.ContainerStatus{{
+				Name:         ateomContainerName,
+				RestartCount: 1,
+				LastTerminationState: corev1.ContainerState{
+					Terminated: &corev1.ContainerStateTerminated{
+						ExitCode: tc.exitCode,
+					},
+				},
+			}}
+			if _, err := fakeK8s.CoreV1().Pods(ns).Update(ctx, crashed, metav1.UpdateOptions{}); err != nil {
+				t.Fatalf("update pod with crash status: %v", err)
+			}
+
+			var got *ateapipb.Actor
+			if err := wait.PollUntilContextTimeout(ctx, 50*time.Millisecond, 2*time.Second, true, func(c context.Context) (bool, error) {
+				a, gerr := persistence.GetActor(c, "team-crash", actorName)
+				if gerr != nil {
+					return false, gerr
+				}
+				got = a
+				return a.GetStatus() == tc.wantStatus, nil
+			}); err != nil {
+				t.Fatalf("actor status = %v, want %v: %v", got.GetStatus(), tc.wantStatus, err)
+			}
+			if got.AteomPodName != "" || got.AteomPodNamespace != "" || got.AteomPodIp != "" || got.WorkerPoolName != "" {
+				t.Errorf("bind fields not cleared: %+v", got)
+			}
+			if got.GetLatestSnapshotInfo().GetExternal().SnapshotUriPrefix == "" {
+				t.Errorf("External SnapshotUriPrefix must be preserved")
+			}
+			if err := wait.PollUntilContextTimeout(ctx, 50*time.Millisecond, 2*time.Second, true, func(c context.Context) (bool, error) {
+				_, gerr := persistence.GetWorker(c, ns, pool, pod)
+				return errors.Is(gerr, store.ErrNotFound), nil
+			}); err != nil {
+				t.Fatalf("worker row not deleted from store: %v", err)
+			}
+			if err := wait.PollUntilContextTimeout(ctx, 50*time.Millisecond, 2*time.Second, true, func(c context.Context) (bool, error) {
+				_, gerr := fakeK8s.CoreV1().Pods(ns).Get(c, pod, metav1.GetOptions{})
+				return apierrors.IsNotFound(gerr), nil
+			}); err != nil {
+				t.Fatalf("crashed worker pod not deleted: %v", err)
+			}
+		})
+	}
+}
+
+// TestMarkActorCrashed_Crashed covers the actor status write without
+// going through informer events: only a RUNNING actor bound to a vanished
+// worker is crashed. Actors in transition states (SUSPENDING/RESUMING) are
+// owned by each workflow itself.
+func TestMarkActorCrashed_Crashed(t *testing.T) {
+	tests := []struct {
+		name       string
+		status     ateapipb.Actor_Status
+		wantStatus ateapipb.Actor_Status
+	}{
+		{
+			name:       "running actor",
+			status:     ateapipb.Actor_STATUS_RUNNING,
+			wantStatus: ateapipb.Actor_STATUS_CRASHED,
+		},
+		{
+			name:       "suspending actor",
+			status:     ateapipb.Actor_STATUS_SUSPENDING,
+			wantStatus: ateapipb.Actor_STATUS_SUSPENDING,
+		},
+		{
+			name:       "resuming actor",
+			status:     ateapipb.Actor_STATUS_RESUMING,
+			wantStatus: ateapipb.Actor_STATUS_RESUMING,
+		},
+	}
+	for i, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			persistence, cleanup := storetest.SetupTestStore(t)
+			defer cleanup()
+
+			ns, pool, pod := fmt.Sprintf("ns-release-crash-%d", i), "pool1", "worker-release-crash"
+			actorName := "actor-release-crash"
+			if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
+				WorkerNamespace: ns,
+				WorkerPool:      pool,
+				WorkerPod:       pod,
+				WorkerPodUid:    testPodUID,
+				Ip:              "10.0.0.3",
+			}); err != nil {
+				t.Fatalf("create worker: %v", err)
+			}
+			if _, err := persistence.CreateActor(ctx, &ateapipb.Actor{
+				Metadata: &ateapipb.ResourceMetadata{Name: actorName, Atespace: "team-release"}, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
+				Status:            tc.status,
+				AteomPodNamespace: ns, AteomPodName: pod, AteomPodIp: "10.0.0.3", AteomPodUid: testPodUID,
+			}); err != nil {
+				t.Fatalf("create actor: %v", err)
+			}
+			w, _ := persistence.GetWorker(ctx, ns, pool, pod)
+			w.Assignment = &ateapipb.Assignment{
+				Actor: &ateapipb.ObjectRef{Name: actorName, Atespace: "team-release"},
+			}
+			if err := persistence.UpdateWorker(ctx, w, w.Version); err != nil {
+				t.Fatalf("update worker: %v", err)
+			}
+
+			syncer := &WorkerPoolSyncer{persistence: persistence}
+			if err := syncer.markActorCrashed(ctx, ns, pool, pod); err != nil {
+				t.Fatalf("markActorCrashed: %v", err)
+			}
+
+			got, err := persistence.GetActor(ctx, "team-release", actorName)
+			if err != nil {
+				t.Fatalf("get actor: %v", err)
+			}
+			if got.GetStatus() != tc.wantStatus {
+				t.Errorf("actor status = %v, want %v", got.GetStatus(), tc.wantStatus)
+			}
+			if tc.wantStatus == ateapipb.Actor_STATUS_CRASHED {
+				if got.AteomPodName != "" || got.AteomPodNamespace != "" || got.AteomPodIp != "" {
+					t.Errorf("bind fields not cleared: %+v", got)
+				}
+			} else {
+				if got.AteomPodName != pod || got.AteomPodNamespace != ns || got.AteomPodUid != testPodUID {
+					t.Errorf("bind fields not preserved: %+v", got)
+				}
+			}
+		})
+	}
+}
+
+// failingActorStore wraps a store.Interface and fails GetActor, to exercise
+// handleTerminatedAteom's release-failure path.
+type failingActorStore struct {
+	store.Interface
+}
+
+func (f *failingActorStore) GetActor(ctx context.Context, atespace, name string) (*ateapipb.Actor, error) {
+	return nil, errors.New("injected store failure")
+}
+
+// TestHandleTerminatedAteom_ReleaseFailureKeepsPod verifies that when the
+// actor release fails, handleTerminatedAteom returns early without deleting
+// the worker row or the pod: the pod's termination record is the retry
+// trigger, and the informer resync replays it.
+func TestHandleTerminatedAteom_ReleaseFailureKeepsPod(t *testing.T) {
+	ctx := context.Background()
+	persistence, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+
+	ns, pool, podName := "ns-release-fail", "pool1", "worker-release-fail"
+	if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
+		WorkerNamespace: ns,
+		WorkerPool:      pool,
+		WorkerPod:       podName,
+		Ip:              "10.0.0.4",
+		Assignment: &ateapipb.Assignment{
+			Actor: &ateapipb.ObjectRef{Name: "actor-release-fail", Atespace: "team-release-fail"},
+		},
+	}); err != nil {
+		t.Fatalf("create worker: %v", err)
+	}
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      podName,
+			Namespace: ns,
+			UID:       testPodUID,
+			Labels:    map[string]string{workerPodLabel: pool},
+		},
+	}
+	fakeK8s := fake.NewSimpleClientset(pod)
+
+	syncer := &WorkerPoolSyncer{persistence: &failingActorStore{persistence}, kubeClient: fakeK8s}
+	syncer.handleTerminatedAteom(ctx, pod)
+
+	if _, err := persistence.GetWorker(ctx, ns, pool, podName); err != nil {
+		t.Errorf("worker row must survive a failed release: %v", err)
+	}
+	if _, err := fakeK8s.CoreV1().Pods(ns).Get(ctx, podName, metav1.GetOptions{}); err != nil {
+		t.Errorf("pod must survive a failed release: %v", err)
+	}
+}
+
+// failingDeleteWorkerStore wraps a store.Interface and fails DeleteWorker, to
+// exercise handleTerminatedAteom's idle-worker failure path.
+type failingDeleteWorkerStore struct {
+	store.Interface
+}
+
+func (f *failingDeleteWorkerStore) DeleteWorker(ctx context.Context, namespace, pool, pod string) error {
+	return errors.New("injected store failure")
+}
+
+// TestHandleTerminatedAteom_IdleWorkerRowDeletedBeforePod verifies that for a
+// worker with no assignment, the store row is deleted before the pod: the row
+// is what findFreeWorker assigns from, so it must leave the assignable pool
+// before the recycle starts, not when the pod deletion propagates back
+// through the informer.
+func TestHandleTerminatedAteom_IdleWorkerRowDeletedBeforePod(t *testing.T) {
+	ctx := context.Background()
+	persistence, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+
+	ns, pool, podName := "ns-idle-recycle", "pool1", "worker-idle-recycle"
+	if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
+		WorkerNamespace: ns,
+		WorkerPool:      pool,
+		WorkerPod:       podName,
+		WorkerPodUid:    testPodUID,
+		Ip:              "10.0.0.5",
+	}); err != nil {
+		t.Fatalf("create worker: %v", err)
+	}
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      podName,
+			Namespace: ns,
+			UID:       testPodUID,
+			Labels:    map[string]string{workerPodLabel: pool},
+		},
+	}
+	fakeK8s := fake.NewSimpleClientset(pod)
+	rowGoneAtPodDelete := false
+	fakeK8s.PrependReactor("delete", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		_, gerr := persistence.GetWorker(ctx, ns, pool, podName)
+		rowGoneAtPodDelete = errors.Is(gerr, store.ErrNotFound)
+		return false, nil, nil
+	})
+
+	syncer := &WorkerPoolSyncer{persistence: persistence, kubeClient: fakeK8s}
+	syncer.handleTerminatedAteom(ctx, pod)
+
+	if _, err := persistence.GetWorker(ctx, ns, pool, podName); !errors.Is(err, store.ErrNotFound) {
+		t.Errorf("idle worker row must be deleted, got err=%v", err)
+	}
+	if _, err := fakeK8s.CoreV1().Pods(ns).Get(ctx, podName, metav1.GetOptions{}); !apierrors.IsNotFound(err) {
+		t.Errorf("pod must be deleted, got err=%v", err)
+	}
+	if !rowGoneAtPodDelete {
+		t.Errorf("worker row must leave the store before the pod is deleted")
+	}
+}
+
+// TestHandleTerminatedAteom_IdleDeleteWorkerFailureKeepsPod verifies that
+// when deleting the idle worker's row fails, the pod survives so the resync
+// retries the recycle.
+func TestHandleTerminatedAteom_IdleDeleteWorkerFailureKeepsPod(t *testing.T) {
+	ctx := context.Background()
+	persistence, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+
+	ns, pool, podName := "ns-idle-fail", "pool1", "worker-idle-fail"
+	if err := persistence.CreateWorker(ctx, &ateapipb.Worker{
+		WorkerNamespace: ns,
+		WorkerPool:      pool,
+		WorkerPod:       podName,
+		WorkerPodUid:    testPodUID,
+		Ip:              "10.0.0.6",
+	}); err != nil {
+		t.Fatalf("create worker: %v", err)
+	}
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      podName,
+			Namespace: ns,
+			UID:       testPodUID,
+			Labels:    map[string]string{workerPodLabel: pool},
+		},
+	}
+	fakeK8s := fake.NewSimpleClientset(pod)
+
+	syncer := &WorkerPoolSyncer{persistence: &failingDeleteWorkerStore{persistence}, kubeClient: fakeK8s}
+	syncer.handleTerminatedAteom(ctx, pod)
+
+	if _, err := fakeK8s.CoreV1().Pods(ns).Get(ctx, podName, metav1.GetOptions{}); err != nil {
+		t.Errorf("pod must survive a failed row delete: %v", err)
+	}
+	if _, err := persistence.GetWorker(ctx, ns, pool, podName); err != nil {
+		t.Errorf("worker row must survive in the underlying store: %v", err)
+	}
+}
+
 func TestSyncer_OmittedFields(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -315,7 +739,7 @@ func TestSyncer_OmittedFields(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      podName,
 			Namespace: ns,
-			UID:       "08675309-4a65-6e6e-7973-6e756d626572",
+			UID:       testPodUID,
 			Labels: map[string]string{
 				workerPodLabel: poolName,
 			},
diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go
index c08feb51c..424eac8cf 100644
--- a/cmd/ateapi/main.go
+++ b/cmd/ateapi/main.go
@@ -138,7 +138,7 @@ func main() {
 	workerPodInformerFactory, workerPodInformer := controlapi.WorkerPodInformer(clientset)
 	ateletPodInformerFactory, ateletPodInformer := controlapi.AteletInformer(clientset)
 
-	syncer := controlapi.NewWorkerPoolSyncer(redisPersistence, workerPodInformer, workerPoolLister)
+	syncer := controlapi.NewWorkerPoolSyncer(redisPersistence, clientset, workerPodInformer, workerPoolLister)
 	syncer.Start(ctx)
 
 	stopCh := make(chan struct{})
diff --git a/demos/counter/counter.go b/demos/counter/counter.go
index bd5b90848..0daa2d5bb 100644
--- a/demos/counter/counter.go
+++ b/demos/counter/counter.go
@@ -93,6 +93,18 @@ func main() {
 		w.Write([]byte("ok\n"))
 	})
 
+	// /crash makes the actor exit with status 1 on demand, to exercise
+	// actor crash handling while the actor is actively serving.
+	defaultMux.HandleFunc("/crash", func(w http.ResponseWriter, r *http.Request) {
+		slog.ErrorContext(r.Context(), "Crashing deliberately (/crash requested)")
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("crashing\n"))
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+		os.Exit(1)
+	})
+
 	go func() {
 		slog.InfoContext(ctx, "Starting counter server on port 80")
 		if err := http.ListenAndServe(":80", defaultMux); err != nil {
diff --git a/manifests/ate-install/ate-api-server.yaml b/manifests/ate-install/ate-api-server.yaml
index f86294953..161be5494 100644
--- a/manifests/ate-install/ate-api-server.yaml
+++ b/manifests/ate-install/ate-api-server.yaml
@@ -13,7 +13,7 @@
 # limitations under the License.
 
 
-# Define Permissions (Read-Only for Pods)
+# Define Permissions.
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
@@ -21,7 +21,9 @@ metadata:
 rules:
 - apiGroups: [""]
   resources: ["pods"]
-  verbs: ["get", "watch", "list"]
+  # delete is needed by the worker pool syncer
+  # to recycle worker pods whose ateom container terminated.
+  verbs: ["get", "watch", "list", "delete"]
 - apiGroups: ["ate.dev"]
   resources: ["actortemplates", "workerpools", "sandboxconfigs"]
   verbs: ["get", "watch", "list"]