diff --git a/buf.gen.yaml b/buf.gen.yaml
index d9f3b1c..f55c8ed 100644
--- a/buf.gen.yaml
+++ b/buf.gen.yaml
@@ -4,3 +4,7 @@ plugins:
     out: .
     opt:
       - module=github.com/getoptimum/optimum-gateway
+  - local: ["go", "tool", "protoc-gen-go-grpc"]
+    out: .
+    opt:
+      - module=github.com/getoptimum/optimum-gateway
diff --git a/cmd/main.go b/cmd/main.go
index fa7ada6..3fc843c 100644
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -12,6 +12,8 @@ import (
 	"syscall"
 	"time"
 
+	"google.golang.org/grpc"
+
 	commonio "github.com/getoptimum/optimum-common/pkg/io"
 	"github.com/getoptimum/optimum-common/pkg/logger"
 	"github.com/getoptimum/optimum-gateway/pkg/config"
@@ -161,17 +163,30 @@ func main() {
 	}
 
 	// Consumer block-stream (ADR-0011), opt-in and off by default. The hub must
-	// exist before the gateway so it can be wired as an emit sink.
+	// exist before the gateway so it can be wired as an emit sink; WS and gRPC
+	// share the one hub.
 	var streamServer *stream.Server
+	var streamGRPCServer *stream.GRPCServer
 	var hub *streamhub.Service
 	if appConf.StreamEnable {
 		hub = streamhub.New()
 		authenticator := stream.NewConsumerAuthenticator(authMgr, appConf.StreamRequireAuth)
+		// One limiter across both transports keeps the caps global, not
+		// per-transport; config validation guarantees the caps are > 0.
+		limiter := stream.NewConnLimiter(appConf.StreamMaxConns, appConf.StreamMaxConnsPerSub)
 		streamServer = stream.NewServer(hub, authenticator, stream.Config{
 			Addr:           appConf.StreamAddr,
 			MaxConns:       appConf.StreamMaxConns,
 			MaxConnsPerSub: appConf.StreamMaxConnsPerSub,
 			BufferSize:     appConf.StreamBufferSize,
+			Limiter:        limiter,
+		}, l)
+		streamGRPCServer = stream.NewGRPCServer(hub, authenticator, stream.Config{
+			Addr:           appConf.StreamGRPCAddr,
+			MaxConns:       appConf.StreamMaxConns,
+			MaxConnsPerSub: appConf.StreamMaxConnsPerSub,
+			BufferSize:     appConf.StreamBufferSize,
+			Limiter:        limiter,
 		}, l)
 	}
 
@@ -199,6 +214,14 @@ func main() {
 		}()
 	}
 
+	if streamGRPCServer != nil {
+		go func() {
+			if runErr := streamGRPCServer.Run(); runErr != nil && !errors.Is(runErr, grpc.ErrServerStopped) {
+				l.Fatal("failed to run consumer stream grpc server", runErr)
+			}
+		}()
+	}
+
 	<-c // This blocks the main thread until an interrupt is received
 	cancel()
 	_ = appRouter.Stop()
@@ -207,6 +230,9 @@ func main() {
 		_ = streamServer.Stop(shutdownCtx)
 		cancelShutdown()
 	}
