diff --git a/backend/internal/service/notification_delivery_lifecycle_test.go b/backend/internal/service/notification_delivery_lifecycle_test.go
index 861c52c3..c53c8cd0 100644
--- a/backend/internal/service/notification_delivery_lifecycle_test.go
+++ b/backend/internal/service/notification_delivery_lifecycle_test.go
@@ -2,13 +2,21 @@ package service
 
 import (
 	"context"
+	"fmt"
+	"io"
+	"net"
+	"os/exec"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill-redisstream/pkg/redisstream"
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/gochat/gochat/internal/lifecycle"
+	"github.com/gochat/gochat/internal/pubsub"
+	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/require"
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
@@ -21,6 +29,35 @@ type notificationLifecycleSubscriber struct {
 	ctx      context.Context
 }
 
+type notificationAckGate struct {
+	topic     string
+	started   chan struct{}
+	release   chan struct{}
+	done      chan struct{}
+	group     string
+	startOnce sync.Once
+	doneOnce  sync.Once
+}
+
+func (g *notificationAckGate) DialHook(next redis.DialHook) redis.DialHook { return next }
+func (g *notificationAckGate) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
+	return next
+}
+func (g *notificationAckGate) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
+	return func(ctx context.Context, cmd redis.Cmder) error {
+		args := cmd.Args()
+		if cmd.Name() != "xack" || len(args) < 3 || fmt.Sprint(args[1]) != g.topic {
+			return next(ctx, cmd)
+		}
+		g.group = fmt.Sprint(args[2])
+		g.startOnce.Do(func() { close(g.started) })
+		<-g.release
+		err := next(ctx, cmd)
+		g.doneOnce.Do(func() { close(g.done) })
+		return err
+	}
+}
+
 func (s *notificationLifecycleSubscriber) Subscribe(ctx context.Context, _ string) (<-chan *message.Message, error) {
 	s.ctx = ctx
 	s.once.Do(func() { close(s.ready) })
@@ -91,3 +128,109 @@ func TestNotificationCloseTimeoutWaitsForHandlerBeforeDependenciesClose(t *testi
 	require.Error(t, sqlDB.Ping())
 	require.NoError(t, <-runDone)
 }
+
+func TestNotificationShutdownWaitsForRedisStreamXAck(t *testing.T) {
+	addr := startNotificationRedis(t)
+	topic := pubsub.TopicSystemNotification
+	gate := &notificationAckGate{
+		topic:   topic,
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+		done:    make(chan struct{}),
+	}
+	subscriberClient := redis.NewClient(&redis.Options{Addr: addr, PoolSize: 32})
+	subscriberClient.AddHook(gate)
+	publisherClient := redis.NewClient(&redis.Options{Addr: addr})
+	t.Cleanup(func() {
+		_ = subscriberClient.Close()
+		_ = publisherClient.Close()
+	})
+
+	service, err := NewNotificationDeliveryService(nil, nil, nil, nil, nil, nil, nil, subscriberClient)
+	require.NoError(t, err)
+	handlers := lifecycle.NewHandlerGroup()
+	service.SetHandlerGroup(handlers)
+
+	releaseAck := sync.OnceFunc(func() { close(gate.release) })
+	t.Cleanup(func() {
+		releaseAck()
+		_ = service.Close()
+	})
+	runDone := make(chan error, 1)
+	go func() { runDone <- service.Start(handlers.Context()) }()
+	<-service.router.Running()
+
+	publisher, err := redisstream.NewPublisher(redisstream.PublisherConfig{Client: publisherClient}, watermill.NopLogger{})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = publisher.Close() })
+	require.NoError(t, publisher.Publish(topic, message.NewMessage(watermill.NewUUID(), []byte("{"))))
+	select {
+	case <-gate.started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for XAck")
+	}
+
+	handlers.Wait()
+	pending, err := publisherClient.XPending(context.Background(), topic, gate.group).Result()
+	require.NoError(t, err)
+	require.EqualValues(t, 1, pending.Count)
+
+	handlers.Stop()
+	closeDone := make(chan error, 1)
+	go func() { closeDone <- service.Close() }()
+	select {
+	case err := <-closeDone:
+		t.Fatalf("router closed before XAck completed: %v", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+	require.NoError(t, publisherClient.Ping(context.Background()).Err())
+
+	releaseAck()
+	require.NoError(t, <-closeDone)
+	require.NoError(t, <-runDone)
+	select {
+	case <-gate.done:
+	default:
+		t.Fatal("router closed without completing XAck")
+	}
+	pending, err = publisherClient.XPending(context.Background(), topic, gate.group).Result()
+	require.NoError(t, err)
+	require.Zero(t, pending.Count)
+	require.Error(t, subscriberClient.Ping(context.Background()).Err())
+}
+
+func startNotificationRedis(t *testing.T) string {
+	t.Helper()
+	redisServer, err := exec.LookPath("redis-server")
+	if err != nil {
+		t.Skip("redis-server is required for this integration test")
+	}
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	port := listener.Addr().(*net.TCPAddr).Port
+	require.NoError(t, listener.Close())
+
+	cmd := exec.Command(redisServer,
+		"--bind", "127.0.0.1",
+		"--protected-mode", "no",
+		"--port", strconv.Itoa(port),
+		"--save", "",
+		"--appendonly", "no",
+		"--dir", t.TempDir(),
+	)
+	cmd.Stdout = io.Discard
+	cmd.Stderr = io.Discard
+	require.NoError(t, cmd.Start())
+	t.Cleanup(func() {
+		_ = cmd.Process.Kill()
+		_ = cmd.Wait()
+	})
+
+	addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port))
+	client := redis.NewClient(&redis.Options{Addr: addr})
+	t.Cleanup(func() { _ = client.Close() })
+	require.Eventually(t, func() bool {
+		return client.Ping(context.Background()).Err() == nil
+	}, 5*time.Second, 10*time.Millisecond)
+	return addr
+}
diff --git a/backend/internal/service/notification_delivery_service.go b/backend/internal/service/notification_delivery_service.go
index d1611cba..49d3a20a 100644
--- a/backend/internal/service/notification_delivery_service.go
+++ b/backend/internal/service/notification_delivery_service.go
@@ -46,6 +46,19 @@ type NotificationDeliveryService struct {
 	handlers            *lifecycle.HandlerGroup
 }
 
