diff --git a/docs-website/router/metrics-and-monitoring.mdx b/docs-website/router/metrics-and-monitoring.mdx
index 43f05d58f..71cdbad31 100644
--- a/docs-website/router/metrics-and-monitoring.mdx
+++ b/docs-website/router/metrics-and-monitoring.mdx
@@ -247,6 +247,14 @@ telemetry:
 
 * `router.engine.messages.sent`: The number of total messages for subscriptions sent over from the subgraph to the router.
 
+* `router.subscription.delivery.attempts`: Downstream subscription frame delivery attempts, tagged with `wg.subscription.transport`, `wg.subscription.frame_type`, and, for WebSockets, `wg.websocket.subprotocol`.
+
+* `router.subscription.delivery.write.failures`: Downstream writes that the router knows failed. The bounded `wg.subscription.failure_stage` and `wg.subscription.failure_reason` dimensions distinguish deadline, serialization, write, and flush failures without attaching client or event identifiers to metrics.
+
+* `router.subscription.disconnects`: Closed SSE requests and WebSocket connections, tagged with the transport, disconnect initiator, and disconnect reason. A WebSocket connection is counted once even when it carries multiple subscriptions.
+
+Failed event writes also produce a structured `Subscription event delivery failed` log containing request, connection, subscription, and operation identifiers; a router-local delivery sequence; the write duration; and the configured timeout. Payloads are represented by a SHA-256 hash and byte count and are not logged. The delivery sequence is scoped to one subscription and is intended to distinguish its attempted events; it is not a broker offset. A successful transport write means the router handed the frame to the connection; SSE and WebSocket do not provide application-level client acknowledgements.
+
 
 ### Resolver Metrics
 