+	if streamGRPCServer != nil {
+		streamGRPCServer.Stop()
+	}
 	srvGateway.Stop()
 	if lokiDone != nil {
 		<-lokiDone // wait for final Loki flush to complete
diff --git a/go.mod b/go.mod
index 3835493..c02bc39 100644
--- a/go.mod
+++ b/go.mod
@@ -5,6 +5,7 @@ go 1.26.5
 tool (
 	github.com/bufbuild/buf/cmd/buf
 	github.com/golangci/golangci-lint/v2/cmd/golangci-lint
+	google.golang.org/grpc/cmd/protoc-gen-go-grpc
 	google.golang.org/protobuf/cmd/protoc-gen-go
 )
 
@@ -31,6 +32,7 @@ require (
 	github.com/prometheus/common v0.70.1
 	github.com/prometheus/prometheus v0.311.3
 	github.com/stretchr/testify v1.11.1
+	google.golang.org/grpc v1.82.1
 	google.golang.org/protobuf v1.36.11
 	gopkg.in/yaml.v2 v2.4.0
 )
@@ -453,7 +455,7 @@ require (
 	google.golang.org/api v0.272.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20260523011958-0a33c5d7ca68 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260523011958-0a33c5d7ca68 // indirect
-	google.golang.org/grpc v1.82.1 // indirect
+	google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	honnef.co/go/tools v0.7.0 // indirect
diff --git a/go.sum b/go.sum
index 6f2b410..24361fc 100644
--- a/go.sum
+++ b/go.sum
@@ -1275,6 +1275,8 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20260523011958-0a33c5d7ca68 h1:
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260523011958-0a33c5d7ca68/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
 google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
 google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
+google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2 h1:rgSNvqscFZ1JgV/4wH5GOsZFSFkR2Eua9As3KIr2LlM=
+google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.6.2/go.mod h1:iMEtFwDlAhjDU9L5mY6U1XLwlIId/G3h+QcBHDIvrJ8=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
 google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/pkg/config/config.go b/pkg/config/config.go
index 97a1a1c..9ca05f4 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -263,6 +263,9 @@ func (c *AppConfig) Validate() error {
 		if err := validateStreamListener("stream_grpc_addr", c.StreamGRPCAddr, c.StreamRequireAuth); err != nil {
 			return err
 		}
+		if strings.TrimSpace(c.StreamAddr) == strings.TrimSpace(c.StreamGRPCAddr) {
+			return fmt.Errorf("stream_addr and stream_grpc_addr must differ, got %q", c.StreamAddr)
+		}
 		if c.StreamMaxConns <= 0 {
 			return fmt.Errorf("stream_max_conns must be > 0")
 		}
diff --git a/pkg/service/stream/grpc.go b/pkg/service/stream/grpc.go
new file mode 100644
index 0000000..3209655
--- /dev/null
+++ b/pkg/service/stream/grpc.go
@@ -0,0 +1,157 @@
+package stream
+
+import (
+	"context"
+	"net"
+	"strings"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+
+	"github.com/getoptimum/optimum-common/pkg/logger"
+	streamv1 "github.com/getoptimum/optimum-gateway/pkg/service/stream/v1"
+	"github.com/getoptimum/optimum-gateway/pkg/service/streamhub"
+	"github.com/getoptimum/optimum-gateway/pkg/service/telemetry"
+)
+
+// GRPCServer serves the consumer block-stream over gRPC on its own listener,
+// reusing the hub, authenticator, and connection caps of the WS transport.
+type GRPCServer struct {
+	streamv1.UnimplementedBlockStreamServiceServer
+	hub     *streamhub.Service
+	auth    ConsumerAuthenticator
+	cfg     Config
+	log     logger.AppLogger
+	limiter *ConnLimiter
+	grpcSrv *grpc.Server
+}
+
+// maxConcurrentStreams bounds Subscribe streams per connection. The caps run
+// after auth, so one unauthenticated socket would otherwise be unbounded.
+const maxConcurrentStreams = 256
+
+// NewGRPCServer builds the consumer gRPC server. It does not start listening;
+// call Run.
+func NewGRPCServer(hub *streamhub.Service, auth ConsumerAuthenticator, cfg Config, log logger.AppLogger) *GRPCServer {
+	cfg = withDefaults(cfg)
+	g := &GRPCServer{
+		hub:     hub,
+		auth:    auth,
+		cfg:     cfg,
+		log:     log.With(logger.WithService("stream-grpc")),
+		limiter: cfg.Limiter,
+		grpcSrv: grpc.NewServer(
+			// Reap dead peers on the WS clock; the gRPC default is a 2h ping.
+			grpc.KeepaliveParams(keepalive.ServerParameters{Time: pingPeriod, Timeout: writeWait}),
+			grpc.MaxConcurrentStreams(maxConcurrentStreams),
+		),
+	}
+	streamv1.RegisterBlockStreamServiceServer(g.grpcSrv, g)
+	return g
+}
+
+// Run serves until Stop is called; grpc.ErrServerStopped on a clean stop is
+// treated as normal by the caller.
+func (g *GRPCServer) Run() error {
+	lis, err := net.Listen("tcp", g.cfg.Addr)
+	if err != nil {
+		return err
+	}
+	g.log.Info("starting consumer stream grpc server", logger.WithString("addr", g.cfg.Addr))
+	return g.grpcSrv.Serve(lis)
+}
+
+// Stop hard-stops the server, canceling active Subscribe streams so shutdown
+// does not block on long-lived consumers.
+func (g *GRPCServer) Stop() { g.grpcSrv.Stop() }
+
+// Subscribe authenticates and enforces caps before opening the stream, then
+// drains the buffer as proto frames (metadata omits Raw); lagged on overflow.
+func (g *GRPCServer) Subscribe(req *streamv1.SubscribeRequest, stream grpc.ServerStreamingServer[streamv1.BlockEvent]) error {
+	mode, ok := normalizeMode(req.GetMode())
+	if !ok {
+		return status.Error(codes.InvalidArgument, "invalid mode")
+	}
+	if !topicsOK(req.GetTopics()...) {
+		return status.Error(codes.InvalidArgument, "unsupported topics")
+	}
+
+	ctx := stream.Context()
+	subject, err := g.auth.Authenticate(metadataToken(ctx))
+	if err != nil {
+		telemetry.RecordStreamAuthFailure()
+		return status.Error(codes.Unauthenticated, "unauthorized")
+	}
+	if !g.limiter.acquire(subject) {
+		return status.Error(codes.ResourceExhausted, "too many connections")
+	}
+	defer g.limiter.release(subject)
+
+	sub := g.hub.Subscribe(g.cfg.BufferSize)
+	defer sub.Close()
+
+	raw := mode == modeRaw
+	var lastDropped uint64
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case ev, ok := <-sub.Events():
+			if !ok {
+				return nil
+			}
+			if d := sub.Dropped(); d != lastDropped {
+				lastDropped = d
+				lag := &streamv1.BlockEvent{Frame: &streamv1.BlockEvent_Lagged{Lagged: &streamv1.Lagged{Dropped: d}}}
+				if err := stream.Send(lag); err != nil {
+					return err
+				}
+			}
+			if err := stream.Send(toProto(ev, raw)); err != nil {
+				return err
+			}
+			telemetry.RecordStreamEventSent()
+		}
+	}
+}
+
+func toProto(ev *streamhub.BlockEvent, raw bool) *streamv1.BlockEvent {
+	b := &streamv1.Block{
+		Slot:           ev.Slot,
+		ProposerIndex:  ev.ProposerIndex,
+		ParentRoot:     ev.ParentRoot,
+		StateRoot:      ev.StateRoot,
+		BlockSizeBytes: ev.BlockSizeBytes,
+		Topic:          ev.Topic,
+		Source:         string(ev.Source),
+		ReceivedAtMs:   ev.ReceivedAtMs,
+		GatewayId:      ev.GatewayID,
+		ForkDigest:     ev.ForkDigest,
+		Stale:          ev.Stale,
+	}
+	if raw {
+		b.Raw = ev.Raw
+	}
+	return &streamv1.BlockEvent{Frame: &streamv1.BlockEvent_Block{Block: b}}
+}
+
+// metadataToken reads the consumer JWT from the "authorization" gRPC metadata,
+// accepting either a bare token or a "Bearer <jwt>" value.
+func metadataToken(ctx context.Context) string {
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return ""
+	}
+	vals := md.Get("authorization")
+	if len(vals) == 0 {
+		return ""
+	}
+	tok := vals[0]
+	if after, ok := strings.CutPrefix(tok, "Bearer "); ok {
+		tok = after
+	}
+	return strings.TrimSpace(tok)
+}
diff --git a/pkg/service/stream/grpc_test.go b/pkg/service/stream/grpc_test.go
new file mode 100644
index 0000000..a611a1a
--- /dev/null
+++ b/pkg/service/stream/grpc_test.go
@@ -0,0 +1,147 @@
+package stream
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"google.golang.org/grpc/test/bufconn"
+
+	"github.com/getoptimum/optimum-common/pkg/logger"
+	streamv1 "github.com/getoptimum/optimum-gateway/pkg/service/stream/v1"
+	"github.com/getoptimum/optimum-gateway/pkg/service/streamhub"
+	"github.com/getoptimum/optimum-gateway/pkg/test_utils"
+)
+
+// newGRPCTestServer starts a GRPCServer over an in-memory bufconn and returns a
+// connected client. Auth is always required; loopback no-auth is covered by WS.
+func newGRPCTestServer(t *testing.T, cfg Config) (client streamv1.BlockStreamServiceClient, srv *GRPCServer, hub *streamhub.Service, rig *test_utils.AuthTestRig) {
+	t.Helper()
+	var authenticator ConsumerAuthenticator
+	authenticator, rig = testAuth(t, true)
+	hub = streamhub.New()
+	srv = NewGRPCServer(hub, authenticator, cfg, logger.NewAppSLogger(logger.Debug))
+
+	lis := bufconn.Listen(1 << 20)
+	go func() { _ = srv.grpcSrv.Serve(lis) }()
+	t.Cleanup(srv.grpcSrv.Stop)
+
+	conn, err := grpc.NewClient("passthrough:///bufnet",
+		grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }),
+		grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+	return streamv1.NewBlockStreamServiceClient(conn), srv, hub, rig
+}
+
+func authCtx(t *testing.T, rig *test_utils.AuthTestRig, subject string) context.Context {
+	t.Helper()
+	return metadata.AppendToOutgoingContext(context.Background(), "authorization", "Bearer "+streamToken(t, rig, subject))
+}
+
+func TestGRPC_RejectsWithoutToken(t *testing.T) {
+	client, _, hub, _ := newGRPCTestServer(t, Config{})
+
+	sub, err := client.Subscribe(context.Background(), &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+	_, err = sub.Recv()
+	require.Equal(t, codes.Unauthenticated, status.Code(err))
+	require.Zero(t, hub.SubscriberCount(), "rejected consumer must not create a subscriber")
+}
+
+func TestGRPC_DeliversFraming(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		mode      string
+		expectRaw bool
+	}{
+		{"metadata omits raw", "metadata", false},
+		{"raw includes bytes", "raw", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			client, _, hub, rig := newGRPCTestServer(t, Config{})
+			sub, err := client.Subscribe(authCtx(t, rig, "sub-1"), &streamv1.SubscribeRequest{Mode: tc.mode})
+			require.NoError(t, err)
+
+			waitSubscribed(t, hub, 1)
+			hub.Emit(sampleEvent())
+
+			ev, err := sub.Recv()
+			require.NoError(t, err)
+			require.Nil(t, ev.GetLagged(), "a block frame carries no lagged signal")
+			require.EqualValues(t, 42, ev.GetBlock().GetSlot())
+			if tc.expectRaw {
+				require.Equal(t, []byte("ssz-snappy-bytes"), ev.GetBlock().GetRaw())
+			} else {
+				require.Empty(t, ev.GetBlock().GetRaw())
+			}
+		})
+	}
+}
+
+func TestGRPC_LaggedOnOverflow(t *testing.T) {
+	client, _, hub, rig := newGRPCTestServer(t, Config{BufferSize: 1})
+	// Above the loop deadline below, so a blocked Recv fails instead of hanging.
+	ctx, cancel := context.WithTimeout(authCtx(t, rig, "sub-1"), 5*time.Second)
+	defer cancel()
+	sub, err := client.Subscribe(ctx, &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+
+	waitSubscribed(t, hub, 1)
+	for range 3000 {
+		hub.Emit(sampleEvent())
+	}
+
+	var sawLagged bool
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) && !sawLagged {
+		ev, rerr := sub.Recv()
+		require.NoError(t, rerr)
+		if lag := ev.GetLagged(); lag != nil {
+			require.Positive(t, lag.GetDropped())
+			sawLagged = true
+		}
+	}
+	require.True(t, sawLagged, "a lagged frame must be sent after overflow")
+}
+
+func TestGRPC_GlobalCapRejects(t *testing.T) {
+	client, _, hub, rig := newGRPCTestServer(t, Config{MaxConns: 1})
+
+	first, err := client.Subscribe(authCtx(t, rig, "sub-a"), &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+	waitSubscribed(t, hub, 1)
+
+	second, err := client.Subscribe(authCtx(t, rig, "sub-b"), &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+	_, err = second.Recv()
+	require.Equal(t, codes.ResourceExhausted, status.Code(err))
+
+	_ = first // keep the first stream open for the duration of the assertion
+}
+
+func TestGRPC_CleanupOnCancel(t *testing.T) {
+	client, srv, hub, rig := newGRPCTestServer(t, Config{})
+	ctx, cancel := context.WithCancel(authCtx(t, rig, "sub-1"))
+	_, err := client.Subscribe(ctx, &streamv1.SubscribeRequest{})
+	require.NoError(t, err)
+
+	waitSubscribed(t, hub, 1)
+	cancel()
+
+	// Cancel must unwind the handler, closing the subscriber (drop-counter entry
+	// included) and releasing the cap slot, so nothing leaks.
+	waitSubscribed(t, hub, 0)
+	require.Eventually(t, func() bool {
+		srv.limiter.mu.Lock()
+		defer srv.limiter.mu.Unlock()
+		return srv.limiter.conns == 0 && len(srv.limiter.perSub) == 0
+	}, 2*time.Second, 10*time.Millisecond)
+}
diff --git a/pkg/service/stream/transport.go b/pkg/service/stream/transport.go
new file mode 100644
index 0000000..0bda8a9
--- /dev/null
+++ b/pkg/service/stream/transport.go
@@ -0,0 +1,92 @@
+package stream
+
+import (
+	"sync"
+
+	"github.com/getoptimum/optimum-gateway/pkg/service/streamhub"
+	"github.com/getoptimum/optimum-gateway/pkg/service/telemetry"
+)
+
+const (
+	defaultMaxConns       = 256
+	defaultMaxConnsPerSub = 8
+)
+
+// withDefaults fills unset (<=0) caps so both transports share the same limits.
+func withDefaults(cfg Config) Config {
+	if cfg.MaxConns <= 0 {
+		cfg.MaxConns = defaultMaxConns
+	}
+	if cfg.MaxConnsPerSub <= 0 {
+		cfg.MaxConnsPerSub = defaultMaxConnsPerSub
+	}
+	if cfg.BufferSize <= 0 {
+		cfg.BufferSize = streamhub.DefaultBufferSize
+	}
+	if cfg.Limiter == nil {
+		cfg.Limiter = NewConnLimiter(cfg.MaxConns, cfg.MaxConnsPerSub)
+	}
+	return cfg
+}
+
+// normalizeMode defaults empty to metadata and reports whether the value is allowed.
+func normalizeMode(mode string) (string, bool) {
+	if mode == "" {
+		mode = modeMetadata
+	}
+	return mode, mode == modeMetadata || mode == modeRaw
+}
+
+// topicsOK is true when every topic is empty or the v1-only beacon_block topic.
+func topicsOK(topics ...string) bool {
+	for _, t := range topics {
+		if t != "" && t != defaultTopic {
+			return false
+		}
+	}
+	return true
+}
+
+// ConnLimiter enforces the global and per-subject connection caps. One instance
+// is shared by both transports so the caps stay global, not per-transport (ADR-0011).
+type ConnLimiter struct {
+	maxConns       int
+	maxConnsPerSub int
+
+	mu     sync.Mutex
+	conns  int
+	perSub map[string]int
+}
+
+// NewConnLimiter returns a limiter for the given caps.
+func NewConnLimiter(maxConns, maxConnsPerSub int) *ConnLimiter {
+	return &ConnLimiter{
+		maxConns:       maxConns,
+		maxConnsPerSub: maxConnsPerSub,
+		perSub:         make(map[string]int),
+	}
+}
+
+// acquire admits a connection for subject when both caps allow it.
+func (l *ConnLimiter) acquire(subject string) bool {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.conns >= l.maxConns || l.perSub[subject] >= l.maxConnsPerSub {
+		return false
+	}
+	l.conns++
+	l.perSub[subject]++
+	telemetry.IncStreamConnections()
+	return true
+}
+
+func (l *ConnLimiter) release(subject string) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	l.conns--
+	l.perSub[subject]--
+	if l.perSub[subject] <= 0 {
+		delete(l.perSub, subject)
+	}
+	telemetry.DecStreamConnections()
+}
diff --git a/pkg/service/stream/v1/stream.pb.go b/pkg/service/stream/v1/stream.pb.go
new file mode 100644
index 0000000..c3c377d
--- /dev/null
+++ b/pkg/service/stream/v1/stream.pb.go
@@ -0,0 +1,433 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.36.11
+// 	protoc        (unknown)
+// source: getoptimum/optimum_gateway/service/stream/v1/stream.proto
+
+package v1
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+	unsafe "unsafe"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// SubscribeRequest selects the payload mode and topics for a subscription.
+type SubscribeRequest struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Mode          string                 `protobuf:"bytes,1,opt,name=mode,proto3" json:"mode,omitempty"`     // "metadata" (default) or "raw"
+	Topics        []string               `protobuf:"bytes,2,rep,name=topics,proto3" json:"topics,omitempty"` // validated only; v1 has one topic, so nothing is filtered
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *SubscribeRequest) Reset() {
+	*x = SubscribeRequest{}
+	mi := &file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[0]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *SubscribeRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SubscribeRequest) ProtoMessage() {}
+
+func (x *SubscribeRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[0]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SubscribeRequest.ProtoReflect.Descriptor instead.
+func (*SubscribeRequest) Descriptor() ([]byte, []int) {
+	return file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *SubscribeRequest) GetMode() string {
+	if x != nil {
+		return x.Mode
+	}
+	return ""
+}
+
+func (x *SubscribeRequest) GetTopics() []string {
+	if x != nil {
+		return x.Topics
+	}
+	return nil
+}
+
+// BlockEvent is one frame: a block observation or a lag signal (ADR-0011).
+type BlockEvent struct {
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// frame tells an observation apart from a control signal.
+	//
+	// Types that are valid to be assigned to Frame:
+	//
+	//	*BlockEvent_Block
+	//	*BlockEvent_Lagged
+	Frame         isBlockEvent_Frame `protobuf_oneof:"frame"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *BlockEvent) Reset() {
+	*x = BlockEvent{}
+	mi := &file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[1]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *BlockEvent) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*BlockEvent) ProtoMessage() {}
+
+func (x *BlockEvent) ProtoReflect() protoreflect.Message {
+	mi := &file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[1]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use BlockEvent.ProtoReflect.Descriptor instead.
+func (*BlockEvent) Descriptor() ([]byte, []int) {
+	return file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *BlockEvent) GetFrame() isBlockEvent_Frame {
+	if x != nil {
+		return x.Frame
+	}
+	return nil
+}
+
+func (x *BlockEvent) GetBlock() *Block {
+	if x != nil {
+		if x, ok := x.Frame.(*BlockEvent_Block); ok {
+			return x.Block
+		}
+	}
+	return nil
+}
+
+func (x *BlockEvent) GetLagged() *Lagged {
+	if x != nil {
+		if x, ok := x.Frame.(*BlockEvent_Lagged); ok {
+			return x.Lagged
+		}
+	}
+	return nil
+}
+
+type isBlockEvent_Frame interface {
+	isBlockEvent_Frame()
+}
+
+type BlockEvent_Block struct {
+	Block *Block `protobuf:"bytes,1,opt,name=block,proto3,oneof"`
+}
+
+type BlockEvent_Lagged struct {
+	Lagged *Lagged `protobuf:"bytes,2,opt,name=lagged,proto3,oneof"`
+}
+
+func (*BlockEvent_Block) isBlockEvent_Frame() {}
+
+func (*BlockEvent_Lagged) isBlockEvent_Frame() {}
+
+// Block is one block observation (fields mirror the streamhub hub type).
+type Block struct {
+	state          protoimpl.MessageState `protogen:"open.v1"`
+	Slot           uint64                 `protobuf:"varint,1,opt,name=slot,proto3" json:"slot,omitempty"`
+	ProposerIndex  uint64                 `protobuf:"varint,2,opt,name=proposer_index,json=proposerIndex,proto3" json:"proposer_index,omitempty"`
+	ParentRoot     []byte                 `protobuf:"bytes,3,opt,name=parent_root,json=parentRoot,proto3" json:"parent_root,omitempty"`
+	StateRoot      []byte                 `protobuf:"bytes,4,opt,name=state_root,json=stateRoot,proto3" json:"state_root,omitempty"`
+	BlockSizeBytes uint64                 `protobuf:"varint,5,opt,name=block_size_bytes,json=blockSizeBytes,proto3" json:"block_size_bytes,omitempty"`
+	Topic          string                 `protobuf:"bytes,6,opt,name=topic,proto3" json:"topic,omitempty"`
+	Source         string                 `protobuf:"bytes,7,opt,name=source,proto3" json:"source,omitempty"`
+	ReceivedAtMs   int64                  `protobuf:"varint,8,opt,name=received_at_ms,json=receivedAtMs,proto3" json:"received_at_ms,omitempty"`
+	GatewayId      string                 `protobuf:"bytes,9,opt,name=gateway_id,json=gatewayId,proto3" json:"gateway_id,omitempty"`
+	ForkDigest     string                 `protobuf:"bytes,10,opt,name=fork_digest,json=forkDigest,proto3" json:"fork_digest,omitempty"`
+	Stale          bool                   `protobuf:"varint,11,opt,name=stale,proto3" json:"stale,omitempty"`
+	Raw            []byte                 `protobuf:"bytes,12,opt,name=raw,proto3" json:"raw,omitempty"` // present only in raw mode
+	unknownFields  protoimpl.UnknownFields
+	sizeCache      protoimpl.SizeCache
+}
+
+func (x *Block) Reset() {
+	*x = Block{}
+	mi := &file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[2]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Block) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Block) ProtoMessage() {}
+
+func (x *Block) ProtoReflect() protoreflect.Message {
+	mi := &file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[2]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Block.ProtoReflect.Descriptor instead.
+func (*Block) Descriptor() ([]byte, []int) {
+	return file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *Block) GetSlot() uint64 {
+	if x != nil {
+		return x.Slot
+	}
+	return 0
+}
+
+func (x *Block) GetProposerIndex() uint64 {
+	if x != nil {
+		return x.ProposerIndex
+	}
+	return 0
+}
+
+func (x *Block) GetParentRoot() []byte {
+	if x != nil {
+		return x.ParentRoot
+	}
+	return nil
+}
+
+func (x *Block) GetStateRoot() []byte {
+	if x != nil {
+		return x.StateRoot
+	}
+	return nil
+}
+
+func (x *Block) GetBlockSizeBytes() uint64 {
+	if x != nil {
+		return x.BlockSizeBytes
+	}
+	return 0
+}
+
+func (x *Block) GetTopic() string {
+	if x != nil {
+		return x.Topic
+	}
+	return ""
+}
+
+func (x *Block) GetSource() string {
+	if x != nil {
+		return x.Source
+	}
+	return ""
+}
+
+func (x *Block) GetReceivedAtMs() int64 {
+	if x != nil {
+		return x.ReceivedAtMs
+	}
+	return 0
+}
+
+func (x *Block) GetGatewayId() string {
+	if x != nil {
+		return x.GatewayId
+	}
+	return ""
+}
+
+func (x *Block) GetForkDigest() string {
+	if x != nil {
+		return x.ForkDigest
+	}
+	return ""
+}
+
+func (x *Block) GetStale() bool {
+	if x != nil {
+		return x.Stale
+	}
+	return false
+}
+
+func (x *Block) GetRaw() []byte {
+	if x != nil {
+		return x.Raw
+	}
+	return nil
+}
+
+// Lagged reports the cumulative drop count after a buffer overflow.
+type Lagged struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Dropped       uint64                 `protobuf:"varint,1,opt,name=dropped,proto3" json:"dropped,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *Lagged) Reset() {
+	*x = Lagged{}
+	mi := &file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[3]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *Lagged) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Lagged) ProtoMessage() {}
+
+func (x *Lagged) ProtoReflect() protoreflect.Message {
+	mi := &file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[3]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Lagged.ProtoReflect.Descriptor instead.
+func (*Lagged) Descriptor() ([]byte, []int) {
+	return file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *Lagged) GetDropped() uint64 {
+	if x != nil {
+		return x.Dropped
+	}
+	return 0
+}
+
+var File_getoptimum_optimum_gateway_service_stream_v1_stream_proto protoreflect.FileDescriptor
+
+const file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDesc = "" +
+	"\n" +
+	"9getoptimum/optimum_gateway/service/stream/v1/stream.proto\x12,getoptimum.optimum_gateway.service.stream.v1\">\n" +
+	"\x10SubscribeRequest\x12\x12\n" +
+	"\x04mode\x18\x01 \x01(\tR\x04mode\x12\x16\n" +
+	"\x06topics\x18\x02 \x03(\tR\x06topics\"\xb2\x01\n" +
+	"\n" +
+	"BlockEvent\x12K\n" +
+	"\x05block\x18\x01 \x01(\v23.getoptimum.optimum_gateway.service.stream.v1.BlockH\x00R\x05block\x12N\n" +
+	"\x06lagged\x18\x02 \x01(\v24.getoptimum.optimum_gateway.service.stream.v1.LaggedH\x00R\x06laggedB\a\n" +
+	"\x05frame\"\xe8\x02\n" +
+	"\x05Block\x12\x12\n" +
+	"\x04slot\x18\x01 \x01(\x04R\x04slot\x12%\n" +
+	"\x0eproposer_index\x18\x02 \x01(\x04R\rproposerIndex\x12\x1f\n" +
+	"\vparent_root\x18\x03 \x01(\fR\n" +
+	"parentRoot\x12\x1d\n" +
+	"\n" +
+	"state_root\x18\x04 \x01(\fR\tstateRoot\x12(\n" +
+	"\x10block_size_bytes\x18\x05 \x01(\x04R\x0eblockSizeBytes\x12\x14\n" +
+	"\x05topic\x18\x06 \x01(\tR\x05topic\x12\x16\n" +
+	"\x06source\x18\a \x01(\tR\x06source\x12$\n" +
+	"\x0ereceived_at_ms\x18\b \x01(\x03R\freceivedAtMs\x12\x1d\n" +
+	"\n" +
+	"gateway_id\x18\t \x01(\tR\tgatewayId\x12\x1f\n" +
+	"\vfork_digest\x18\n" +
+	" \x01(\tR\n" +
+	"forkDigest\x12\x14\n" +
+	"\x05stale\x18\v \x01(\bR\x05stale\x12\x10\n" +
+	"\x03raw\x18\f \x01(\fR\x03raw\"\"\n" +
+	"\x06Lagged\x12\x18\n" +
+	"\adropped\x18\x01 \x01(\x04R\adropped2\x9e\x01\n" +
+	"\x12BlockStreamService\x12\x87\x01\n" +
+	"\tSubscribe\x12>.getoptimum.optimum_gateway.service.stream.v1.SubscribeRequest\x1a8.getoptimum.optimum_gateway.service.stream.v1.BlockEvent0\x01B=Z;github.com/getoptimum/optimum-gateway/pkg/service/stream/v1b\x06proto3"
+
+var (
+	file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescOnce sync.Once
+	file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescData []byte
+)
+
+func file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescGZIP() []byte {
+	file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescOnce.Do(func() {
+		file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDesc), len(file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDesc)))
+	})
+	return file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDescData
+}
+
+var file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
+var file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_goTypes = []any{
+	(*SubscribeRequest)(nil), // 0: getoptimum.optimum_gateway.service.stream.v1.SubscribeRequest
+	(*BlockEvent)(nil),       // 1: getoptimum.optimum_gateway.service.stream.v1.BlockEvent
+	(*Block)(nil),            // 2: getoptimum.optimum_gateway.service.stream.v1.Block
+	(*Lagged)(nil),           // 3: getoptimum.optimum_gateway.service.stream.v1.Lagged
+}
+var file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_depIdxs = []int32{
+	2, // 0: getoptimum.optimum_gateway.service.stream.v1.BlockEvent.block:type_name -> getoptimum.optimum_gateway.service.stream.v1.Block
+	3, // 1: getoptimum.optimum_gateway.service.stream.v1.BlockEvent.lagged:type_name -> getoptimum.optimum_gateway.service.stream.v1.Lagged
+	0, // 2: getoptimum.optimum_gateway.service.stream.v1.BlockStreamService.Subscribe:input_type -> getoptimum.optimum_gateway.service.stream.v1.SubscribeRequest
+	1, // 3: getoptimum.optimum_gateway.service.stream.v1.BlockStreamService.Subscribe:output_type -> getoptimum.optimum_gateway.service.stream.v1.BlockEvent
+	3, // [3:4] is the sub-list for method output_type
+	2, // [2:3] is the sub-list for method input_type
+	2, // [2:2] is the sub-list for extension type_name
+	2, // [2:2] is the sub-list for extension extendee
+	0, // [0:2] is the sub-list for field type_name
+}
+
+func init() { file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_init() }
+func file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_init() {
+	if File_getoptimum_optimum_gateway_service_stream_v1_stream_proto != nil {
+		return
+	}
+	file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes[1].OneofWrappers = []any{
+		(*BlockEvent_Block)(nil),
+		(*BlockEvent_Lagged)(nil),
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: unsafe.Slice(unsafe.StringData(file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDesc), len(file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_rawDesc)),
+			NumEnums:      0,
+			NumMessages:   4,
+			NumExtensions: 0,
+			NumServices:   1,
+		},
+		GoTypes:           file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_goTypes,
+		DependencyIndexes: file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_depIdxs,
+		MessageInfos:      file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_msgTypes,
+	}.Build()
+	File_getoptimum_optimum_gateway_service_stream_v1_stream_proto = out.File
+	file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_goTypes = nil
+	file_getoptimum_optimum_gateway_service_stream_v1_stream_proto_depIdxs = nil
+}
diff --git a/pkg/service/stream/v1/stream_grpc.pb.go b/pkg/service/stream/v1/stream_grpc.pb.go
new file mode 100644
index 0000000..c58f9b0
--- /dev/null
+++ b/pkg/service/stream/v1/stream_grpc.pb.go
@@ -0,0 +1,130 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.6.2
+// - protoc             (unknown)
+// source: getoptimum/optimum_gateway/service/stream/v1/stream.proto
+
+package v1
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+	BlockStreamService_Subscribe_FullMethodName = "/getoptimum.optimum_gateway.service.stream.v1.BlockStreamService/Subscribe"
+)
+
+// BlockStreamServiceClient is the client API for BlockStreamService service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+//
+// BlockStreamService streams decoded beacon-block observations to consumers (ADR-0011).
+type BlockStreamServiceClient interface {
+	// Subscribe opens a read-only server stream of block observations.
+	Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[BlockEvent], error)
+}
+
+type blockStreamServiceClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewBlockStreamServiceClient(cc grpc.ClientConnInterface) BlockStreamServiceClient {
+	return &blockStreamServiceClient{cc}
+}
+
+func (c *blockStreamServiceClient) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[BlockEvent], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &BlockStreamService_ServiceDesc.Streams[0], BlockStreamService_Subscribe_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[SubscribeRequest, BlockEvent]{ClientStream: stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type BlockStreamService_SubscribeClient = grpc.ServerStreamingClient[BlockEvent]
+
+// BlockStreamServiceServer is the server API for BlockStreamService service.
+// All implementations must embed UnimplementedBlockStreamServiceServer
+// for forward compatibility.
+//
+// BlockStreamService streams decoded beacon-block observations to consumers (ADR-0011).
+type BlockStreamServiceServer interface {
+	// Subscribe opens a read-only server stream of block observations.
+	Subscribe(*SubscribeRequest, grpc.ServerStreamingServer[BlockEvent]) error
+	mustEmbedUnimplementedBlockStreamServiceServer()
+}
+
+// UnimplementedBlockStreamServiceServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedBlockStreamServiceServer struct{}
+
+func (UnimplementedBlockStreamServiceServer) Subscribe(*SubscribeRequest, grpc.ServerStreamingServer[BlockEvent]) error {
+	return status.Error(codes.Unimplemented, "method Subscribe not implemented")
+}
+func (UnimplementedBlockStreamServiceServer) mustEmbedUnimplementedBlockStreamServiceServer() {}
+func (UnimplementedBlockStreamServiceServer) testEmbeddedByValue()                            {}
+
+// UnsafeBlockStreamServiceServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to BlockStreamServiceServer will
+// result in compilation errors.
+type UnsafeBlockStreamServiceServer interface {
+	mustEmbedUnimplementedBlockStreamServiceServer()
+}
+
+func RegisterBlockStreamServiceServer(s grpc.ServiceRegistrar, srv BlockStreamServiceServer) {
+	// If the following call panics, it indicates UnimplementedBlockStreamServiceServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&BlockStreamService_ServiceDesc, srv)
+}
+
+func _BlockStreamService_Subscribe_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(SubscribeRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(BlockStreamServiceServer).Subscribe(m, &grpc.GenericServerStream[SubscribeRequest, BlockEvent]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type BlockStreamService_SubscribeServer = grpc.ServerStreamingServer[BlockEvent]
+
+// BlockStreamService_ServiceDesc is the grpc.ServiceDesc for BlockStreamService service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var BlockStreamService_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "getoptimum.optimum_gateway.service.stream.v1.BlockStreamService",
+	HandlerType: (*BlockStreamServiceServer)(nil),
+	Methods:     []grpc.MethodDesc{},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "Subscribe",
+			Handler:       _BlockStreamService_Subscribe_Handler,
+			ServerStreams: true,
+		},
+	},
+	Metadata: "getoptimum/optimum_gateway/service/stream/v1/stream.proto",
+}
diff --git a/pkg/service/stream/ws.go b/pkg/service/stream/ws.go
index 730cc2b..ef038e0 100644
--- a/pkg/service/stream/ws.go
+++ b/pkg/service/stream/ws.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"net/http"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -40,6 +39,9 @@ type Config struct {
 	MaxConns       int
 	MaxConnsPerSub int
 	BufferSize     int
+	// Limiter is shared across transports so caps stay global; withDefaults
+	// creates one if nil.
+	Limiter *ConnLimiter
 }
 
 // Server exposes streamhub over WebSocket on its own listener, keeping the