+type ackCompletingRedisClient struct{ redis.UniversalClient }
+
+func (c ackCompletingRedisClient) XAck(ctx context.Context, stream, group string, ids ...string) *redis.IntCmd {
+	// Subscriber.Close cancels ctx while waiting; the in-flight XAck must finish first.
+	ackCtx := context.WithoutCancel(ctx)
+	if deadline, ok := ctx.Deadline(); ok {
+		var cancel context.CancelFunc
+		ackCtx, cancel = context.WithDeadline(ackCtx, deadline)
+		defer cancel()
+	}
+	return c.UniversalClient.XAck(ackCtx, stream, group, ids...)
+}
+
 // NewNotificationDeliveryService creates a delivery service and registers Watermill handlers.
 func NewNotificationDeliveryService(
 	notificationService *NotificationService,
@@ -58,10 +71,13 @@ func NewNotificationDeliveryService(
 	redisClient redis.UniversalClient,
 ) (*NotificationDeliveryService, error) {
 	loggerAdapter := &deliveryWatermillAdapter{}
+	if redisClient == nil {
+		return nil, fmt.Errorf("failed to create delivery subscriber: redis client is empty")
+	}
 
 	subscriber, err := redisstream.NewSubscriber(
 		redisstream.SubscriberConfig{
-			Client:        redisClient,
+			Client:        ackCompletingRedisClient{redisClient},
 			ConsumerGroup: "notif-delivery-" + watermill.NewUUID(),
 		},
 		loggerAdapter,