diff --git a/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx b/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx
index ab2ac8a46..d94998a23 100644
--- a/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx
+++ b/docs-website/router/metrics-and-monitoring/prometheus-metric-reference.mdx
@@ -122,6 +122,12 @@ telemetry:
 
 * [`router_engine_messages_sent_total`](#router-engine-messages-sent-total): The number of total messages for subscriptions sent over from the subgraph to the router.
 
+* `router_subscription_delivery_attempts_total`: The number of downstream SSE and WebSocket subscription frame delivery attempts.
+
+* `router_subscription_delivery_write_failures_total`: The number of downstream subscription frame writes known to have failed.
+
+* `router_subscription_disconnects_total`: The number of downstream SSE requests and WebSocket connections that closed, grouped by bounded initiator and reason dimensions.
+
 ### Resolver Metrics
 
 These metrics expose usage of the GraphQL engine's resolver concurrency pool. Use them to detect when operations queue because the pool is saturated.
diff --git a/router-tests/events/kafka_sse_write_timeout_test.go b/router-tests/events/kafka_sse_write_timeout_test.go
new file mode 100644
index 000000000..a7c602a29
--- /dev/null
+++ b/router-tests/events/kafka_sse_write_timeout_test.go
@@ -0,0 +1,248 @@
+package events_test
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"net/http"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/wundergraph/cosmo/router-tests/events"
+	"github.com/wundergraph/cosmo/router-tests/testenv"
+	"github.com/wundergraph/cosmo/router/core"
+	nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1"
+	"github.com/wundergraph/cosmo/router/pkg/config"
+)
+
+const blockSSEWriteHeader = "X-Test-Block-SSE-Write"
+
+var (
+	_ core.Module                 = (*blockingSSEWriterModule)(nil)
+	_ core.RouterOnRequestHandler = (*blockingSSEWriterModule)(nil)
+)
+
+// blockingSSEWriterModule simulates a client that stops draining its SSE
+// connection without closing it. The wrapped writer only returns when the
+// router sets a write deadline or the test releases it during cleanup.
+type blockingSSEWriterModule struct {
+	armed        *atomic.Bool
+	writeStarted chan struct{}
+	startedOnce  *sync.Once
+	release      chan struct{}
+}
+
+func (m *blockingSSEWriterModule) Module() core.ModuleInfo {
+	return core.ModuleInfo{
+		ID:       "blockingSSEWriterModule",
+		Priority: 1,
+		New: func() core.Module {
+			return &blockingSSEWriterModule{
+				armed:        m.armed,
+				writeStarted: m.writeStarted,
+				startedOnce:  m.startedOnce,
+				release:      m.release,
+			}
+		},
+	}
+}
+
+func (m *blockingSSEWriterModule) RouterOnRequest(ctx core.RequestContext, next http.Handler) {
+	if ctx.Request().Header.Get(blockSSEWriteHeader) != "true" {
+		next.ServeHTTP(ctx.ResponseWriter(), ctx.Request())
+		return
+	}
+
+	next.ServeHTTP(&deadlineBlockingResponseWriter{
+		ResponseWriter: ctx.ResponseWriter(),
+		armed:          m.armed,
+		writeStarted:   m.writeStarted,
+		startedOnce:    m.startedOnce,
+		release:        m.release,
+	}, ctx.Request())
+}
+
+type deadlineBlockingResponseWriter struct {
+	http.ResponseWriter
+	armed         *atomic.Bool
+	writeStarted  chan struct{}
+	startedOnce   *sync.Once
+	release       chan struct{}
+	deadlineNanos atomic.Int64
+}
+
+func (w *deadlineBlockingResponseWriter) Write(data []byte) (int, error) {
+	if !w.armed.CompareAndSwap(true, false) {
+		return w.ResponseWriter.Write(data)
+	}
+
+	w.startedOnce.Do(func() { close(w.writeStarted) })
+	deadlineNanos := w.deadlineNanos.Load()
+	if deadlineNanos == 0 {
+		<-w.release
+		return 0, os.ErrDeadlineExceeded
+	}
+
+	wait := time.Until(time.Unix(0, deadlineNanos))
+	if wait <= 0 {
+		return 0, os.ErrDeadlineExceeded
+	}
+	timer := time.NewTimer(wait)
+	defer timer.Stop()
+	select {
+	case <-w.release:
+		return 0, os.ErrDeadlineExceeded
+	case <-timer.C:
+		return 0, os.ErrDeadlineExceeded
+	}
+}
+
+func (w *deadlineBlockingResponseWriter) Flush() {
+	if flusher, ok := w.ResponseWriter.(http.Flusher); ok {
+		flusher.Flush()
+	}
+}
+
+func (w *deadlineBlockingResponseWriter) FlushError() error {
+	if flusher, ok := w.ResponseWriter.(interface{ FlushError() error }); ok {
+		return flusher.FlushError()
+	}
+	w.Flush()
+	return nil
+}
+
+func (w *deadlineBlockingResponseWriter) SetWriteDeadline(deadline time.Time) error {
+	w.deadlineNanos.Store(deadline.UnixNano())
+	return nil
+}
+
+func (w *deadlineBlockingResponseWriter) Unwrap() http.ResponseWriter {
+	return w.ResponseWriter
+}
+
+func TestKafkaSubscriptionRecoversAfterSSEWriteTimeout(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping Kafka integration test in short mode")
+	}
+
+	const topic = "employeeUpdated-sse-write-timeout"
+	armed := &atomic.Bool{}
+	writeStarted := make(chan struct{})
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) })
+
+	module := &blockingSSEWriterModule{
+		armed:        armed,
+		writeStarted: writeStarted,
+		startedOnce:  &sync.Once{},
+		release:      release,
+	}
+
+	testenv.Run(t, &testenv.Config{
+		RouterConfigJSONTemplate: testenv.ConfigWithEdfsKafkaJSONTemplate,
+		EnableKafka:              true,
+		RouterOptions:            []core.Option{core.WithCustomModules(module)},
+		ModifyRouterConfig: func(routerConfig *nodev1.RouterConfig) {
+			overrideKafkaTopicsForField(t, routerConfig, "employeeUpdatedMyKafka",
+				[]string{"employeeUpdated", "employeeUpdatedTwo"}, topic)
+		},
+		ModifyEngineExecutionConfiguration: func(cfg *config.EngineExecutionConfiguration) {
+			cfg.SSEServerWriteTimeout = 100 * time.Millisecond
+		},
+	}, func(t *testing.T, xEnv *testenv.Environment) {
+		events.KafkaEnsureTopicExists(t, xEnv, EventWaitTimeout, topic)
+
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+		client := &http.Client{}
+		blockedResp := openSSESubscription(t, ctx, client, xEnv.GraphQLRequestURL(), true)
+		defer blockedResp.Body.Close()
+		healthyResp := openSSESubscription(t, ctx, client, xEnv.GraphQLRequestURL(), false)
+		defer healthyResp.Body.Close()
+		healthyReader := bufio.NewReader(healthyResp.Body)
+
+		xEnv.WaitForSubscriptionCount(2, EventWaitTimeout)
+		xEnv.WaitForTriggerCount(1, EventWaitTimeout)
+
+		armed.Store(true)
+		xEnv.KafkaPublishUntilReceived(topic,
+			`{"__typename":"Employee","id":1,"update":{"name":"blocked"}}`, 1, EventWaitTimeout)
+
+		select {
+		case <-writeStarted:
+		case <-time.After(EventWaitTimeout):
+			t.Fatal("timed out waiting for the SSE write to block")
+		}
+
+		require.Contains(t, readSSEData(t, healthyReader), `"id":1`)
+
+		xEnv.WaitForSubscriptionCount(1, EventWaitTimeout)
+		xEnv.KafkaPublishUntilReceived(topic,
+			`{"__typename":"Employee","id":2,"update":{"name":"recovery"}}`, 1, EventWaitTimeout)
+
+		recovery := make(chan string, 1)
+		go func() {
+			data, err := readSSEDataLine(healthyReader)
+			if err != nil {
+				recovery <- "error: " + err.Error()
+				return
+			}
+			recovery <- data
+		}()
+
+		select {
+		case data := <-recovery:
+			require.Contains(t, data, `"id":2`)
+		case <-time.After(EventWaitTimeout):
+			t.Fatal("healthy subscription did not receive the queued event after the SSE write deadline")
+		}
+	})
+}
+
+func openSSESubscription(t *testing.T, ctx context.Context, client *http.Client, url string, blocked bool) *http.Response {
+	t.Helper()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url,
+		strings.NewReader(`{"query":"subscription { employeeUpdatedMyKafka(employeeID: 3) { id } }"}`))
+	require.NoError(t, err)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "text/event-stream")
+	if blocked {
+		req.Header.Set(blockSSEWriteHeader, "true")
+	}
+
+	resp, err := client.Do(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	require.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
+	return resp
+}
+
+func readSSEData(t *testing.T, reader *bufio.Reader) string {
+	t.Helper()
+	data, err := readSSEDataLine(reader)
+	require.NoError(t, err)
+	return data
+}
+
+func readSSEDataLine(reader *bufio.Reader) (string, error) {
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			return "", err
+		}
+		line = strings.TrimSpace(line)
+		if strings.HasPrefix(line, "data: ") {
+			return strings.TrimPrefix(line, "data: "), nil
+		}
+		if strings.HasPrefix(line, "event: complete") {
+			return "", errors.New("subscription completed before receiving data")
+		}
+	}
+}
diff --git a/router/core/graph_server.go b/router/core/graph_server.go
index 3cc45d5f9..e4876a514 100644
--- a/router/core/graph_server.go
+++ b/router/core/graph_server.go
@@ -1806,6 +1806,7 @@ func (s *graphServer) buildGraphMux(
 		SubgraphErrorPropagation:        s.subgraphErrorPropagation,
 		EngineLoaderHooks:               loaderHooks,
 		HeaderPropagation:               s.headerPropagation,
+		SSEServerWriteTimeout:           s.engineExecutionConfiguration.SSEServerWriteTimeout,
 	}
 
 	if s.redisClient != nil {
diff --git a/router/core/graphql_handler.go b/router/core/graphql_handler.go
index 4ef92da46..e8322a21d 100644
--- a/router/core/graphql_handler.go
+++ b/router/core/graphql_handler.go
@@ -10,7 +10,9 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
+	"github.com/go-chi/chi/v5/middleware"
 	otelmetric "go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
@@ -87,6 +89,7 @@ type HandlerOptions struct {
 	EnableCostResponseHeaders       bool
 
 	ApolloSubscriptionMultipartPrintBoundary bool
+	SSEServerWriteTimeout                    time.Duration
 	HeaderPropagation                        *HeaderPropagation
 }
 
@@ -109,6 +112,7 @@ func NewGraphQLHandler(opts HandlerOptions) *GraphQLHandler {
 		subgraphErrorPropagation:                 opts.SubgraphErrorPropagation,
 		engineLoaderHooks:                        opts.EngineLoaderHooks,
 		apolloSubscriptionMultipartPrintBoundary: opts.ApolloSubscriptionMultipartPrintBoundary,
+		sseServerWriteTimeout:                    opts.SSEServerWriteTimeout,
 		headerPropagation:                        opts.HeaderPropagation,
 	}
 	return graphQLHandler
@@ -143,6 +147,7 @@ type GraphQLHandler struct {
 	enableCostResponseHeaders       bool
 
 	apolloSubscriptionMultipartPrintBoundary bool
+	sseServerWriteTimeout                    time.Duration
 }
 
 func (h *GraphQLHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -284,26 +289,40 @@ func (h *GraphQLHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 	case *plan.SubscriptionResponsePlan:
 		var (
-			writer resolve.SubscriptionResponseWriter
-			ok     bool
+			writer    resolve.SubscriptionResponseWriter
+			writerErr error
 		)
 		h.setDebugCacheHeaders(w, reqCtx.operation)
 
 		defer propagateSubgraphErrors(resolveCtx)
-		resolveCtx, writer, ok = GetSubscriptionResponseWriter(resolveCtx, r, w, h.apolloSubscriptionMultipartPrintBoundary)
-		if !ok {
-			reqCtx.logger.Error("unable to get subscription response writer", zap.Error(errCouldNotFlushResponse))
-			trackFinalResponseError(r.Context(), errCouldNotFlushResponse)
+		resolveCtx, writer, writerErr = GetSubscriptionResponseWriter(resolveCtx, r, w, SubscriptionResponseWriterOptions{
+			ApolloSubscriptionMultipartPrintBoundary: h.apolloSubscriptionMultipartPrintBoundary,
+			SSEWriteTimeout:                          h.sseServerWriteTimeout,
+			Logger:                                   reqCtx.logger,
+			Stats:                                    h.engineStats,
+			Telemetry: subscriptionTelemetryContext{
+				transport:     subscriptionTransportSSE,
+				requestID:     middleware.GetReqID(r.Context()),
+				operationName: reqCtx.operation.name,
+				writeTimeout:  h.sseServerWriteTimeout,
+			},
+		})
+		if writerErr != nil {
+			reqCtx.logger.Error("unable to get subscription response writer", zap.Error(writerErr))
+			trackFinalResponseError(r.Context(), writerErr)
 			writeRequestErrors(writeRequestErrorsParams{
 				request:           r,
 				writer:            w,
 				statusCode:        http.StatusInternalServerError,
-				requestErrors:     graphqlerrors.RequestErrorsFromError(errCouldNotFlushResponse),
+				requestErrors:     graphqlerrors.RequestErrorsFromError(writerErr),
 				logger:            reqCtx.logger,
 				headerPropagation: h.headerPropagation,
 			})
 			return
 		}
+		if lifecycle, ok := writer.(*HttpFlushWriter); ok {
+			defer lifecycle.subscriptionRequestEnded()
+		}
 
 		if !resolveCtx.ExecutionOptions.SkipLoader {
 			h.engineStats.ConnectionsInc()
diff --git a/router/core/subscription_delivery_observability.go b/router/core/subscription_delivery_observability.go
new file mode 100644
index 000000000..3bd41ab8a
--- /dev/null
+++ b/router/core/subscription_delivery_observability.go
@@ -0,0 +1,243 @@
+package core
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"net"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"github.com/wundergraph/cosmo/router/pkg/statistics"
+	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve"
+	"go.uber.org/zap"
+)
+
+const (
+	subscriptionTransportSSE       = "sse"
+	subscriptionTransportWebSocket = "websocket"
+)
+
+type subscriptionTelemetryContext struct {
+	transport     string
+	subprotocol   string
+	requestID     string
+	operationName string
+	connectionID  resolve.ConnectionID
+	writeTimeout  time.Duration
+}
+
+type subscriptionWriteError struct {
+	stage string
+	err   error
+}
+
+type subscriptionDisconnectTracker struct {
+	once      sync.Once
+	stats     statistics.EngineStatistics
+	logger    *zap.Logger
+	telemetry subscriptionTelemetryContext
+}
+
+type subscriptionDeliveryTracker struct {
+	sequence      atomic.Uint64
+	failureLogged atomic.Bool
+	stats         statistics.EngineStatistics
+	logger        *zap.Logger
+	subscription  string
+}
+
+func newSubscriptionDisconnectTracker(stats statistics.EngineStatistics, logger *zap.Logger, telemetry subscriptionTelemetryContext) *subscriptionDisconnectTracker {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	return &subscriptionDisconnectTracker{stats: stats, logger: logger, telemetry: telemetry}
+}
+
+func newSubscriptionDeliveryTracker(stats statistics.EngineStatistics, logger *zap.Logger, subscription string) *subscriptionDeliveryTracker {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	return &subscriptionDeliveryTracker{stats: stats, logger: logger, subscription: subscription}
+}
+
+func (t *subscriptionDisconnectTracker) disconnect(initiator, reason string, err error) {
+	if t == nil {
+		return
+	}
+	t.once.Do(func() {
+		observeSubscription(t.stats, statistics.SubscriptionObservation{
+			Kind:             statistics.SubscriptionObservationDisconnect,
+			Transport:        t.telemetry.transport,
+			Initiator:        initiator,
+			DisconnectReason: reason,
+			Subprotocol:      t.telemetry.subprotocol,
+		})
+		fields := []zap.Field{
+			zap.String("transport", t.telemetry.transport),
+			zap.String("websocket_subprotocol", t.telemetry.subprotocol),
+			zap.String("request_id", t.telemetry.requestID),
+			zap.Int64("connection_id", int64(t.telemetry.connectionID)),
+			zap.String("disconnect_initiator", initiator),
+			zap.String("disconnect_reason", reason),
+		}
+		if err != nil {
+			fields = append(fields, zap.Error(err))
+		}
+		if reason == "normal_completion" || reason == "client_closed" || reason == "context_canceled" {
+			t.logger.Debug("Subscription client disconnected", fields...)
+			return
+		}
+		t.logger.Info("Subscription client disconnected", fields...)
+	})
+}
+
+func disconnectReasonFromWriteError(err error) (initiator, reason string) {
+	_, failureReason := classifySubscriptionWriteFailure(err)
+	switch failureReason {
+	case "timeout":
+		return "router", "write_timeout"
+	case "client_disconnected":
+		return "client", "client_closed"
+	case "connection_closed":
+		return "router", "connection_closed"
+	case "context_canceled":
+		return "client", "context_canceled"
+	default:
+		return "network", "network_error"
+	}
+}
+
+func (e *subscriptionWriteError) Error() string { return e.err.Error() }
+func (e *subscriptionWriteError) Unwrap() error { return e.err }
+
+func wrapSubscriptionWriteError(stage string, err error) error {
+	if err == nil {
+		return nil
+	}
+	return &subscriptionWriteError{stage: stage, err: err}
+}
+
+func (t *subscriptionDeliveryTracker) observe(telemetry subscriptionTelemetryContext, payload []byte, duration time.Duration, err error) {
+	if t == nil {
+		return
+	}
+	deliverySequence := t.sequence.Add(1)
+	observeSubscription(t.stats, statistics.SubscriptionObservation{
+		Kind:        statistics.SubscriptionObservationDeliveryAttempt,
+		Transport:   telemetry.transport,
+		FrameType:   "next",
+		Subprotocol: telemetry.subprotocol,
+	})
+	if err == nil {
+		return
+	}
+
+	stage, reason := classifySubscriptionWriteFailure(err)
+	observeSubscription(t.stats, statistics.SubscriptionObservation{
+		Kind:          statistics.SubscriptionObservationDeliveryFailure,
+		Transport:     telemetry.transport,
+		FrameType:     "next",
+		FailureStage:  stage,
+		FailureReason: reason,
+		Subprotocol:   telemetry.subprotocol,
+	})
+	payloadHash := sha256.Sum256(payload)
+	fields := []zap.Field{
+		zap.String("transport", telemetry.transport),
+		zap.String("websocket_subprotocol", telemetry.subprotocol),
+		zap.String("request_id", telemetry.requestID),
+		zap.String("operation_name", telemetry.operationName),
+		zap.Int64("connection_id", int64(telemetry.connectionID)),
+		zap.String("subscription_id", t.subscription),
+		zap.Uint64("delivery_sequence", deliverySequence),
+		zap.String("payload_sha256", hex.EncodeToString(payloadHash[:])),
+		zap.Int("payload_bytes", len(payload)),
+		zap.String("frame_type", "next"),
+		zap.String("failure_stage", stage),
+		zap.String("failure_reason", reason),
+		zap.Int64("configured_write_timeout_ms", telemetry.writeTimeout.Milliseconds()),
+		zap.Float64("write_duration_ms", float64(duration)/float64(time.Millisecond)),
+		zap.Error(err),
+	}
+	if t.failureLogged.CompareAndSwap(false, true) {
+		t.logger.Warn("Subscription event delivery failed", fields...)
+		return
+	}
+	t.logger.Debug("Subscription event delivery failed", fields...)
+}
+
+func observeSubscriptionFrame(stats statistics.EngineStatistics, logger *zap.Logger, telemetry subscriptionTelemetryContext, frameType string, err error) {
+	observeSubscription(stats, statistics.SubscriptionObservation{
+		Kind:        statistics.SubscriptionObservationDeliveryAttempt,
+		Transport:   telemetry.transport,
+		FrameType:   frameType,
+		Subprotocol: telemetry.subprotocol,
+	})
+	if err == nil {
+		return
+	}
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	stage, reason := classifySubscriptionWriteFailure(err)
+	observeSubscription(stats, statistics.SubscriptionObservation{
+		Kind:          statistics.SubscriptionObservationDeliveryFailure,
+		Transport:     telemetry.transport,
+		FrameType:     frameType,
+		FailureStage:  stage,
+		FailureReason: reason,
+		Subprotocol:   telemetry.subprotocol,
+	})
+	logger.Warn("Subscription frame delivery failed",
+		zap.String("transport", telemetry.transport),
+		zap.String("websocket_subprotocol", telemetry.subprotocol),
+		zap.String("request_id", telemetry.requestID),
+		zap.String("operation_name", telemetry.operationName),
+		zap.Int64("connection_id", int64(telemetry.connectionID)),
+		zap.String("frame_type", frameType),
+		zap.String("failure_stage", stage),
+		zap.String("failure_reason", reason),
+		zap.Int64("configured_write_timeout_ms", telemetry.writeTimeout.Milliseconds()),
+		zap.Error(err),
+	)
+}
+
+func observeSubscription(stats statistics.EngineStatistics, observation statistics.SubscriptionObservation) {
+	observer, ok := stats.(statistics.SubscriptionObserver)
+	if !ok {
+		return
+	}
+	observer.ObserveSubscription(observation)
+}
+
+func classifySubscriptionWriteFailure(err error) (stage, reason string) {
+	stage = "write"
+	var writeErr *subscriptionWriteError
+	if errors.As(err, &writeErr) {
+		stage = writeErr.stage
+	}
+
+	var netErr net.Error
+	switch {
+	case errors.Is(err, context.DeadlineExceeded):
+		return stage, "timeout"
+	case errors.As(err, &netErr) && netErr.Timeout():
+		return stage, "timeout"
+	case errors.Is(err, context.Canceled):
+		return stage, "context_canceled"
+	case errors.Is(err, net.ErrClosed):
+		return stage, "connection_closed"
+	case errors.Is(err, syscall.EPIPE), errors.Is(err, syscall.ECONNRESET):
+		return stage, "client_disconnected"
+	case errors.Is(err, errors.ErrUnsupported):
+		return stage, "unsupported"
+	case stage == "serialize":
+		return stage, "serialization_error"
+	default:
+		return stage, "network_error"
+	}
+}
diff --git a/router/core/subscription_delivery_observability_test.go b/router/core/subscription_delivery_observability_test.go
new file mode 100644
index 000000000..c376d2f2a
--- /dev/null
+++ b/router/core/subscription_delivery_observability_test.go
@@ -0,0 +1,156 @@
+package core
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/gobwas/ws"
+	"github.com/stretchr/testify/require"
+	"github.com/wundergraph/cosmo/router/internal/wsproto"
+	"github.com/wundergraph/cosmo/router/pkg/statistics"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	zapobserver "go.uber.org/zap/zaptest/observer"
+)
+
+type failingSubscriptionProtocol struct {
+	writeErr error
+}
+
+func (p *failingSubscriptionProtocol) Subprotocol() string { return wsproto.GraphQLWSSubprotocol }
+func (p *failingSubscriptionProtocol) Initialize() (json.RawMessage, error) {
+	return nil, nil
+}
+func (p *failingSubscriptionProtocol) ReadMessage() (*wsproto.Message, error) { return nil, nil }
+func (p *failingSubscriptionProtocol) Pong(*wsproto.Message) error            { return nil }
+func (p *failingSubscriptionProtocol) WriteGraphQLData(string, json.RawMessage, json.RawMessage) error {
+	return p.writeErr
+}
+func (p *failingSubscriptionProtocol) WriteGraphQLErrors(string, json.RawMessage, json.RawMessage) error {
+	return p.writeErr
+}
+func (p *failingSubscriptionProtocol) Complete(string) error { return p.writeErr }
+
+func TestSubscriptionDeliveryTrackerRecordsFailureWithoutPayload(t *testing.T) {
+	logCore, logs := zapobserver.New(zapcore.DebugLevel)
+	stats := statistics.NewEngineStats(t.Context(), zap.NewNop(), false)
+	telemetry := subscriptionTelemetryContext{
+		transport:     subscriptionTransportSSE,
+		requestID:     "request-1",
+		operationName: "ProductUpdated",
+	}
+
+	payload := []byte(`{"data":{"productUpdated":{"id":"1"}}}`)
+	tracker := newSubscriptionDeliveryTracker(stats, zap.New(logCore), "subscription-7")
+	tracker.observe(telemetry, payload, 25*time.Millisecond, wrapSubscriptionWriteError("flush", context.DeadlineExceeded))
+
+	report := stats.GetReport()
+	require.Len(t, report.SubscriptionObservations, 2)
+	observations := make(map[statistics.SubscriptionObservationKind]statistics.SubscriptionObservationCount, 2)
+	for _, observation := range report.SubscriptionObservations {
+		observations[observation.Observation.Kind] = observation
+	}
+	require.Equal(t, uint64(1), observations[statistics.SubscriptionObservationDeliveryAttempt].Count)
+	require.Equal(t, uint64(1), observations[statistics.SubscriptionObservationDeliveryFailure].Count)
+	require.Equal(t, "flush", observations[statistics.SubscriptionObservationDeliveryFailure].Observation.FailureStage)
+	require.Equal(t, "timeout", observations[statistics.SubscriptionObservationDeliveryFailure].Observation.FailureReason)
+	require.Equal(t, 1, logs.Len())
+	fields := logs.All()[0].ContextMap()
+	payloadHash := sha256.Sum256(payload)
+	require.Equal(t, "subscription-7", fields["subscription_id"])
+	require.Equal(t, uint64(1), fields["delivery_sequence"])
+	require.Equal(t, hex.EncodeToString(payloadHash[:]), fields["payload_sha256"])
+	require.Equal(t, int64(len(payload)), fields["payload_bytes"])
+	require.Equal(t, 25.0, fields["write_duration_ms"])
+	require.Equal(t, "flush", fields["failure_stage"])
+	require.Equal(t, "timeout", fields["failure_reason"])
+	require.NotContains(t, fields, "payload")
+	require.NotContains(t, fields, "event_source_id")
+	require.NotContains(t, fields, "client_name")
+	require.NotContains(t, fields, "client_version")
+}
+
+func TestSubscriptionDisconnectTrackerRecordsOnce(t *testing.T) {
+	logCore, logs := zapobserver.New(zapcore.DebugLevel)
+	stats := statistics.NewEngineStats(t.Context(), zap.NewNop(), false)
+	tracker := newSubscriptionDisconnectTracker(stats, zap.New(logCore), subscriptionTelemetryContext{
+		transport:    subscriptionTransportWebSocket,
+		subprotocol:  wsproto.GraphQLWSSubprotocol,
+		connectionID: 41,
+	})
+
+	tracker.disconnect("client", "client_closed", nil)
+	tracker.disconnect("network", "network_error", errors.New("late error"))
+
+	require.Equal(t, 1, logs.Len())
+	report := stats.GetReport()
+	require.Len(t, report.SubscriptionObservations, 1)
+	require.Equal(t, statistics.SubscriptionObservationDisconnect, report.SubscriptionObservations[0].Observation.Kind)
+	require.Equal(t, "client_closed", report.SubscriptionObservations[0].Observation.DisconnectReason)
+}
+
+func TestWebsocketResponseWriterObservesFailedEventAtTransport(t *testing.T) {
+	logCore, logs := zapobserver.New(zapcore.DebugLevel)
+	stats := statistics.NewEngineStats(t.Context(), zap.NewNop(), false)
+	payload := []byte(`{"data":{"productUpdated":{"id":"1"}}}`)
+	rw := newWebsocketResponseWriter(
+		"subscription-1",
+		&failingSubscriptionProtocol{writeErr: context.DeadlineExceeded},
+		false,
+		zap.New(logCore),
+		stats,
+		nil,
+		subscriptionTelemetryContext{
+			transport:     subscriptionTransportWebSocket,
+			subprotocol:   wsproto.GraphQLWSSubprotocol,
+			requestID:     "request-1",
+			operationName: "ProductUpdated",
+		},
+	)
+
+	_, err := rw.Write(payload)
+	require.NoError(t, err)
+	require.ErrorIs(t, rw.Flush(), context.DeadlineExceeded)
+
+	report := stats.GetReport()
+	require.Len(t, report.SubscriptionObservations, 2)
+	require.Equal(t, 1, logs.Len())
+	fields := logs.All()[0].ContextMap()
+	require.Equal(t, "websocket", fields["transport"])
+	require.Equal(t, "subscription-1", fields["subscription_id"])
+	require.Equal(t, uint64(1), fields["delivery_sequence"])
+	require.Equal(t, "write", fields["failure_stage"])
+	require.Equal(t, "timeout", fields["failure_reason"])
+}
+
+func TestWebsocketDisconnectReasonUsesOriginalError(t *testing.T) {
+	initiator, reason := websocketDisconnectReason(context.DeadlineExceeded, wsproto.CloseKindNormal)
+	require.Equal(t, "network", initiator)
+	require.Equal(t, "timeout", reason)
+
+	initiator, reason = websocketDisconnectReason(errClientTerminatedConnection, wsproto.CloseKindNormal)
+	require.Equal(t, "client", initiator)
+	require.Equal(t, "client_closed", reason)
+
+	initiator, reason = websocketDisconnectReason(&wsproto.CloseError{
+		Kind: wsproto.CloseKind{Code: ws.StatusProtocolError, Reason: "bad frame"},
+	}, wsproto.CloseKindNormal)
+	require.Equal(t, "client", initiator)
+	require.Equal(t, "protocol_error", reason)
+}
+
+func TestHttpFlushWriterMarksContextFailuresAsDeliveryErrors(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	writer := &HttpFlushWriter{ctx: ctx}
+
+	_, err := writer.Write([]byte(`{"data":{}}`))
+	stage, reason := classifySubscriptionWriteFailure(err)
+	require.Equal(t, "buffer", stage)
+	require.Equal(t, "context_canceled", reason)
+}
diff --git a/router/core/subscription_response_writer.go b/router/core/subscription_response_writer.go
index abe951a38..2cfca556b 100644
--- a/router/core/subscription_response_writer.go
+++ b/router/core/subscription_response_writer.go
@@ -3,14 +3,19 @@ package core
 import (
 	"bytes"
 	"context"
+	"errors"
+	"fmt"
 	"io"
 	"mime"
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/wundergraph/astjson"
+	"github.com/wundergraph/cosmo/router/pkg/statistics"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve"
+	"go.uber.org/zap"
 )
 
 const (
@@ -31,16 +36,32 @@ type withFlushWriter interface {
 	SubscriptionResponseWriter() resolve.SubscriptionResponseWriter
 }
 
+type SubscriptionResponseWriterOptions struct {
+	ApolloSubscriptionMultipartPrintBoundary bool
+	SSEWriteTimeout                          time.Duration
+	Logger                                   *zap.Logger
+	Stats                                    statistics.EngineStatistics
+	Telemetry                                subscriptionTelemetryContext
+}
+
 type HttpFlushWriter struct {
-	ctx           context.Context
-	cancel        context.CancelFunc
-	writer        io.Writer
-	flusher       http.Flusher
-	subscribeOnce bool
-	sse           bool
-	multipart     bool
-	buf           *bytes.Buffer
-	firstMessage  bool
+	ctx             context.Context
+	cancel          context.CancelFunc
+	writer          io.Writer
+	flusher         http.Flusher
+	responseControl *http.ResponseController
+	subscribeOnce   bool
+	sse             bool
+	multipart       bool
+	buf             *bytes.Buffer
+	firstMessage    bool
+	sseWriteTimeout time.Duration
+	logger          *zap.Logger
+	stats           statistics.EngineStatistics
+	telemetry       subscriptionTelemetryContext
+	requestContext  context.Context
+	disconnect      *subscriptionDisconnectTracker
+	delivery        *subscriptionDeliveryTracker
 	// apolloSubscriptionMultipartPrintBoundary if set to true will send the multipart boundary at the end of the message to allow
 	// misbehaving client (like apollo client) to read the message just sent before the next one or the heartbeat
 	apolloSubscriptionMultipartPrintBoundary bool
@@ -53,7 +74,17 @@ func (f *HttpFlushWriter) Complete() {
 		return
 	}
 	if f.sse {
-		_, _ = f.writer.Write([]byte("event: complete\ndata: \n\n"))
+		err := f.writeAndFlushSSE(func() error {
+			_, err := f.writer.Write([]byte("event: complete\ndata: \n\n"))
+			return err
+		})
+		observeSubscriptionFrame(f.stats, f.logger, f.telemetry, "complete", err)
+		if err != nil {
+			initiator, reason := disconnectReasonFromWriteError(err)
+			f.disconnect.disconnect(initiator, reason, err)
+		} else {
+			f.disconnect.disconnect("server", "normal_completion", nil)
+		}
 	} else if f.multipart {
 		// Write the final boundary in the multipart response
 		if f.apolloSubscriptionMultipartPrintBoundary {
@@ -63,15 +94,23 @@ func (f *HttpFlushWriter) Complete() {
 		}
 	}
 
-	// Flush before closing the writer to ensure all data is sent
-	f.flusher.Flush()
+	if !f.sse {
+		// Flush before closing the writer to ensure all data is sent.
+		f.flusher.Flush()
+	}
 
 	f.cancel()
 }
 
 func (f *HttpFlushWriter) Write(p []byte) (n int, err error) {
 	if err = f.ctx.Err(); err != nil {
-		return
+		err = wrapSubscriptionWriteError("buffer", err)
+		if f.sse {
+			f.delivery.observe(f.telemetry, p, 0, err)
+			initiator, reason := disconnectReasonFromWriteError(err)
+			f.disconnect.disconnect(initiator, reason, err)
+		}
+		return 0, err
 	}
 
 	return f.buf.Write(p)
@@ -85,12 +124,16 @@ func (f *HttpFlushWriter) Heartbeat() error {
 	var heartbeat []byte
 	if f.sse {
 		heartbeat = []byte(":heartbeat\n\n")
-
-		if _, err := f.writer.Write(heartbeat); err != nil {
+		err := f.writeAndFlushSSE(func() error {
+			_, err := f.writer.Write(heartbeat)
 			return err
+		})
+		observeSubscriptionFrame(f.stats, f.logger, f.telemetry, "heartbeat", err)
+		if err != nil {
+			initiator, reason := disconnectReasonFromWriteError(err)
+			f.disconnect.disconnect(initiator, reason, err)
 		}
-
-		f.flusher.Flush()
+		return err
 	} else if f.multipart {
 		if _, err := f.Write([]byte("{}")); err != nil {
 			return err
@@ -109,12 +152,42 @@ func (f *HttpFlushWriter) Error(data []byte) {
 		return
 	}
 	_, _ = f.buf.Write(data)
-	_ = f.Flush()
+	err := f.flush("terminal_error")
+	if f.sse {
+		observeSubscriptionFrame(f.stats, f.logger, f.telemetry, "terminal_error", err)
+		if err != nil {
+			initiator, reason := disconnectReasonFromWriteError(err)
+			f.disconnect.disconnect(initiator, reason, err)
+		} else {
+			f.disconnect.disconnect("server", "normal_completion", nil)
+		}
+	}
 	f.cancel()
 }
 
+func (f *HttpFlushWriter) subscriptionRequestEnded() {
+	if !f.sse {
+		return
+	}
+	if err := f.requestContext.Err(); err != nil {
+		f.disconnect.disconnect("client", "context_canceled", err)
+		return
+	}
+	f.disconnect.disconnect("server", "normal_completion", nil)
+}
+
 func (f *HttpFlushWriter) Flush() (err error) {
+	return f.flush("next")
+}
+
+func (f *HttpFlushWriter) flush(frameType string) (err error) {
 	if err = f.ctx.Err(); err != nil {
+		if f.sse && frameType == "next" {
+			err = wrapSubscriptionWriteError("buffer", err)
+			f.delivery.observe(f.telemetry, f.buf.Bytes(), 0, err)
+			initiator, reason := disconnectReasonFromWriteError(err)
+			f.disconnect.disconnect(initiator, reason, err)
+		}
 		return err
 	}
 
@@ -151,14 +224,30 @@ func (f *HttpFlushWriter) Flush() (err error) {
 	}
 
 	full := flushBreak + string(resp) + separation
-	_, err = f.writer.Write([]byte(full))
+	if f.sse {
+		started := time.Now()
+		err = f.writeAndFlushSSE(func() error {
+			_, writeErr := f.writer.Write([]byte(full))
+			return writeErr
+		})
+		if frameType == "next" {
+			f.delivery.observe(f.telemetry, []byte(full), time.Since(started), err)
+			if err != nil {
+				initiator, reason := disconnectReasonFromWriteError(err)
+				f.disconnect.disconnect(initiator, reason, err)
+			}
+		}
+	} else {
+		_, err = f.writer.Write([]byte(full))
+		if err == nil {
+			// Flush before closing the writer to ensure all data is sent.
+			f.flusher.Flush()
+		}
+	}
 	if err != nil {
 		return err
 	}
 
-	// Flush before closing the writer to ensure all data is sent
-	f.flusher.Flush()
-
 	if f.subscribeOnce {
 		defer f.cancel()
 	}
@@ -166,15 +255,31 @@ func (f *HttpFlushWriter) Flush() (err error) {
 	return nil
 }
 
-func GetSubscriptionResponseWriter(ctx *resolve.Context, r *http.Request, w http.ResponseWriter, apolloSubscriptionMultipartPrintBoundary bool) (*resolve.Context, resolve.SubscriptionResponseWriter, bool) {
+func (f *HttpFlushWriter) writeAndFlushSSE(write func() error) error {
+	if f.sseWriteTimeout > 0 {
+		if err := f.responseControl.SetWriteDeadline(time.Now().Add(f.sseWriteTimeout)); err != nil {
+			// Failing closed prevents a response writer without deadline support from
+			// reintroducing an unbounded shared-trigger stall.
+			return wrapSubscriptionWriteError("deadline", fmt.Errorf("set SSE write deadline: %w", err))
+		}
+	}
+
+	if err := write(); err != nil {
+		return wrapSubscriptionWriteError("write", err)
+	}
+
+	return wrapSubscriptionWriteError("flush", f.responseControl.Flush())
+}
+
+func GetSubscriptionResponseWriter(ctx *resolve.Context, r *http.Request, w http.ResponseWriter, opts SubscriptionResponseWriterOptions) (*resolve.Context, resolve.SubscriptionResponseWriter, error) {
 	if wfw, ok := w.(withFlushWriter); ok {
-		return ctx, wfw.SubscriptionResponseWriter(), true
+		return ctx, wfw.SubscriptionResponseWriter(), nil
 	}
 	wgParams := NegotiateSubscriptionParams(r, false)
 
 	flusher, ok := w.(http.Flusher)
 	if !ok {
-		return ctx, nil, false
+		return ctx, nil, errors.New("subscription response writer does not support flushing")
 	}
 
 	setSubscriptionHeaders(wgParams, r, w)
@@ -182,12 +287,22 @@ func GetSubscriptionResponseWriter(ctx *resolve.Context, r *http.Request, w http
 	flushWriter := &HttpFlushWriter{
 		writer:                                   w,
 		flusher:                                  flusher,
+		responseControl:                          http.NewResponseController(w),
 		sse:                                      wgParams.UseSse,
 		multipart:                                wgParams.UseMultipart,
 		subscribeOnce:                            wgParams.SubscribeOnce,
 		buf:                                      &bytes.Buffer{},
 		firstMessage:                             true,
-		apolloSubscriptionMultipartPrintBoundary: apolloSubscriptionMultipartPrintBoundary,
+		sseWriteTimeout:                          opts.SSEWriteTimeout,
+		logger:                                   opts.Logger,
+		stats:                                    opts.Stats,
+		telemetry:                                opts.Telemetry,
+		requestContext:                           r.Context(),
+		apolloSubscriptionMultipartPrintBoundary: opts.ApolloSubscriptionMultipartPrintBoundary,
+	}
+	if flushWriter.sse {
+		flushWriter.disconnect = newSubscriptionDisconnectTracker(flushWriter.stats, flushWriter.logger, flushWriter.telemetry)
+		flushWriter.delivery = newSubscriptionDeliveryTracker(flushWriter.stats, flushWriter.logger, "")
 	}
 
 	flushWriter.ctx, flushWriter.cancel = context.WithCancel(ctx.Context())
@@ -197,10 +312,21 @@ func GetSubscriptionResponseWriter(ctx *resolve.Context, r *http.Request, w http
 		ctx.ExecutionOptions.SendHeartbeat = true
 		// Flush the response head immediately so the client establishes the connection
 		// before the first message, instead of blocking until one is streamed.
-		flusher.Flush()
+		if wgParams.UseSse {
+			err := flushWriter.writeAndFlushSSE(func() error { return nil })
+			observeSubscriptionFrame(flushWriter.stats, flushWriter.logger, flushWriter.telemetry, "headers", err)
+			if err != nil {
+				initiator, reason := disconnectReasonFromWriteError(err)
+				flushWriter.disconnect.disconnect(initiator, reason, err)
+				flushWriter.cancel()
+				return ctx, nil, fmt.Errorf("flush initial SSE response headers: %w", err)
+			}
+		} else {
+			flusher.Flush()
+		}
 	}
 
-	return ctx, flushWriter, true
+	return ctx, flushWriter, nil
 }
 
 func wrapMultipartMessage(resp []byte, wrapPayload bool) ([]byte, error) {
diff --git a/router/core/subscription_response_writer_test.go b/router/core/subscription_response_writer_test.go
index 02db6b740..9b9dab0a0 100644
--- a/router/core/subscription_response_writer_test.go
+++ b/router/core/subscription_response_writer_test.go
@@ -2,16 +2,45 @@ package core
 
 import (
 	"context"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/wundergraph/cosmo/router/pkg/statistics"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	zapobserver "go.uber.org/zap/zaptest/observer"
 )
 
+type deadlineRecorder struct {
+	*httptest.ResponseRecorder
+	deadlines   []time.Time
+	deadlineErr error
+	flushErr    error
+}
+
+func (r *deadlineRecorder) SetWriteDeadline(deadline time.Time) error {
+	if r.deadlineErr != nil {
+		return r.deadlineErr
+	}
+	r.deadlines = append(r.deadlines, deadline)
+	return nil
+}
+
+func (r *deadlineRecorder) FlushError() error {
+	if r.flushErr != nil {
+		return r.flushErr
+	}
+	r.Flush()
+	return nil
+}
+
 func TestNegotiateSubscriptionParams(t *testing.T) {
 	type args struct {
 		r *http.Request
@@ -137,10 +166,114 @@ func TestGetSubscriptionResponseWriter(t *testing.T) {
 		req := httptest.NewRequest(http.MethodPost, "/graphql", nil)
 		req.Header.Set("Accept", sseMimeType)
 
-		_, _, ok := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, false)
-		require.True(t, ok)
+		_, _, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{})
+		require.NoError(t, err)
 
 		assert.Equal(t, sseMimeType, recorder.Header().Get("Content-Type"))
 		assert.True(t, recorder.Flushed, "expected the SSE response head to be flushed before any message is written")
 	})
+
+	t.Run("sets a fresh deadline for every SSE write and flush", func(t *testing.T) {
+		recorder := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+		req := httptest.NewRequest(http.MethodPost, "/graphql", nil)
+		req.Header.Set("Accept", sseMimeType)
+
+		_, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{SSEWriteTimeout: time.Second})
+		require.NoError(t, err)
+		require.Len(t, recorder.deadlines, 1, "expected the initial header flush to have a deadline")
+
+		_, err = writer.Write([]byte(`{"data":{"id":1}}`))
+		require.NoError(t, err)
+		require.NoError(t, writer.Flush())
+		require.Len(t, recorder.deadlines, 2, "expected the data frame to refresh the deadline")
+		assert.False(t, recorder.deadlines[1].Before(recorder.deadlines[0]))
+
+		require.NoError(t, writer.Heartbeat())
+		require.Len(t, recorder.deadlines, 3, "expected the heartbeat to refresh the deadline")
+
+		writer.Complete()
+		require.Len(t, recorder.deadlines, 4, "expected the completion frame to refresh the deadline")
+	})
+
+	t.Run("propagates an SSE flush error", func(t *testing.T) {
+		recorder := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+		req := httptest.NewRequest(http.MethodPost, "/graphql", nil)
+		req.Header.Set("Accept", sseMimeType)
+
+		_, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{})
+		require.NoError(t, err)
+
+		flushErr := errors.New("flush failed")
+		recorder.flushErr = flushErr
+		require.ErrorIs(t, writer.Heartbeat(), flushErr)
+	})
+
+	t.Run("observes a failed SSE event at the transport writer", func(t *testing.T) {
+		recorder := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+		req := httptest.NewRequest(http.MethodPost, "/graphql", nil)
+		req.Header.Set("Accept", sseMimeType)
+		logCore, logs := zapobserver.New(zapcore.DebugLevel)
+		stats := statistics.NewEngineStats(t.Context(), zap.NewNop(), false)
+
+		_, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{
+			Logger: zap.New(logCore),
+			Stats:  stats,
+			Telemetry: subscriptionTelemetryContext{
+				transport:     subscriptionTransportSSE,
+				requestID:     "request-1",
+				operationName: "ProductUpdated",
+			},
+		})
+		require.NoError(t, err)
+		recorder.flushErr = context.DeadlineExceeded
+
+		_, err = writer.Write([]byte(`{"data":{"productUpdated":{"id":"1"}}}`))
+		require.NoError(t, err)
+		require.ErrorIs(t, writer.Flush(), context.DeadlineExceeded)
+
+		require.Len(t, stats.GetReport().SubscriptionObservations, 4)
+		require.Equal(t, 1, logs.FilterMessage("Subscription event delivery failed").Len())
+		fields := logs.FilterMessage("Subscription event delivery failed").All()[0].ContextMap()
+		require.Equal(t, "sse", fields["transport"])
+		require.Equal(t, uint64(1), fields["delivery_sequence"])
+		require.Equal(t, "flush", fields["failure_stage"])
+		require.Equal(t, "timeout", fields["failure_reason"])
+	})
+
+	t.Run("propagates an SSE deadline error", func(t *testing.T) {
+		recorder := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+		req := httptest.NewRequest(http.MethodPost, "/graphql", nil)
+		req.Header.Set("Accept", sseMimeType)
+
+		_, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{SSEWriteTimeout: time.Second})
+		require.NoError(t, err)
+
+		deadlineErr := errors.New("deadline failed")
+		recorder.deadlineErr = deadlineErr
+		err = writer.Heartbeat()
+		assert.ErrorIs(t, err, deadlineErr)
+		assert.ErrorContains(t, err, "set SSE write deadline")
+	})
+
+	t.Run("fails closed when an SSE deadline is configured but unsupported", func(t *testing.T) {
+		recorder := httptest.NewRecorder()
+		req := httptest.NewRequest(http.MethodPost, "/graphql", nil)
+		req.Header.Set("Accept", sseMimeType)
+
+		_, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{SSEWriteTimeout: time.Second})
+		require.Error(t, err)
+		assert.ErrorIs(t, err, http.ErrNotSupported)
+		assert.ErrorContains(t, err, "set SSE write deadline")
+		assert.Nil(t, writer)
+	})
+
+	t.Run("does not require deadline support when the timeout is disabled", func(t *testing.T) {
+		recorder := httptest.NewRecorder()
+		req := httptest.NewRequest(http.MethodPost, "/graphql", nil)
+		req.Header.Set("Accept", sseMimeType)
+
+		_, writer, err := GetSubscriptionResponseWriter(resolve.NewContext(context.Background()), req, recorder, SubscriptionResponseWriterOptions{})
+		require.NoError(t, err)
+		assert.NotNil(t, writer)
+	})
 }
diff --git a/router/core/websocket.go b/router/core/websocket.go
index 3fe6ec334..513c58867 100644
--- a/router/core/websocket.go
+++ b/router/core/websocket.go
@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"regexp"
@@ -383,7 +384,7 @@ func (h *WebsocketHandler) handleUpgradeRequest(w http.ResponseWriter, r *http.R
 
 		requestLogger.Debug("Initializing websocket connection", zap.Error(err))
 
-		handler.Close(false, wsproto.CloseKindOf(err))
+		handler.CloseWithError(false, wsproto.CloseKindOf(err), err)
 		return
 	}
 
@@ -405,7 +406,7 @@ func (h *WebsocketHandler) handleUpgradeRequest(w http.ResponseWriter, r *http.R
 				}
 				http.Error(handler.w, http.StatusText(statusCode), statusCode)
 				_ = handler.writeErrorMessage(requestID, errorMessage)
-				handler.Close(false, wsproto.CloseKindNormal)
+				handler.close(false, wsproto.CloseKindNormal, "router", "authentication_rejected", err)
 				return
 			}
 		}
@@ -417,7 +418,7 @@ func (h *WebsocketHandler) handleUpgradeRequest(w http.ResponseWriter, r *http.R
 			if err != nil {
 				requestLogger.Error("Error parsing initial payload: %v", zap.Error(err))
 				_ = handler.writeErrorMessage(requestID, err)
-				handler.Close(false, wsproto.CloseKindNormal)
+				handler.CloseWithError(false, wsproto.CloseKindNormal, err)
 				return
 			}
 			jwtToken, ok := initialPayloadMap[fromInitialPayloadConfig.Key].(string)
@@ -425,7 +426,7 @@ func (h *WebsocketHandler) handleUpgradeRequest(w http.ResponseWriter, r *http.R
 				err := fmt.Errorf("invalid JWT token in initial payload: JWT token is not a string")
 				requestLogger.Error(err.Error())
 				_ = handler.writeErrorMessage(requestID, err)
-				handler.Close(false, wsproto.CloseKindNormal)
+				handler.CloseWithError(false, wsproto.CloseKindNormal, err)
 				return
 			}
 			handler.request.Header.Set(fromInitialPayloadConfig.ExportToken.HeaderKey, jwtToken)
@@ -439,7 +440,7 @@ func (h *WebsocketHandler) handleUpgradeRequest(w http.ResponseWriter, r *http.R
 		err = h.addConnection(c, handler)
 		if err != nil {
 			requestLogger.Error("Adding connection to net poller", zap.Error(err))
-			handler.Close(true, wsproto.CloseKindNormal)
+			handler.CloseWithError(true, wsproto.CloseKindNormal, err)
 		}
 		return
 	}
@@ -466,7 +467,7 @@ func (h *WebsocketHandler) handleConnectionSync(handler *WebSocketConnectionHand
 					continue
 				}
 				h.logger.Debug("Client closed connection", zap.Error(err))
-				handler.Close(true, wsproto.CloseKindOf(err))
+				handler.CloseWithError(true, wsproto.CloseKindOf(err), err)
 				return
 			}
 			err = h.HandleMessage(handler, msg)
@@ -474,7 +475,7 @@ func (h *WebsocketHandler) handleConnectionSync(handler *WebSocketConnectionHand
 				h.logger.Debug("Handling websocket message", zap.Error(err))
 				var closeErr *wsproto.CloseError
 				if errors.As(err, &closeErr) {
-					handler.Close(true, closeErr.Kind)
+					handler.CloseWithError(true, closeErr.Kind, err)
 					return
 				}
 			}
@@ -494,7 +495,7 @@ func (h *WebsocketHandler) addConnection(conn net.Conn, handler *WebSocketConnec
 	return h.netPoll.Add(underlyingConn(conn))
 }
 
-func (h *WebsocketHandler) removeConnection(conn net.Conn, handler *WebSocketConnectionHandler, fd int, closeKind wsproto.CloseKind) {
+func (h *WebsocketHandler) removeConnection(conn net.Conn, handler *WebSocketConnectionHandler, fd int, closeKind wsproto.CloseKind, cause error) {
 	h.stats.ConnectionsDec()
 	h.connectionsMu.Lock()
 	delete(h.connections, fd)
@@ -503,7 +504,7 @@ func (h *WebsocketHandler) removeConnection(conn net.Conn, handler *WebSocketCon
 	if err != nil {
 		h.logger.Warn("Removing connection from net poller", zap.Error(err))
 	}
-	handler.Close(true, closeKind)
+	handler.CloseWithError(true, closeKind, cause)
 }
 
 // underlyingConn unwraps a *tls.Conn to the network connection it wraps. wss
@@ -581,7 +582,7 @@ func (h *WebsocketHandler) runPoller() {
 
 				if fd == 0 {
 					h.logger.Debug("Invalid socket fd", zap.Int("fd", fd))
-					h.removeConnection(conn, handler, fd, wsproto.CloseKindNormal)
+					h.removeConnection(conn, handler, fd, wsproto.CloseKindNormal, errors.New("invalid socket file descriptor"))
 					continue
 				}
 
@@ -591,7 +592,7 @@ func (h *WebsocketHandler) runPoller() {
 						continue
 					}
 					h.logger.Debug("Client closed connection", zap.Error(err))
-					h.removeConnection(conn, handler, fd, wsproto.CloseKindOf(err))
+					h.removeConnection(conn, handler, fd, wsproto.CloseKindOf(err), err)
 					continue
 				}
 				err = h.HandleMessage(handler, msg)
@@ -602,7 +603,7 @@ func (h *WebsocketHandler) runPoller() {
 					// which defaults to CloseKindNormal
 					var closeErr *wsproto.CloseError
 					if errors.As(err, &closeErr) {
-						h.removeConnection(conn, handler, fd, closeErr.Kind)
+						h.removeConnection(conn, handler, fd, closeErr.Kind, err)
 						continue
 					}
 				}
@@ -636,6 +637,8 @@ type websocketResponseWriter struct {
 	stats           statistics.EngineStatistics
 	propagateErrors bool
 	subscriptions   *sync.Map
+	telemetry       subscriptionTelemetryContext
+	delivery        *subscriptionDeliveryTracker
 }
 
 var (
@@ -643,7 +646,7 @@ var (
 	_ resolve.SubscriptionResponseWriter = (*websocketResponseWriter)(nil)
 )
 
-func newWebsocketResponseWriter(id string, protocol wsproto.Proto, propagateErrors bool, logger *zap.Logger, stats statistics.EngineStatistics, subscriptions *sync.Map) *websocketResponseWriter {
+func newWebsocketResponseWriter(id string, protocol wsproto.Proto, propagateErrors bool, logger *zap.Logger, stats statistics.EngineStatistics, subscriptions *sync.Map, telemetry subscriptionTelemetryContext) *websocketResponseWriter {
 	return &websocketResponseWriter{
 		id:              id,
 		protocol:        protocol,
@@ -652,6 +655,8 @@ func newWebsocketResponseWriter(id string, protocol wsproto.Proto, propagateErro
 		stats:           stats,
 		propagateErrors: propagateErrors,
 		subscriptions:   subscriptions,
+		telemetry:       telemetry,
+		delivery:        newSubscriptionDeliveryTracker(stats, logger, id),
 	}
 }
 
@@ -668,6 +673,8 @@ func (rw *websocketResponseWriter) Complete() {
 		rw.subscriptions.Delete(rw.id)
 	}
 	err := rw.protocol.Complete(rw.id)
+	err = wrapSubscriptionWriteError("write", err)
+	observeSubscriptionFrame(rw.stats, rw.logger, rw.telemetry, "complete", err)
 	if err != nil {
 		rw.logger.Debug("Sending complete message", zap.Error(err))
 	}
@@ -700,15 +707,22 @@ func (rw *websocketResponseWriter) Error(data []byte) {
 		errors = json.RawMessage(`[{"message":"Unable to subscribe"}]`)
 	}
 	if err := rw.protocol.WriteGraphQLErrors(rw.id, errors, nil); err != nil {
+		err = wrapSubscriptionWriteError("write", err)
+		observeSubscriptionFrame(rw.stats, rw.logger, rw.telemetry, "terminal_error", err)
 		rw.logger.Debug("Sending error message", zap.Error(err))
 		return
 	}
+	observeSubscriptionFrame(rw.stats, rw.logger, rw.telemetry, "terminal_error", nil)
 	// subscriptions-transport-ws clients rely on an explicit "complete" to end
 	// the stream after a data+errors frame. graphql-transport-ws treats the
 	// "error" frame as terminal per spec, so no follow-up is needed there.
 	if rw.protocol.Subprotocol() == wsproto.SubscriptionsTransportWSSubprotocol {
 		if err := rw.protocol.Complete(rw.id); err != nil {
+			err = wrapSubscriptionWriteError("write", err)
+			observeSubscriptionFrame(rw.stats, rw.logger, rw.telemetry, "complete", err)
 			rw.logger.Debug("Sending complete after error", zap.Error(err))
+		} else {
+			observeSubscriptionFrame(rw.stats, rw.logger, rw.telemetry, "complete", nil)
 		}
 	}
 }
@@ -728,6 +742,8 @@ func (rw *websocketResponseWriter) Flush() error {
 				"response_headers": rw.header,
 			})
 			if err != nil {
+				err = wrapSubscriptionWriteError("serialize", err)
+				rw.delivery.observe(rw.telemetry, payload, 0, err)
 				rw.logger.Warn("Serializing response headers", zap.Error(err))
 				return err
 			}
@@ -743,7 +759,9 @@ func (rw *websocketResponseWriter) Flush() error {
 			}
 		}
 
-		err = rw.protocol.WriteGraphQLData(rw.id, payload, extensions)
+		started := time.Now()
+		err = wrapSubscriptionWriteError("write", rw.protocol.WriteGraphQLData(rw.id, payload, extensions))
+		rw.delivery.observe(rw.telemetry, payload, time.Since(started), err)
 		rw.buf.Reset()
 		if err != nil {
 			return err
@@ -813,6 +831,7 @@ type WebSocketConnectionHandler struct {
 	subscriptionIDs atomic.Int64
 	subscriptions   sync.Map
 	stats           statistics.EngineStatistics
+	disconnect      *subscriptionDisconnectTracker
 
 	forwardInitialPayload bool
 
@@ -837,7 +856,14 @@ type forwardConfig struct {
 var detectNonRegex = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
 
 func NewWebsocketConnectionHandler(ctx context.Context, opts WebSocketConnectionHandlerOptions) *WebSocketConnectionHandler {
-	return &WebSocketConnectionHandler{
+	telemetry := subscriptionTelemetryContext{
+		transport:    subscriptionTransportWebSocket,
+		subprotocol:  opts.Protocol.Subprotocol(),
+		requestID:    opts.InitRequestID,
+		connectionID: opts.ConnectionID,
+		writeTimeout: opts.Connection.writeTimeout,
+	}
+	handler := &WebSocketConnectionHandler{
 		ctx:                          ctx,
 		operationProcessor:           opts.OperationProcessor,
 		operationBlocker:             opts.OperationBlocker,
@@ -862,6 +888,8 @@ func NewWebsocketConnectionHandler(ctx context.Context, opts WebSocketConnection
 		apolloCompatibilityFlags:     opts.ApolloCompatibilityFlags,
 		clientInfoFromInitialPayload: opts.ClientInfoFromInitialPayload,
 	}
+	handler.disconnect = newSubscriptionDisconnectTracker(opts.Stats, opts.Logger, telemetry)
+	return handler
 }
 
 func (h *WebSocketConnectionHandler) requestError(err error) error {
@@ -895,7 +923,9 @@ func (h *WebSocketConnectionHandler) writeErrorMessage(operationID string, err e
 	if err != nil {
 		return fmt.Errorf("encoding GraphQL errors: %w", err)
 	}
-	return h.protocol.WriteGraphQLErrors(operationID, payload, nil)
+	writeErr := wrapSubscriptionWriteError("write", h.protocol.WriteGraphQLErrors(operationID, payload, nil))
+	observeSubscriptionFrame(h.stats, h.logger, h.disconnect.telemetry, "terminal_error", writeErr)
+	return writeErr
 }
 
 func (h *WebSocketConnectionHandler) parseAndPlan(registration *SubscriptionRegistration) (*ParsedOperation, *operationContext, error) {
@@ -1066,7 +1096,13 @@ func (h *WebSocketConnectionHandler) parseAndPlan(registration *SubscriptionRegi
 }
 
 func (h *WebSocketConnectionHandler) executeSubscription(registration *SubscriptionRegistration) {
-	rw := newWebsocketResponseWriter(registration.msg.ID, h.protocol, h.graphqlHandler.subgraphErrorPropagation.Enabled, h.logger, h.stats, &h.subscriptions)
+	rw := newWebsocketResponseWriter(registration.msg.ID, h.protocol, h.graphqlHandler.subgraphErrorPropagation.Enabled, h.logger, h.stats, &h.subscriptions, subscriptionTelemetryContext{
+		transport:    subscriptionTransportWebSocket,
+		subprotocol:  h.protocol.Subprotocol(),
+		requestID:    h.initRequestID,
+		connectionID: h.connectionID,
+		writeTimeout: h.conn.writeTimeout,
+	})
 
 	_, operationCtx, err := h.parseAndPlan(registration)
 	if err != nil {
@@ -1076,6 +1112,7 @@ func (h *WebSocketConnectionHandler) executeSubscription(registration *Subscript
 		}
 		return
 	}
+	rw.telemetry.operationName = operationCtx.name
 
 	if h.forwardUpgradeHeaders.enabled && h.upgradeRequestHeaders != nil {
 		if operationCtx.extensions == nil {
@@ -1248,7 +1285,8 @@ func (h *WebSocketConnectionHandler) handleComplete(msg *wsproto.Message) error
 		ConnectionID:   h.connectionID,
 		SubscriptionID: subscriptionID,
 	}
-	_ = h.protocol.Complete(msg.ID)
+	writeErr := wrapSubscriptionWriteError("write", h.protocol.Complete(msg.ID))
+	observeSubscriptionFrame(h.stats, h.logger, h.disconnect.telemetry, "complete", writeErr)
 	return h.graphqlHandler.executor.Resolver.UnsubscribeSubscription(id)
 }
 
@@ -1257,7 +1295,8 @@ func (h *WebsocketHandler) HandleMessage(handler *WebSocketConnectionHandler, ms
 	case wsproto.MessageTypeTerminate:
 		return errClientTerminatedConnection
 	case wsproto.MessageTypePing:
-		_ = handler.protocol.Pong(msg)
+		writeErr := wrapSubscriptionWriteError("write", handler.protocol.Pong(msg))
+		observeSubscriptionFrame(handler.stats, handler.logger, handler.disconnect.telemetry, "pong", writeErr)
 	case wsproto.MessageTypePong:
 		// "Furthermore, the Pong message may even be sent unsolicited as a unidirectional heartbeat"
 		return nil
@@ -1311,7 +1350,6 @@ func (h *WebSocketConnectionHandler) Initialize() (err error) {
 				h.request.Header.Set(h.clientInfoFromInitialPayload.ForwardToRequestHeaders.VersionTargetHeader, clientVersion)
 			}
 		}
-
 		// Update planner options with new client info
 		h.plannerOptions.ClientInfo = h.clientInfo
 	}
@@ -1396,6 +1434,17 @@ func (h *WebSocketConnectionHandler) shouldComputeOperationSha256(operationKit *
 }
 
 func (h *WebSocketConnectionHandler) Close(unsubscribe bool, closeKind wsproto.CloseKind) {
+	initiator, reason := websocketDisconnectReason(nil, closeKind)
+	h.close(unsubscribe, closeKind, initiator, reason, nil)
+}
+
+func (h *WebSocketConnectionHandler) CloseWithError(unsubscribe bool, closeKind wsproto.CloseKind, err error) {
+	initiator, reason := websocketDisconnectReason(err, closeKind)
+	h.close(unsubscribe, closeKind, initiator, reason, err)
+}
+
+func (h *WebSocketConnectionHandler) close(unsubscribe bool, closeKind wsproto.CloseKind, initiator, reason string, cause error) {
+	h.disconnect.disconnect(initiator, reason, cause)
 	if unsubscribe {
 		// Remove any pending IDs associated with this connection
 		err := h.graphqlHandler.executor.Resolver.UnsubscribeClient(h.connectionID)
@@ -1412,3 +1461,38 @@ func (h *WebSocketConnectionHandler) Close(unsubscribe bool, closeKind wsproto.C
 		h.logger.Debug("Closing websocket connection", zap.Error(err))
 	}
 }
+
+func websocketDisconnectReason(err error, closeKind wsproto.CloseKind) (initiator, reason string) {
+	var netErr net.Error
+	var closeErr *wsproto.CloseError
+	var closedErr wsutil.ClosedError
+	var syntaxErr *json.SyntaxError
+	var typeErr *json.UnmarshalTypeError
+	switch {
+	case errors.Is(err, errClientTerminatedConnection), errors.Is(err, io.EOF):
+		return "client", "client_closed"
+	case errors.Is(err, net.ErrClosed):
+		return "router", "connection_closed"
+	case errors.As(err, &closedErr):
+		return "client", "client_closed"
+	case errors.As(err, &netErr) && netErr.Timeout():
+		return "network", "timeout"
+	case errors.As(err, &syntaxErr), errors.As(err, &typeErr):
+		return "client", "protocol_error"
+	case errors.As(err, &closeErr):
+		if closeErr.Kind.Code == ws.StatusNormalClosure {
+			return "client", "client_closed"
+		}
+		return "client", "protocol_error"
+	case closeKind == wsproto.CloseKindGoingAway:
+		return "server", "server_shutdown"
+	case closeKind == wsproto.CloseKindUnauthorized:
+		return "router", "authentication_rejected"
+	case closeKind == wsproto.CloseKindInvalidMessageType, closeKind == wsproto.CloseKindTooManyInits:
+		return "client", "protocol_error"
+	case err != nil:
+		return "network", "network_error"
+	default:
+		return "server", "normal_completion"
+	}
+}
diff --git a/router/pkg/config/config.go b/router/pkg/config/config.go
index 90d99e551..60aa2256f 100644
--- a/router/pkg/config/config.go
+++ b/router/pkg/config/config.go
@@ -500,6 +500,7 @@ type EngineExecutionConfiguration struct {
 	DisableVariablesRemapping                        bool          `envDefault:"false" env:"ENGINE_DISABLE_VARIABLES_REMAPPING" yaml:"disable_variables_remapping"`
 	EnableRequireFetchReasons                        bool          `envDefault:"false" env:"ENGINE_ENABLE_REQUIRE_FETCH_REASONS" yaml:"enable_require_fetch_reasons"`
 	SubscriptionFetchTimeout                         time.Duration `envDefault:"30s" env:"ENGINE_SUBSCRIPTION_FETCH_TIMEOUT" yaml:"subscription_fetch_timeout,omitempty"`
+	SSEServerWriteTimeout                            time.Duration `envDefault:"0s" env:"ENGINE_SSE_SERVER_WRITE_TIMEOUT" yaml:"sse_server_write_timeout,omitempty"`
 	EnableDefer                                      bool          `envDefault:"false" env:"ENGINE_ENABLE_DEFER" yaml:"enable_defer"`
 
 	// EnableMultiFetch merges entity fetches to the same subgraph that execute
@@ -1721,6 +1722,10 @@ func LoadConfig(configFilePaths []string) (*LoadResult, error) {
 		}
 	}
 
+	if cfg.Config.EngineExecutionConfiguration.SSEServerWriteTimeout < 0 {
+		return nil, errors.New("engine.sse_server_write_timeout must be greater or equal to 0s")
+	}
+
 	// Post-process the config
 	if cfg.Config.DevelopmentMode {
 		cfg.Config.JSONLog = false
diff --git a/router/pkg/config/config.schema.json b/router/pkg/config/config.schema.json
index 069292f78..15d8b15f1 100644
--- a/router/pkg/config/config.schema.json
+++ b/router/pkg/config/config.schema.json
@@ -4151,6 +4151,15 @@
           "default": "30s",
           "description": "The maximum time a subscription fetch can take before it is considered timed out. The period is specified as a string with a number and a unit, e.g. 10ms, 1s, 1m, 1h. The supported units are 'ms', 's', 'm', 'h'."
         },
+        "sse_server_write_timeout": {
+          "type": "string",
+          "format": "go-duration",
+          "duration": {
+            "minimum": "0s"
+          },
+          "default": "0s",
+          "description": "The maximum time allowed for each downstream SSE write and flush. When exceeded, the affected SSE subscription is terminated so it cannot indefinitely block other subscriptions sharing a trigger. A value of 0s disables the deadline."
+        },
         "enable_defer": {
           "type": "boolean",
           "default": false,
diff --git a/router/pkg/config/config_test.go b/router/pkg/config/config_test.go
index e05dca122..da7db70f1 100644
--- a/router/pkg/config/config_test.go
+++ b/router/pkg/config/config_test.go
@@ -478,6 +478,36 @@ telemetry:
 	require.Equal(t, "at '/telemetry/tracing/exporters/0/export_timeout': duration must be less or equal than 2m0s", js.Causes[0].Error())
 }
 
+func TestSSEServerWriteTimeoutRejectsNegativeValues(t *testing.T) {
+	t.Run("config file", func(t *testing.T) {
+		f := createTempFileFromFixture(t, `
+version: "1"
+
+graph:
+  token: "token"
+
+engine:
+  sse_server_write_timeout: -1s
+`)
+
+		_, err := LoadConfig([]string{f})
+		require.ErrorContains(t, err, "duration must be greater or equal than 0s")
+	})
+
+	t.Run("environment variable", func(t *testing.T) {
+		t.Setenv("ENGINE_SSE_SERVER_WRITE_TIMEOUT", "-1s")
+		f := createTempFileFromFixture(t, `
+version: "1"
+
+graph:
+  token: "token"
+`)
+
+		_, err := LoadConfig([]string{f})
+		require.EqualError(t, err, "engine.sse_server_write_timeout must be greater or equal to 0s")
+	})
+}
+
 func TestLoadFullConfig(t *testing.T) {
 	t.Parallel()
 
diff --git a/router/pkg/config/fixtures/full.yaml b/router/pkg/config/fixtures/full.yaml
index b953f28ae..46afdb6c2 100644
--- a/router/pkg/config/fixtures/full.yaml
+++ b/router/pkg/config/fixtures/full.yaml
@@ -450,6 +450,7 @@ engine:
   websocket_client_write_timeout: 10s
   websocket_server_read_timeout: 5s
   websocket_server_write_timeout: 10s
+  sse_server_write_timeout: 10s
   websocket_server_poll_timeout: 1s
   websocket_server_conn_buffer_size: 128
   websocket_client_read_limit: 1MB
diff --git a/router/pkg/config/json_schema.go b/router/pkg/config/json_schema.go
index 46bc4432a..dc0675f6f 100644
--- a/router/pkg/config/json_schema.go
+++ b/router/pkg/config/json_schema.go
@@ -27,8 +27,10 @@ const (
 )
 
 type duration struct {
-	min time.Duration
-	max time.Duration
+	min    time.Duration
+	max    time.Duration
+	hasMin bool
+	hasMax bool
 }
 
 func (d duration) Validate(ctx *jsonschema.ValidatorContext, v any) {
@@ -51,7 +53,7 @@ func (d duration) Validate(ctx *jsonschema.ValidatorContext, v any) {
 		return
 	}
 
-	if d.min > 0 {
+	if d.hasMin {
 		if duration < d.min {
 			ctx.AddError(&validationErrorKind{
 				fmt.Sprintf("duration must be greater or equal than %s", d.min),
@@ -61,7 +63,7 @@ func (d duration) Validate(ctx *jsonschema.ValidatorContext, v any) {
 		}
 	}
 
-	if d.max > 0 {
+	if d.hasMax {
 		if duration > d.max {
 			ctx.AddError(&validationErrorKind{
 				fmt.Sprintf("duration must be less or equal than %s", d.max),
@@ -118,23 +120,25 @@ func compileDuration(ctx *jsonschema.CompilerContext, m map[string]any) (jsonsch
 			var minDuration, maxDuration time.Duration
 			var err error
 
-			minDurationString, ok := mapVal["minimum"].(string)
-			if ok {
+			minDurationString, hasMin := mapVal["minimum"].(string)
+			if hasMin {
 				minDuration, err = time.ParseDuration(minDurationString)
 				if err != nil {
 					return nil, err
 				}
 			}
-			maxDurationString, ok := mapVal["maximum"].(string)
-			if ok {
+			maxDurationString, hasMax := mapVal["maximum"].(string)
+			if hasMax {
 				maxDuration, err = time.ParseDuration(maxDurationString)
 				if err != nil {
 					return nil, err
 				}
 			}
 			return duration{
-				min: minDuration,
-				max: maxDuration,
+				min:    minDuration,
+				max:    maxDuration,
+				hasMin: hasMin,
+				hasMax: hasMax,
 			}, nil
 		}
 
diff --git a/router/pkg/config/testdata/config_defaults.json b/router/pkg/config/testdata/config_defaults.json
index ad7e252cf..36751bb81 100644
--- a/router/pkg/config/testdata/config_defaults.json
+++ b/router/pkg/config/testdata/config_defaults.json
@@ -519,6 +519,7 @@
     "DisableVariablesRemapping": false,
     "EnableRequireFetchReasons": false,
     "SubscriptionFetchTimeout": 30000000000,
+    "SSEServerWriteTimeout": 0,
     "EnableDefer": false,
     "EnableMultiFetch": false,
     "EnableScheduleFetches": false,
diff --git a/router/pkg/config/testdata/config_full.json b/router/pkg/config/testdata/config_full.json
index f610b51c1..189a37e0e 100644
--- a/router/pkg/config/testdata/config_full.json
+++ b/router/pkg/config/testdata/config_full.json
@@ -987,6 +987,7 @@
     "DisableVariablesRemapping": false,
     "EnableRequireFetchReasons": false,
     "SubscriptionFetchTimeout": 30000000000,
+    "SSEServerWriteTimeout": 10000000000,
     "EnableDefer": false,
     "EnableMultiFetch": false,
     "EnableScheduleFetches": false,
diff --git a/router/pkg/metric/engine_metrics.go b/router/pkg/metric/engine_metrics.go
index 204331c8e..26eecd89c 100644
--- a/router/pkg/metric/engine_metrics.go
+++ b/router/pkg/metric/engine_metrics.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 
+	rotel "github.com/wundergraph/cosmo/router/pkg/otel"
 	"github.com/wundergraph/cosmo/router/pkg/statistics"
 	"go.opentelemetry.io/otel/attribute"
 	otelmetric "go.opentelemetry.io/otel/metric"
@@ -21,6 +22,9 @@ const (
 	engineSubscriptionCountKey      = engineMetricBaseKey + "subscriptions"
 	engineTriggerCountKey           = engineMetricBaseKey + "triggers"
 	engineMessagesSentKey           = engineMetricBaseKey + "messages.sent"
+	subscriptionDeliveryAttemptsKey = "router.subscription.delivery.attempts"
+	subscriptionDeliveryFailuresKey = "router.subscription.delivery.write.failures"
+	subscriptionDisconnectsKey      = "router.subscription.disconnects"
 	engineResolversMaxConcurrentKey = engineMetricBaseKey + "resolvers.max_concurrent"
 	engineResolversInflightKey      = engineMetricBaseKey + "resolvers.inflight"
 )
@@ -30,6 +34,9 @@ type engineInstruments struct {
 	subscriptionCount      otelmetric.Int64ObservableUpDownCounter
 	triggerCount           otelmetric.Int64ObservableUpDownCounter
 	messagesSent           otelmetric.Int64ObservableCounter
+	deliveryAttempts       otelmetric.Int64ObservableCounter
+	deliveryFailures       otelmetric.Int64ObservableCounter
+	disconnects            otelmetric.Int64ObservableCounter
 	resolversMaxConcurrent otelmetric.Int64ObservableUpDownCounter
 	resolversInflight      otelmetric.Int64ObservableUpDownCounter
 }
@@ -52,6 +59,15 @@ func (i *engineInstruments) toList() []otelmetric.Observable {
 	if i.messagesSent != nil {
 		result = append(result, i.messagesSent)
 	}
+	if i.deliveryAttempts != nil {
+		result = append(result, i.deliveryAttempts)
+	}
+	if i.deliveryFailures != nil {
+		result = append(result, i.deliveryFailures)
+	}
+	if i.disconnects != nil {
+		result = append(result, i.disconnects)
+	}
 
 	if i.resolversMaxConcurrent != nil {
 		result = append(result, i.resolversMaxConcurrent)
@@ -115,6 +131,9 @@ func setupInstruments(m otelmetric.Meter, statConfig *EngineStatsConfig, resolve
 		subscriptionCount      otelmetric.Int64ObservableUpDownCounter
 		triggerCount           otelmetric.Int64ObservableUpDownCounter
 		messagesSent           otelmetric.Int64ObservableCounter
+		deliveryAttempts       otelmetric.Int64ObservableCounter
+		deliveryFailures       otelmetric.Int64ObservableCounter
+		disconnects            otelmetric.Int64ObservableCounter
 		resolversMaxConcurrent otelmetric.Int64ObservableUpDownCounter
 		resolversInflight      otelmetric.Int64ObservableUpDownCounter
 	)
@@ -144,6 +163,22 @@ func setupInstruments(m otelmetric.Meter, statConfig *EngineStatsConfig, resolve
 		if err != nil {
 			return nil, err
 		}
+
+		deliveryAttempts, err = m.Int64ObservableCounter(subscriptionDeliveryAttemptsKey,
+			otelmetric.WithDescription("Number of downstream subscription delivery attempts."))
+		if err != nil {
+			return nil, err
+		}
+		deliveryFailures, err = m.Int64ObservableCounter(subscriptionDeliveryFailuresKey,
+			otelmetric.WithDescription("Number of downstream subscription write failures."))
+		if err != nil {
+			return nil, err
+		}
+		disconnects, err = m.Int64ObservableCounter(subscriptionDisconnectsKey,
+			otelmetric.WithDescription("Number of downstream subscription transport disconnects."))
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	if resolverStats {
@@ -165,6 +200,9 @@ func setupInstruments(m otelmetric.Meter, statConfig *EngineStatsConfig, resolve
 		subscriptionCount:      subscriptionCount,
 		triggerCount:           triggerCount,
 		messagesSent:           messagesSent,
+		deliveryAttempts:       deliveryAttempts,
+		deliveryFailures:       deliveryFailures,
+		disconnects:            disconnects,
 		resolversMaxConcurrent: resolversMaxConcurrent,
 		resolversInflight:      resolversInflight,
 	}, nil
@@ -200,6 +238,31 @@ func (e *EngineMetrics) observeInstruments(o otelmetric.Observer, stats statisti
 		o.ObserveInt64(e.instruments.subscriptionCount, int64(report.Subscriptions), otelmetric.WithAttributes(e.baseAttributes...))
 		o.ObserveInt64(e.instruments.triggerCount, int64(report.Triggers), otelmetric.WithAttributes(e.baseAttributes...))
 		o.ObserveInt64(e.instruments.messagesSent, int64(report.MessagesSent), otelmetric.WithAttributes(e.baseAttributes...))
+		for _, item := range report.SubscriptionObservations {
+			attrs := append([]attribute.KeyValue{}, e.baseAttributes...)
+			attrs = append(attrs, rotel.WgSubscriptionTransport.String(item.Observation.Transport))
+			if item.Observation.Subprotocol != "" {
+				attrs = append(attrs, rotel.WgWebSocketSubprotocol.String(item.Observation.Subprotocol))
+			}
+			switch item.Observation.Kind {
+			case statistics.SubscriptionObservationDeliveryAttempt:
+				attrs = append(attrs, rotel.WgSubscriptionFrameType.String(item.Observation.FrameType))
+				o.ObserveInt64(e.instruments.deliveryAttempts, int64(item.Count), otelmetric.WithAttributes(attrs...))
+			case statistics.SubscriptionObservationDeliveryFailure:
+				attrs = append(attrs,
+					rotel.WgSubscriptionFrameType.String(item.Observation.FrameType),
+					rotel.WgSubscriptionFailureStage.String(item.Observation.FailureStage),
+					rotel.WgSubscriptionFailureReason.String(item.Observation.FailureReason),
+				)
+				o.ObserveInt64(e.instruments.deliveryFailures, int64(item.Count), otelmetric.WithAttributes(attrs...))
+			case statistics.SubscriptionObservationDisconnect:
+				attrs = append(attrs,
+					rotel.WgSubscriptionDisconnectInitiator.String(item.Observation.Initiator),
+					rotel.WgSubscriptionDisconnectReason.String(item.Observation.DisconnectReason),
+				)
+				o.ObserveInt64(e.instruments.disconnects, int64(item.Count), otelmetric.WithAttributes(attrs...))
+			}
+		}
 	}
 
 	if e.instruments.resolversMaxConcurrent != nil {
diff --git a/router/pkg/metric/engine_metrics_test.go b/router/pkg/metric/engine_metrics_test.go
new file mode 100644
index 000000000..664759094
--- /dev/null
+++ b/router/pkg/metric/engine_metrics_test.go
@@ -0,0 +1,77 @@
+package metric
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	rotel "github.com/wundergraph/cosmo/router/pkg/otel"
+	"github.com/wundergraph/cosmo/router/pkg/statistics"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.uber.org/zap"
+)
+
+func TestEngineMetricsExportsSubscriptionDeliveryAndDisconnectCounters(t *testing.T) {
+	reader := sdkmetric.NewManualReader()
+	provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	stats := statistics.NewEngineStats(t.Context(), zap.NewNop(), false)
+	metrics, err := NewEngineMetrics(zap.NewNop(), nil, provider, stats, &EngineStatsConfig{Subscription: true}, false)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, metrics.Shutdown()) })
+
+	stats.ObserveSubscription(statistics.SubscriptionObservation{
+		Kind:      statistics.SubscriptionObservationDeliveryAttempt,
+		Transport: "sse",
+		FrameType: "next",
+	})
+	stats.ObserveSubscription(statistics.SubscriptionObservation{
+		Kind:          statistics.SubscriptionObservationDeliveryFailure,
+		Transport:     "sse",
+		FrameType:     "next",
+		FailureStage:  "flush",
+		FailureReason: "timeout",
+	})
+	stats.ObserveSubscription(statistics.SubscriptionObservation{
+		Kind:             statistics.SubscriptionObservationDisconnect,
+		Transport:        "sse",
+		Initiator:        "router",
+		DisconnectReason: "write_timeout",
+	})
+
+	var resourceMetrics metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(context.Background(), &resourceMetrics))
+	type metricPoint struct {
+		value      int64
+		attributes map[string]string
+	}
+	points := make(map[string]metricPoint)
+	for _, scope := range resourceMetrics.ScopeMetrics {
+		for _, metric := range scope.Metrics {
+			sum, ok := metric.Data.(metricdata.Sum[int64])
+			if !ok || len(sum.DataPoints) == 0 {
+				continue
+			}
+			attrs := make(map[string]string)
+			for _, attr := range sum.DataPoints[0].Attributes.ToSlice() {
+				attrs[string(attr.Key)] = attr.Value.AsString()
+			}
+			points[metric.Name] = metricPoint{value: sum.DataPoints[0].Value, attributes: attrs}
+		}
+	}
+	require.Equal(t, metricPoint{value: 1, attributes: map[string]string{
+		string(rotel.WgSubscriptionTransport): "sse",
+		string(rotel.WgSubscriptionFrameType): "next",
+	}}, points[subscriptionDeliveryAttemptsKey])
+	require.Equal(t, metricPoint{value: 1, attributes: map[string]string{
+		string(rotel.WgSubscriptionTransport):     "sse",
+		string(rotel.WgSubscriptionFrameType):     "next",
+		string(rotel.WgSubscriptionFailureStage):  "flush",
+		string(rotel.WgSubscriptionFailureReason): "timeout",
+	}}, points[subscriptionDeliveryFailuresKey])
+	require.Equal(t, metricPoint{value: 1, attributes: map[string]string{
+		string(rotel.WgSubscriptionTransport):           "sse",
+		string(rotel.WgSubscriptionDisconnectInitiator): "router",
+		string(rotel.WgSubscriptionDisconnectReason):    "write_timeout",
+	}}, points[subscriptionDisconnectsKey])
+}
diff --git a/router/pkg/otel/attributes.go b/router/pkg/otel/attributes.go
index 07e0deaab..0b1337663 100644
--- a/router/pkg/otel/attributes.go
+++ b/router/pkg/otel/attributes.go
@@ -52,6 +52,13 @@ const (
 	WgIsBatchingOperation              = attribute.Key("wg.operation.batching.is_batched")
 	WgBatchingOperationsCount          = attribute.Key("wg.operation.batching.operations_count")
 	WgBatchingOperationIndex           = attribute.Key("wg.operation.batching.operation_index")
+	WgSubscriptionTransport            = attribute.Key("wg.subscription.transport")
+	WgSubscriptionFrameType            = attribute.Key("wg.subscription.frame_type")
+	WgSubscriptionFailureStage         = attribute.Key("wg.subscription.failure_stage")
+	WgSubscriptionFailureReason        = attribute.Key("wg.subscription.failure_reason")
+	WgSubscriptionDisconnectInitiator  = attribute.Key("wg.subscription.disconnect.initiator")
+	WgSubscriptionDisconnectReason     = attribute.Key("wg.subscription.disconnect.reason")
+	WgWebSocketSubprotocol             = attribute.Key("wg.websocket.subprotocol")
 	// HTTPRequestUploadFileCount is the number of files uploaded in a request (Not specified in the OpenTelemetry specification)
 	HTTPRequestUploadFileCount = attribute.Key("http.request.upload.file_count")
 
diff --git a/router/pkg/statistics/engine_stats.go b/router/pkg/statistics/engine_stats.go
index 80b159afe..6a17850d9 100644
--- a/router/pkg/statistics/engine_stats.go
+++ b/router/pkg/statistics/engine_stats.go
@@ -30,26 +30,59 @@ type EngineStatistics interface {
 	UnregisterResolver(r ResolverConcurrencyReporter)
 }
 
+type SubscriptionObservationKind string
+
+const (
+	SubscriptionObservationDeliveryAttempt SubscriptionObservationKind = "delivery_attempt"
+	SubscriptionObservationDeliveryFailure SubscriptionObservationKind = "delivery_failure"
+	SubscriptionObservationDisconnect      SubscriptionObservationKind = "disconnect"
+)
+
+// SubscriptionObservation contains only bounded values suitable for metric
+// dimensions. Event, connection, operation, and client identity belong in
+// structured logs and must not be added here.
+type SubscriptionObservation struct {
+	Kind             SubscriptionObservationKind
+	Transport        string
+	FrameType        string
+	FailureStage     string
+	FailureReason    string
+	Initiator        string
+	DisconnectReason string
+	Subprotocol      string
+}
+
+type SubscriptionObserver interface {
+	ObserveSubscription(SubscriptionObservation)
+}
+
+type SubscriptionObservationCount struct {
+	Observation SubscriptionObservation
+	Count       uint64
+}
+
 type EngineStats struct {
-	ctx           context.Context
-	logger        *zap.Logger
-	reportStats   bool
-	connections   atomic.Uint64
-	subscriptions atomic.Uint64
-	messagesSent  atomic.Uint64
-	triggers      atomic.Uint64
+	ctx                      context.Context
+	logger                   *zap.Logger
+	reportStats              bool
+	connections              atomic.Uint64
+	subscriptions            atomic.Uint64
+	messagesSent             atomic.Uint64
+	triggers                 atomic.Uint64
+	subscriptionObservations sync.Map // map[SubscriptionObservation]*atomic.Uint64
 
 	resolverMu        sync.RWMutex
 	resolverReporters map[ResolverConcurrencyReporter]struct{}
 }
 
 type UsageReport struct {
-	Connections           uint64
-	Subscriptions         uint64
-	MessagesSent          uint64
-	Triggers              uint64
-	ResolverMaxConcurrent uint64
-	ResolverInflight      uint64
+	Connections              uint64
+	Subscriptions            uint64
+	MessagesSent             uint64
+	Triggers                 uint64
+	ResolverMaxConcurrent    uint64
+	ResolverInflight         uint64
+	SubscriptionObservations []SubscriptionObservationCount
 }
 
 // NewEngineStats creates a new EngineStats instance. If reportStats is true, the stats will be reported every 5 seconds.
@@ -79,6 +112,13 @@ func (s *EngineStats) GetReport() *UsageReport {
 		report.ResolverInflight += uint64(r.InflightResolves())
 	}
 	s.resolverMu.RUnlock()
+	s.subscriptionObservations.Range(func(key, value any) bool {
+		report.SubscriptionObservations = append(report.SubscriptionObservations, SubscriptionObservationCount{
+			Observation: key.(SubscriptionObservation),
+			Count:       value.(*atomic.Uint64).Load(),
+		})
+		return true
+	})
 	return report
 }
 
@@ -108,6 +148,11 @@ func (s *EngineStats) SubscriptionUpdateSent() {
 	s.messagesSent.Inc()
 }
 
+func (s *EngineStats) ObserveSubscription(observation SubscriptionObservation) {
+	counter, _ := s.subscriptionObservations.LoadOrStore(observation, &atomic.Uint64{})
+	counter.(*atomic.Uint64).Inc()
+}
+
 func (s *EngineStats) ConnectionsInc() {
 	s.connections.Inc()
 }
@@ -166,6 +211,8 @@ func (s *NoopEngineStats) GetReport() *UsageReport {
 
 func (s *NoopEngineStats) SubscriptionUpdateSent() {}
 
+func (s *NoopEngineStats) ObserveSubscription(_ SubscriptionObservation) {}
+
 func (s *NoopEngineStats) ConnectionsInc() {}
 
 func (s *NoopEngineStats) ConnectionsDec() {}
@@ -187,3 +234,5 @@ func (s *NoopEngineStats) UnregisterResolver(_ ResolverConcurrencyReporter) {}
 
 var _ EngineStatistics = &EngineStats{}
 var _ EngineStatistics = &NoopEngineStats{}
+var _ SubscriptionObserver = &EngineStats{}
+var _ SubscriptionObserver = &NoopEngineStats{}