@@ -49,32 +51,21 @@ type Server struct {
 	auth     ConsumerAuthenticator
 	cfg      Config
 	log      logger.AppLogger
+	limiter  *ConnLimiter
 	upgrader websocket.Upgrader
 	httpSrv  *http.Server
-
-	mu     sync.Mutex
-	conns  int
-	perSub map[string]int
 }
 
 // NewServer builds the consumer WebSocket server. It does not start listening;
 // call Run.
 func NewServer(hub *streamhub.Service, auth ConsumerAuthenticator, cfg Config, log logger.AppLogger) *Server {
-	if cfg.MaxConns <= 0 {
-		cfg.MaxConns = 256
-	}
-	if cfg.MaxConnsPerSub <= 0 {
-		cfg.MaxConnsPerSub = 8
-	}
-	if cfg.BufferSize <= 0 {
-		cfg.BufferSize = streamhub.DefaultBufferSize
-	}
+	cfg = withDefaults(cfg)
 	s := &Server{
-		hub:    hub,
-		auth:   auth,
-		cfg:    cfg,
-		log:    log.With(logger.WithService("stream-ws")),
-		perSub: make(map[string]int),
+		hub:     hub,
+		auth:    auth,
+		cfg:     cfg,
+		log:     log.With(logger.WithService("stream-ws")),
+		limiter: cfg.Limiter,
 		upgrader: websocket.Upgrader{
 			// JWT gates access (not Origin; TLS/proxy is the exposure control).
 			// Offering only the marker means gorilla never selects bearer.<jwt>.
@@ -105,15 +96,12 @@ func (s *Server) Stop(ctx context.Context) error {
 }
 
 func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
-	mode := r.URL.Query().Get("mode")
-	if mode == "" {
-		mode = modeMetadata
-	}
-	if mode != modeMetadata && mode != modeRaw {
+	mode, ok := normalizeMode(r.URL.Query().Get("mode"))
+	if !ok {
 		http.Error(w, "invalid mode", http.StatusBadRequest)
 		return
 	}
-	if t := r.URL.Query().Get("topics"); t != "" && t != defaultTopic {
+	if !topicsOK(r.URL.Query().Get("topics")) {
 		http.Error(w, "unsupported topics", http.StatusBadRequest)
 		return
 	}
@@ -128,7 +116,7 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
 
 	// Enforce caps before the upgrade too, so a rejected connection allocates
 	// no subscriber.
-	if !s.acquire(subject) {
+	if !s.limiter.acquire(subject) {
 		http.Error(w, "too many connections", http.StatusServiceUnavailable)
 		return
 	}
@@ -136,7 +124,7 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
 	conn, err := s.upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		// Upgrade already wrote the error response.
-		s.release(subject)
+		s.limiter.release(subject)
 		return
 	}
 
@@ -150,7 +138,7 @@ func (s *Server) serve(conn *websocket.Conn, sub *streamhub.Subscription, subjec
 	defer func() {
 		_ = conn.Close()
 		sub.Close()
-		s.release(subject)
+		s.limiter.release(subject)
 	}()
 
 	conn.SetReadLimit(maxReadBytes)
@@ -232,30 +220,6 @@ func (s *Server) writeJSON(conn *websocket.Conn, v any) error {
 	return conn.WriteJSON(v)
 }
 
-// acquire admits a connection if the global and per-subject caps allow it.
-func (s *Server) acquire(subject string) bool {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if s.conns >= s.cfg.MaxConns || s.perSub[subject] >= s.cfg.MaxConnsPerSub {
-		return false
-	}
-	s.conns++
-	s.perSub[subject]++
-	telemetry.IncStreamConnections()
-	return true
-}
-
-func (s *Server) release(subject string) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.conns--
-	s.perSub[subject]--
-	if s.perSub[subject] <= 0 {
-		delete(s.perSub, subject)
-	}
-	telemetry.DecStreamConnections()
-}
-
 // bearerToken reads the consumer JWT from the Authorization header, or from the
 // "bearer.<jwt>" Sec-WebSocket-Protocol offer that browsers must use instead.
 func bearerToken(r *http.Request) string {
diff --git a/pkg/service/stream/ws_test.go b/pkg/service/stream/ws_test.go
index 33d3b98..b1606b1 100644
--- a/pkg/service/stream/ws_test.go
+++ b/pkg/service/stream/ws_test.go
@@ -20,19 +20,23 @@ import (
 	"github.com/getoptimum/optimum-gateway/pkg/test_utils"
 )
 
-// newWSTestServer wires a real Server behind httptest so tests exercise the full
-// upgrade + auth path. requireAuth=false uses the allow-all (loopback) backend.
+// testAuth builds the consumer authenticator used by both WS and gRPC tests.
+// requireAuth=false uses the allow-all (loopback) backend.
+func testAuth(t *testing.T, requireAuth bool) (ConsumerAuthenticator, *test_utils.AuthTestRig) {
+	t.Helper()
+	rig := test_utils.NewAuthTestRig(t)
+	if !requireAuth {
+		return NewConsumerAuthenticator(nil, false), rig
+	}
+	m, err := auth_token.New(t.Context(), logger.NewAppSLogger(logger.Debug), rig.AppCfg(t))
+	require.NoError(t, err)
+	return NewConsumerAuthenticator(m, true), rig
+}
+
 func newWSTestServer(t *testing.T, cfg Config, requireAuth bool) (ts *httptest.Server, s *Server, hub *streamhub.Service, rig *test_utils.AuthTestRig) {
 	t.Helper()
-	rig = test_utils.NewAuthTestRig(t)
 	var authenticator ConsumerAuthenticator
-	if requireAuth {
-		m, err := auth_token.New(t.Context(), logger.NewAppSLogger(logger.Debug), rig.AppCfg(t))
-		require.NoError(t, err)
-		authenticator = NewConsumerAuthenticator(m, true)
-	} else {
-		authenticator = NewConsumerAuthenticator(nil, false)
-	}
+	authenticator, rig = testAuth(t, requireAuth)
 	hub = streamhub.New()
 	s = NewServer(hub, authenticator, cfg, logger.NewAppSLogger(logger.Debug))
 	ts = httptest.NewServer(s.httpSrv.Handler)
@@ -230,8 +234,8 @@ func TestWS_CleanupOnClose(t *testing.T) {
 	// release the cap slot, so nothing leaks.
 	waitSubscribed(t, hub, 0)
 	require.Eventually(t, func() bool {
-		s.mu.Lock()
-		defer s.mu.Unlock()
-		return s.conns == 0 && len(s.perSub) == 0
+		s.limiter.mu.Lock()
+		defer s.limiter.mu.Unlock()
+		return s.limiter.conns == 0 && len(s.limiter.perSub) == 0
 	}, 2*time.Second, 10*time.Millisecond)
 }
diff --git a/proto/getoptimum/optimum_gateway/service/stream/v1/stream.proto b/proto/getoptimum/optimum_gateway/service/stream/v1/stream.proto
new file mode 100644
index 0000000..17d359b
--- /dev/null
+++ b/proto/getoptimum/optimum_gateway/service/stream/v1/stream.proto
@@ -0,0 +1,47 @@
+syntax = "proto3";
+
+package getoptimum.optimum_gateway.service.stream.v1;
+
+option go_package = "github.com/getoptimum/optimum-gateway/pkg/service/stream/v1";
+
+// BlockStreamService streams decoded beacon-block observations to consumers (ADR-0011).
+service BlockStreamService {
+  // Subscribe opens a read-only server stream of block observations.
+  rpc Subscribe(SubscribeRequest) returns (stream BlockEvent);
+}
+
+// SubscribeRequest selects the payload mode and topics for a subscription.
+message SubscribeRequest {
+  string mode = 1; // "metadata" (default) or "raw"
+  repeated string topics = 2; // validated only; v1 has one topic, so nothing is filtered
+}
+
+// BlockEvent is one frame: a block observation or a lag signal (ADR-0011).
+message BlockEvent {
+  // frame tells an observation apart from a control signal.
+  oneof frame {
+    Block block = 1;
+    Lagged lagged = 2;
+  }
+}
+
+// Block is one block observation (fields mirror the streamhub hub type).
+message Block {
+  uint64 slot = 1;
+  uint64 proposer_index = 2;
+  bytes parent_root = 3;
+  bytes state_root = 4;
+  uint64 block_size_bytes = 5;
+  string topic = 6;
+  string source = 7;
+  int64 received_at_ms = 8;
+  string gateway_id = 9;
+  string fork_digest = 10;
+  bool stale = 11;
+  bytes raw = 12; // present only in raw mode
+}
+
+// Lagged reports the cumulative drop count after a buffer overflow.
+message Lagged {
+  uint64 dropped = 1;
+}