From 449ef200e3b379b95d18a812a5c63de1189a40e1 Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Wed, 22 Jul 2026 13:00:07 -0700
Subject: [PATCH 1/9] feat(subscriber): read engine load via GetLoads gRPC (alt
 to HTTP scrape)

---
 cmd/kvevent-subscriber/main.go                |  50 ++-
 pkg/adapters/engine/grpc_loads_scraper.go     | 119 ++++++
 .../engine/grpc_loads_scraper_test.go         | 142 +++++++
 .../engine/vllmengine/vllm_engine.pb.go       | 398 ++++++++++++++++++
 .../engine/vllmengine/vllm_engine.proto       |  49 +++
 .../engine/vllmengine/vllm_engine_grpc.pb.go  | 135 ++++++
 6 files changed, 880 insertions(+), 13 deletions(-)
 create mode 100644 pkg/adapters/engine/grpc_loads_scraper.go
 create mode 100644 pkg/adapters/engine/grpc_loads_scraper_test.go
 create mode 100644 pkg/adapters/engine/vllmengine/vllm_engine.pb.go
 create mode 100644 pkg/adapters/engine/vllmengine/vllm_engine.proto
 create mode 100644 pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go

diff --git a/cmd/kvevent-subscriber/main.go b/cmd/kvevent-subscriber/main.go
index 19c6661b..b57c826c 100644
--- a/cmd/kvevent-subscriber/main.go
+++ b/cmd/kvevent-subscriber/main.go
@@ -44,6 +44,7 @@ func main() {
 		scheme             = flag.String("hash-scheme", "vllm", "engine prefix-hash scheme (required, non-empty)")
 		window             = flag.Duration("window", 100*time.Millisecond, "add-batching/debounce flush window")
 		metricsURL         = flag.String("engine-metrics-url", "http://127.0.0.1:8000/metrics", "engine Prometheus /metrics URL")
+		loadsGrpc          = flag.String("engine-loads-grpc", "", "engine VllmEngine gRPC address (host:port) to read load via the GetLoads RPC instead of scraping --engine-metrics-url. Preferred for SMG gRPC engines, which expose no HTTP /metrics. Empty = use the HTTP scrape.")
 		statsInterval      = flag.Duration("stats-interval", 10*time.Second, "ReplicaStats scrape/emit cadence")
 		cacheSizeBytes     = flag.Int64("engine-cache-size-bytes", 0, "engine total KV-cache capacity in bytes (multiplied by usage_perc to derive cacheMemoryBytes; 0 emits cacheMemoryBytes=0)")
 		ceiling            = flag.Int("max-concurrency-ceiling", 256, "denominator for the pressure proxy = clamp01((num_requests_running+num_requests_waiting)/ceiling)")
@@ -91,21 +92,44 @@ func main() {
 		engine.WithIgnoreBlockRemoved(*ignoreBlockRemoved))
 	sub := engine.NewSubscriber(*endpoint, *topic, engine.WithSubscriberLogger(logger))
 
-	scraper := engine.NewMetricsScraper(
-		&http.Client{Timeout: 5 * time.Second},
-		engine.ScraperConfig{
-			URL:                   *metricsURL,
-			Tier:                  tier,
-			ModelLabel:            *engineModel,
+	// Load source: gRPC GetLoads (preferred for SMG gRPC engines, which expose no
+	// HTTP /metrics) when --engine-loads-grpc is set, else the HTTP /metrics scrape.
+	// Both satisfy the same statsScraper, so the StatsReporter is identical.
+	var statsReporter *engine.StatsReporter
+	if *loadsGrpc != "" {
+		gs, gerr := engine.NewGrpcLoadsScraper(engine.GrpcLoadsScraperConfig{
+			Addr:                  *loadsGrpc,
 			CacheSizeBytes:        *cacheSizeBytes,
 			MaxConcurrencyCeiling: *ceiling,
-		},
-		logger,
-	)
-	statsReporter := engine.NewStatsReporter(client, scraper, cfg,
-		engine.WithStatsInterval(*statsInterval),
-		engine.WithStatsLogger(logger),
-	)
+		})
+		if gerr != nil {
+			logger.Error("grpc loads scraper", "addr", *loadsGrpc, "err", gerr)
+			os.Exit(1)
+		}
+		defer func() { _ = gs.Close() }()
+		logger.Info("engine load source: GetLoads gRPC", "addr", *loadsGrpc)
+		statsReporter = engine.NewStatsReporter(client, gs, cfg,
+			engine.WithStatsInterval(*statsInterval),
+			engine.WithStatsLogger(logger),
+		)
+	} else {
+		scraper := engine.NewMetricsScraper(
+			&http.Client{Timeout: 5 * time.Second},
+			engine.ScraperConfig{
+				URL:                   *metricsURL,
+				Tier:                  tier,
+				ModelLabel:            *engineModel,
+				CacheSizeBytes:        *cacheSizeBytes,
+				MaxConcurrencyCeiling: *ceiling,
+			},
+			logger,
+		)
+		logger.Info("engine load source: HTTP /metrics scrape", "url", *metricsURL)
+		statsReporter = engine.NewStatsReporter(client, scraper, cfg,
+			engine.WithStatsInterval(*statsInterval),
+			engine.WithStatsLogger(logger),
+		)
+	}
 
 	out := make(chan *engine.EventBatch, 256)
 
diff --git a/pkg/adapters/engine/grpc_loads_scraper.go b/pkg/adapters/engine/grpc_loads_scraper.go
new file mode 100644
index 00000000..9ed2661a
--- /dev/null
+++ b/pkg/adapters/engine/grpc_loads_scraper.go
@@ -0,0 +1,119 @@
+package engine
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	vpb "github.com/cachebox-project/inference-cache/pkg/adapters/engine/vllmengine"
+	icpb "github.com/cachebox-project/inference-cache/pkg/server/proto/inferencecache/v1alpha1"
+)
+
+// GrpcLoadsScraperConfig tunes the gRPC GetLoads scraper. Addr is required.
+type GrpcLoadsScraperConfig struct {
+	// Addr is the engine's VllmEngine gRPC address (host:port), e.g.
+	// 127.0.0.1:50051.
+	Addr string
+	// CacheSizeBytes is the engine's total KV-cache capacity, used to map
+	// token_usage (0..1) to cache_memory_bytes. When zero, cache_memory_bytes is
+	// emitted as 0 (an honest "unknown" rather than a fabricated number).
+	CacheSizeBytes int64
+	// MaxConcurrencyCeiling is the pressure denominator:
+	//   pressure = clamp01((num_running_reqs + num_waiting_reqs) / ceiling).
+	// 0 disables pressure (it stays 0).
+	MaxConcurrencyCeiling int
+	// Timeout bounds each GetLoads call; defaults to defaultScrapeTimeout when <= 0.
+	Timeout time.Duration
+}
+
+// GrpcLoadsScraper reads engine load via the SMG engine's GetLoads gRPC RPC and
+// projects it into a ReplicaStats — the gRPC-native alternative to scraping the
+// engine's HTTP /metrics. An SMG gRPC engine (vllm.entrypoints.grpc_server)
+// exposes no HTTP metrics endpoint; live load is available only over GetLoads.
+// It implements the same statsScraper interface as MetricsScraper, so the
+// StatsReporter uses whichever is configured, unchanged.
+//
+// Note: GetLoads carries no external-tier (T2/LMCache) token counters, so
+// T2HitTokens/T2QueryTokens are left 0 here — those remain HTTP-/metrics-only.
+// The load signals the ranker actually uses (pressure, cache-usage, hit-rate)
+// are all provided.
+type GrpcLoadsScraper struct {
+	client vpb.VllmEngineClient
+	cfg    GrpcLoadsScraperConfig
+	closer func() error
+}
+
+// NewGrpcLoadsScraper dials the engine (lazily — grpc.NewClient does not connect
+// until the first RPC, so a down engine doesn't fail construction) and returns a
+// scraper. Call Close to release the connection.
+func NewGrpcLoadsScraper(cfg GrpcLoadsScraperConfig) (*GrpcLoadsScraper, error) {
+	if cfg.Addr == "" {
+		return nil, fmt.Errorf("grpc loads scraper: Addr is required")
+	}
+	if cfg.Timeout <= 0 {
+		cfg.Timeout = defaultScrapeTimeout
+	}
+	conn, err := grpc.NewClient(cfg.Addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, fmt.Errorf("grpc loads scraper: dial %s: %w", cfg.Addr, err)
+	}
+	return &GrpcLoadsScraper{client: vpb.NewVllmEngineClient(conn), cfg: cfg, closer: conn.Close}, nil
+}
+
+// newGrpcLoadsScraperWithClient injects a client (tests) and never owns a conn.
+func newGrpcLoadsScraperWithClient(c vpb.VllmEngineClient, cfg GrpcLoadsScraperConfig) *GrpcLoadsScraper {
+	if cfg.Timeout <= 0 {
+		cfg.Timeout = defaultScrapeTimeout
+	}
+	return &GrpcLoadsScraper{client: c, cfg: cfg, closer: func() error { return nil }}
+}
+
+// Close releases the underlying gRPC connection.
+func (s *GrpcLoadsScraper) Close() error { return s.closer() }
+
+// Scrape calls GetLoads once and projects the per-DP-rank SchedulerLoad into a
+// ReplicaStats. On error a zero-valued *icpb.ReplicaStats is returned alongside
+// the error — the StatsReporter logs and skips the tick, same as the HTTP path.
+func (s *GrpcLoadsScraper) Scrape(ctx context.Context) (*icpb.ReplicaStats, error) {
+	rctx, cancel := context.WithTimeout(ctx, s.cfg.Timeout)
+	defer cancel()
+
+	resp, err := s.client.GetLoads(rctx, &vpb.GetLoadsRequest{})
+	if err != nil {
+		return &icpb.ReplicaStats{}, fmt.Errorf("getloads %s: %w", s.cfg.Addr, err)
+	}
+
+	// Aggregate across DP ranks: request counts SUM (total in-flight across the
+	// engine), KV-cache usage is the MAX (worst-case pressure proxy), hit-rate is
+	// the mean of ranks that report one.
+	var running, waiting int64
+	var usage, hitRateSum float64
+	var hitRateN int
+	for _, l := range resp.GetLoads() {
+		running += int64(l.GetNumRunningReqs())
+		waiting += int64(l.GetNumWaitingReqs())
+		if u := l.GetTokenUsage(); u > usage {
+			usage = u
+		}
+		hitRateSum += l.GetCacheHitRate()
+		hitRateN++
+	}
+	hitRate := 0.0
+	if hitRateN > 0 {
+		hitRate = hitRateSum / float64(hitRateN)
+	}
+
+	pressure := pressureFrom(float64(running+waiting), s.cfg.MaxConcurrencyCeiling)
+	var cacheBytes int64
+	if s.cfg.CacheSizeBytes > 0 {
+		cacheBytes = int64(usage * float64(s.cfg.CacheSizeBytes))
+	}
+	return &icpb.ReplicaStats{
+		CacheMemoryBytes: cacheBytes,
+		HitRate:          float32(hitRate),
+		Pressure:         float32(pressure),
+	}, nil
+}
diff --git a/pkg/adapters/engine/grpc_loads_scraper_test.go b/pkg/adapters/engine/grpc_loads_scraper_test.go
new file mode 100644
index 00000000..d2efeaee
--- /dev/null
+++ b/pkg/adapters/engine/grpc_loads_scraper_test.go
@@ -0,0 +1,142 @@
+package engine
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/test/bufconn"
+
+	vpb "github.com/cachebox-project/inference-cache/pkg/adapters/engine/vllmengine"
+)
+
+// fakeLoadsClient injects a canned GetLoads response/error for mapping tests.
+type fakeLoadsClient struct {
+	resp *vpb.GetLoadsResponse
+	err  error
+}
+
+func (f *fakeLoadsClient) GetLoads(_ context.Context, _ *vpb.GetLoadsRequest, _ ...grpc.CallOption) (*vpb.GetLoadsResponse, error) {
+	return f.resp, f.err
+}
+
+func TestGrpcLoadsScraperMapsSingleRank(t *testing.T) {
+	resp := &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
+		{NumRunningReqs: 6, NumWaitingReqs: 2, TokenUsage: 0.5, CacheHitRate: 0.8},
+	}}
+	s := newGrpcLoadsScraperWithClient(&fakeLoadsClient{resp: resp},
+		GrpcLoadsScraperConfig{CacheSizeBytes: 1 << 30, MaxConcurrencyCeiling: 16})
+
+	st, err := s.Scrape(context.Background())
+	if err != nil {
+		t.Fatalf("Scrape: %v", err)
+	}
+	if got, want := st.GetPressure(), float32(0.5); got != want { // (6+2)/16
+		t.Errorf("pressure = %v, want %v", got, want)
+	}
+	if got, want := st.GetCacheMemoryBytes(), int64(0.5*float64(1<<30)); got != want {
+		t.Errorf("cache_memory_bytes = %d, want %d", got, want)
+	}
+	if got, want := st.GetHitRate(), float32(0.8); got != want {
+		t.Errorf("hit_rate = %v, want %v", got, want)
+	}
+}
+
+func TestGrpcLoadsScraperAggregatesRanks(t *testing.T) {
+	// two DP ranks: running/waiting SUM, usage MAX, hit-rate MEAN
+	resp := &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
+		{NumRunningReqs: 2, NumWaitingReqs: 1, TokenUsage: 0.3, CacheHitRate: 0.5},
+		{NumRunningReqs: 5, NumWaitingReqs: 4, TokenUsage: 0.7, CacheHitRate: 0.9},
+	}}
+	s := newGrpcLoadsScraperWithClient(&fakeLoadsClient{resp: resp},
+		GrpcLoadsScraperConfig{CacheSizeBytes: 1000, MaxConcurrencyCeiling: 24})
+
+	st, err := s.Scrape(context.Background())
+	if err != nil {
+		t.Fatalf("Scrape: %v", err)
+	}
+	if got, want := st.GetPressure(), float32(12.0/24.0); got != want { // (7+5)/24
+		t.Errorf("pressure = %v, want %v", got, want)
+	}
+	if got, want := st.GetCacheMemoryBytes(), int64(0.7*1000); got != want { // max usage
+		t.Errorf("cache_memory_bytes = %d, want %d", got, want)
+	}
+	if got, want := st.GetHitRate(), float32(0.7); got != want { // mean(0.5,0.9)
+		t.Errorf("hit_rate = %v, want %v", got, want)
+	}
+}
+
+func TestGrpcLoadsScraperZeroConfigIsHonest(t *testing.T) {
+	// CacheSizeBytes=0 -> 0 bytes (not fabricated); ceiling=0 -> pressure 0
+	resp := &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
+		{NumRunningReqs: 9, NumWaitingReqs: 9, TokenUsage: 0.9},
+	}}
+	s := newGrpcLoadsScraperWithClient(&fakeLoadsClient{resp: resp}, GrpcLoadsScraperConfig{})
+	st, err := s.Scrape(context.Background())
+	if err != nil {
+		t.Fatalf("Scrape: %v", err)
+	}
+	if st.GetCacheMemoryBytes() != 0 || st.GetPressure() != 0 {
+		t.Errorf("want zeroed cache_memory_bytes/pressure, got bytes=%d pressure=%v",
+			st.GetCacheMemoryBytes(), st.GetPressure())
+	}
+}
+
+func TestGrpcLoadsScraperErrorIsSurfaced(t *testing.T) {
+	s := newGrpcLoadsScraperWithClient(&fakeLoadsClient{err: errors.New("unavailable")},
+		GrpcLoadsScraperConfig{Addr: "x:1"})
+	st, err := s.Scrape(context.Background())
+	if err == nil {
+		t.Fatal("want error, got nil")
+	}
+	if st == nil {
+		t.Fatal("want zero ReplicaStats on error, got nil")
+	}
+}
+
+// mockEngineServer implements the generated VllmEngineServer for a bufconn
+// end-to-end test through the real generated stubs.
+type mockEngineServer struct {
+	vpb.UnimplementedVllmEngineServer
+	resp *vpb.GetLoadsResponse
+}
+
+func (m *mockEngineServer) GetLoads(context.Context, *vpb.GetLoadsRequest) (*vpb.GetLoadsResponse, error) {
+	return m.resp, nil
+}
+
+func TestGrpcLoadsScraperEndToEndOverBufconn(t *testing.T) {
+	lis := bufconn.Listen(1 << 20)
+	srv := grpc.NewServer()
+	vpb.RegisterVllmEngineServer(srv, &mockEngineServer{
+		resp: &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
+			{NumRunningReqs: 3, NumWaitingReqs: 1, TokenUsage: 0.25, CacheHitRate: 0.6},
+		}},
+	})
+	go func() { _ = srv.Serve(lis) }()
+	defer srv.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()))
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer func() { _ = conn.Close() }()
+
+	s := newGrpcLoadsScraperWithClient(vpb.NewVllmEngineClient(conn),
+		GrpcLoadsScraperConfig{CacheSizeBytes: 1 << 30, MaxConcurrencyCeiling: 8})
+	st, err := s.Scrape(context.Background())
+	if err != nil {
+		t.Fatalf("Scrape: %v", err)
+	}
+	if got, want := st.GetPressure(), float32(0.5); got != want { // (3+1)/8
+		t.Errorf("pressure = %v, want %v", got, want)
+	}
+	if st.GetCacheMemoryBytes() != int64(0.25*float64(1<<30)) {
+		t.Errorf("cache_memory_bytes = %d", st.GetCacheMemoryBytes())
+	}
+}
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
new file mode 100644
index 00000000..abf05374
--- /dev/null
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
@@ -0,0 +1,398 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.36.5
+// 	protoc        v7.34.1
+// source: vllm_engine.proto
+
+// Minimal vendored subset of the SMG vLLM engine gRPC contract
+// (machxai/smg: crates/grpc_client/proto/vllm_engine.proto) — ONLY the GetLoads
+// RPC and its messages. The kvevent-subscriber uses this to read engine load
+// over gRPC as an alternative to scraping HTTP /metrics (which an SMG gRPC
+// engine does not expose).
+//
+// The package name MUST stay `vllm.grpc.engine` to match the upstream service
+// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
+// field numbers mirror upstream. Kept out of the `proto/` module on purpose —
+// it is a vendored external contract, not part of IC's own gRPC API, so it is
+// not subject to IC's buf-lint conventions or the proto drift check.
+
+package vllmengine
+
+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)
+)
+
+type GetLoadsRequest struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	DpRank        *int32                 `protobuf:"varint,1,opt,name=dp_rank,json=dpRank,proto3,oneof" json:"dp_rank,omitempty"`
+	Include       []string               `protobuf:"bytes,2,rep,name=include,proto3" json:"include,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *GetLoadsRequest) Reset() {
+	*x = GetLoadsRequest{}
+	mi := &file_vllm_engine_proto_msgTypes[0]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *GetLoadsRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetLoadsRequest) ProtoMessage() {}
+
+func (x *GetLoadsRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_vllm_engine_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 GetLoadsRequest.ProtoReflect.Descriptor instead.
+func (*GetLoadsRequest) Descriptor() ([]byte, []int) {
+	return file_vllm_engine_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *GetLoadsRequest) GetDpRank() int32 {
+	if x != nil && x.DpRank != nil {
+		return *x.DpRank
+	}
+	return 0
+}
+
+func (x *GetLoadsRequest) GetInclude() []string {
+	if x != nil {
+		return x.Include
+	}
+	return nil
+}
+
+type GetLoadsResponse struct {
+	state         protoimpl.MessageState `protogen:"open.v1"`
+	Timestamp     string                 `protobuf:"bytes,1,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
+	Version       string                 `protobuf:"bytes,2,opt,name=version,proto3" json:"version,omitempty"`
+	DpRankCount   int32                  `protobuf:"varint,3,opt,name=dp_rank_count,json=dpRankCount,proto3" json:"dp_rank_count,omitempty"`
+	Loads         []*SchedulerLoad       `protobuf:"bytes,4,rep,name=loads,proto3" json:"loads,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *GetLoadsResponse) Reset() {
+	*x = GetLoadsResponse{}
+	mi := &file_vllm_engine_proto_msgTypes[1]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *GetLoadsResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*GetLoadsResponse) ProtoMessage() {}
+
+func (x *GetLoadsResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_vllm_engine_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 GetLoadsResponse.ProtoReflect.Descriptor instead.
+func (*GetLoadsResponse) Descriptor() ([]byte, []int) {
+	return file_vllm_engine_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *GetLoadsResponse) GetTimestamp() string {
+	if x != nil {
+		return x.Timestamp
+	}
+	return ""
+}
+
+func (x *GetLoadsResponse) GetVersion() string {
+	if x != nil {
+		return x.Version
+	}
+	return ""
+}
+
+func (x *GetLoadsResponse) GetDpRankCount() int32 {
+	if x != nil {
+		return x.DpRankCount
+	}
+	return 0
+}
+
+func (x *GetLoadsResponse) GetLoads() []*SchedulerLoad {
+	if x != nil {
+		return x.Loads
+	}
+	return nil
+}
+
+// Per-DP-rank scheduler load snapshot. Mirrors upstream field numbers; unset
+// fields are 0.
+type SchedulerLoad struct {
+	state              protoimpl.MessageState `protogen:"open.v1"`
+	DpRank             int32                  `protobuf:"varint,1,opt,name=dp_rank,json=dpRank,proto3" json:"dp_rank,omitempty"`
+	NumRunningReqs     int32                  `protobuf:"varint,2,opt,name=num_running_reqs,json=numRunningReqs,proto3" json:"num_running_reqs,omitempty"`
+	NumWaitingReqs     int32                  `protobuf:"varint,3,opt,name=num_waiting_reqs,json=numWaitingReqs,proto3" json:"num_waiting_reqs,omitempty"`
+	NumTotalReqs       int32                  `protobuf:"varint,4,opt,name=num_total_reqs,json=numTotalReqs,proto3" json:"num_total_reqs,omitempty"`
+	NumUsedTokens      int32                  `protobuf:"varint,5,opt,name=num_used_tokens,json=numUsedTokens,proto3" json:"num_used_tokens,omitempty"`
+	MaxTotalNumTokens  int32                  `protobuf:"varint,6,opt,name=max_total_num_tokens,json=maxTotalNumTokens,proto3" json:"max_total_num_tokens,omitempty"`
+	TokenUsage         float64                `protobuf:"fixed64,7,opt,name=token_usage,json=tokenUsage,proto3" json:"token_usage,omitempty"` // KV-cache utilization in [0,1)
+	GenThroughput      float64                `protobuf:"fixed64,8,opt,name=gen_throughput,json=genThroughput,proto3" json:"gen_throughput,omitempty"`
+	CacheHitRate       float64                `protobuf:"fixed64,9,opt,name=cache_hit_rate,json=cacheHitRate,proto3" json:"cache_hit_rate,omitempty"`
+	Utilization        float64                `protobuf:"fixed64,10,opt,name=utilization,proto3" json:"utilization,omitempty"`
+	MaxRunningRequests int32                  `protobuf:"varint,11,opt,name=max_running_requests,json=maxRunningRequests,proto3" json:"max_running_requests,omitempty"`
+	unknownFields      protoimpl.UnknownFields
+	sizeCache          protoimpl.SizeCache
+}
+
+func (x *SchedulerLoad) Reset() {
+	*x = SchedulerLoad{}
+	mi := &file_vllm_engine_proto_msgTypes[2]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *SchedulerLoad) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SchedulerLoad) ProtoMessage() {}
+
+func (x *SchedulerLoad) ProtoReflect() protoreflect.Message {
+	mi := &file_vllm_engine_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 SchedulerLoad.ProtoReflect.Descriptor instead.
+func (*SchedulerLoad) Descriptor() ([]byte, []int) {
+	return file_vllm_engine_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *SchedulerLoad) GetDpRank() int32 {
+	if x != nil {
+		return x.DpRank
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetNumRunningReqs() int32 {
+	if x != nil {
+		return x.NumRunningReqs
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetNumWaitingReqs() int32 {
+	if x != nil {
+		return x.NumWaitingReqs
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetNumTotalReqs() int32 {
+	if x != nil {
+		return x.NumTotalReqs
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetNumUsedTokens() int32 {
+	if x != nil {
+		return x.NumUsedTokens
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetMaxTotalNumTokens() int32 {
+	if x != nil {
+		return x.MaxTotalNumTokens
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetTokenUsage() float64 {
+	if x != nil {
+		return x.TokenUsage
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetGenThroughput() float64 {
+	if x != nil {
+		return x.GenThroughput
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetCacheHitRate() float64 {
+	if x != nil {
+		return x.CacheHitRate
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetUtilization() float64 {
+	if x != nil {
+		return x.Utilization
+	}
+	return 0
+}
+
+func (x *SchedulerLoad) GetMaxRunningRequests() int32 {
+	if x != nil {
+		return x.MaxRunningRequests
+	}
+	return 0
+}
+
+var File_vllm_engine_proto protoreflect.FileDescriptor
+
+var file_vllm_engine_proto_rawDesc = string([]byte{
+	0x0a, 0x11, 0x76, 0x6c, 0x6c, 0x6d, 0x5f, 0x65, 0x6e, 0x67, 0x69, 0x6e, 0x65, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x12, 0x10, 0x76, 0x6c, 0x6c, 0x6d, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x65,
+	0x6e, 0x67, 0x69, 0x6e, 0x65, 0x22, 0x55, 0x0a, 0x0f, 0x47, 0x65, 0x74, 0x4c, 0x6f, 0x61, 0x64,
+	0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x07, 0x64, 0x70, 0x5f, 0x72,
+	0x61, 0x6e, 0x6b, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x48, 0x00, 0x52, 0x06, 0x64, 0x70, 0x52,
+	0x61, 0x6e, 0x6b, 0x88, 0x01, 0x01, 0x12, 0x18, 0x0a, 0x07, 0x69, 0x6e, 0x63, 0x6c, 0x75, 0x64,
+	0x65, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x07, 0x69, 0x6e, 0x63, 0x6c, 0x75, 0x64, 0x65,
+	0x42, 0x0a, 0x0a, 0x08, 0x5f, 0x64, 0x70, 0x5f, 0x72, 0x61, 0x6e, 0x6b, 0x22, 0xa5, 0x01, 0x0a,
+	0x10, 0x47, 0x65, 0x74, 0x4c, 0x6f, 0x61, 0x64, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x12, 0x1c, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12,
+	0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x22, 0x0a, 0x0d, 0x64, 0x70, 0x5f,
+	0x72, 0x61, 0x6e, 0x6b, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05,
+	0x52, 0x0b, 0x64, 0x70, 0x52, 0x61, 0x6e, 0x6b, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x35, 0x0a,
+	0x05, 0x6c, 0x6f, 0x61, 0x64, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1f, 0x2e, 0x76,
+	0x6c, 0x6c, 0x6d, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x65, 0x6e, 0x67, 0x69, 0x6e, 0x65, 0x2e,
+	0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x4c, 0x6f, 0x61, 0x64, 0x52, 0x05, 0x6c,
+	0x6f, 0x61, 0x64, 0x73, 0x22, 0xbd, 0x03, 0x0a, 0x0d, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c,
+	0x65, 0x72, 0x4c, 0x6f, 0x61, 0x64, 0x12, 0x17, 0x0a, 0x07, 0x64, 0x70, 0x5f, 0x72, 0x61, 0x6e,
+	0x6b, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x64, 0x70, 0x52, 0x61, 0x6e, 0x6b, 0x12,
+	0x28, 0x0a, 0x10, 0x6e, 0x75, 0x6d, 0x5f, 0x72, 0x75, 0x6e, 0x6e, 0x69, 0x6e, 0x67, 0x5f, 0x72,
+	0x65, 0x71, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0e, 0x6e, 0x75, 0x6d, 0x52, 0x75,
+	0x6e, 0x6e, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x73, 0x12, 0x28, 0x0a, 0x10, 0x6e, 0x75, 0x6d,
+	0x5f, 0x77, 0x61, 0x69, 0x74, 0x69, 0x6e, 0x67, 0x5f, 0x72, 0x65, 0x71, 0x73, 0x18, 0x03, 0x20,
+	0x01, 0x28, 0x05, 0x52, 0x0e, 0x6e, 0x75, 0x6d, 0x57, 0x61, 0x69, 0x74, 0x69, 0x6e, 0x67, 0x52,
+	0x65, 0x71, 0x73, 0x12, 0x24, 0x0a, 0x0e, 0x6e, 0x75, 0x6d, 0x5f, 0x74, 0x6f, 0x74, 0x61, 0x6c,
+	0x5f, 0x72, 0x65, 0x71, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0c, 0x6e, 0x75, 0x6d,
+	0x54, 0x6f, 0x74, 0x61, 0x6c, 0x52, 0x65, 0x71, 0x73, 0x12, 0x26, 0x0a, 0x0f, 0x6e, 0x75, 0x6d,
+	0x5f, 0x75, 0x73, 0x65, 0x64, 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x73, 0x18, 0x05, 0x20, 0x01,
+	0x28, 0x05, 0x52, 0x0d, 0x6e, 0x75, 0x6d, 0x55, 0x73, 0x65, 0x64, 0x54, 0x6f, 0x6b, 0x65, 0x6e,
+	0x73, 0x12, 0x2f, 0x0a, 0x14, 0x6d, 0x61, 0x78, 0x5f, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x5f, 0x6e,
+	0x75, 0x6d, 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x73, 0x18, 0x06, 0x20, 0x01, 0x28, 0x05, 0x52,
+	0x11, 0x6d, 0x61, 0x78, 0x54, 0x6f, 0x74, 0x61, 0x6c, 0x4e, 0x75, 0x6d, 0x54, 0x6f, 0x6b, 0x65,
+	0x6e, 0x73, 0x12, 0x1f, 0x0a, 0x0b, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x5f, 0x75, 0x73, 0x61, 0x67,
+	0x65, 0x18, 0x07, 0x20, 0x01, 0x28, 0x01, 0x52, 0x0a, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x55, 0x73,
+	0x61, 0x67, 0x65, 0x12, 0x25, 0x0a, 0x0e, 0x67, 0x65, 0x6e, 0x5f, 0x74, 0x68, 0x72, 0x6f, 0x75,
+	0x67, 0x68, 0x70, 0x75, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28, 0x01, 0x52, 0x0d, 0x67, 0x65, 0x6e,
+	0x54, 0x68, 0x72, 0x6f, 0x75, 0x67, 0x68, 0x70, 0x75, 0x74, 0x12, 0x24, 0x0a, 0x0e, 0x63, 0x61,
+	0x63, 0x68, 0x65, 0x5f, 0x68, 0x69, 0x74, 0x5f, 0x72, 0x61, 0x74, 0x65, 0x18, 0x09, 0x20, 0x01,
+	0x28, 0x01, 0x52, 0x0c, 0x63, 0x61, 0x63, 0x68, 0x65, 0x48, 0x69, 0x74, 0x52, 0x61, 0x74, 0x65,
+	0x12, 0x20, 0x0a, 0x0b, 0x75, 0x74, 0x69, 0x6c, 0x69, 0x7a, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18,
+	0x0a, 0x20, 0x01, 0x28, 0x01, 0x52, 0x0b, 0x75, 0x74, 0x69, 0x6c, 0x69, 0x7a, 0x61, 0x74, 0x69,
+	0x6f, 0x6e, 0x12, 0x30, 0x0a, 0x14, 0x6d, 0x61, 0x78, 0x5f, 0x72, 0x75, 0x6e, 0x6e, 0x69, 0x6e,
+	0x67, 0x5f, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x73, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x05,
+	0x52, 0x12, 0x6d, 0x61, 0x78, 0x52, 0x75, 0x6e, 0x6e, 0x69, 0x6e, 0x67, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x73, 0x32, 0x5f, 0x0a, 0x0a, 0x56, 0x6c, 0x6c, 0x6d, 0x45, 0x6e, 0x67, 0x69,
+	0x6e, 0x65, 0x12, 0x51, 0x0a, 0x08, 0x47, 0x65, 0x74, 0x4c, 0x6f, 0x61, 0x64, 0x73, 0x12, 0x21,
+	0x2e, 0x76, 0x6c, 0x6c, 0x6d, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x65, 0x6e, 0x67, 0x69, 0x6e,
+	0x65, 0x2e, 0x47, 0x65, 0x74, 0x4c, 0x6f, 0x61, 0x64, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x1a, 0x22, 0x2e, 0x76, 0x6c, 0x6c, 0x6d, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x65, 0x6e,
+	0x67, 0x69, 0x6e, 0x65, 0x2e, 0x47, 0x65, 0x74, 0x4c, 0x6f, 0x61, 0x64, 0x73, 0x52, 0x65, 0x73,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x57, 0x5a, 0x55, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e,
+	0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x61, 0x63, 0x68, 0x65, 0x62, 0x6f, 0x78, 0x2d, 0x70, 0x72, 0x6f,
+	0x6a, 0x65, 0x63, 0x74, 0x2f, 0x69, 0x6e, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x2d, 0x63,
+	0x61, 0x63, 0x68, 0x65, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x61, 0x64, 0x61, 0x70, 0x74, 0x65, 0x72,
+	0x73, 0x2f, 0x65, 0x6e, 0x67, 0x69, 0x6e, 0x65, 0x2f, 0x76, 0x6c, 0x6c, 0x6d, 0x65, 0x6e, 0x67,
+	0x69, 0x6e, 0x65, 0x3b, 0x76, 0x6c, 0x6c, 0x6d, 0x65, 0x6e, 0x67, 0x69, 0x6e, 0x65, 0x62, 0x06,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+})
+
+var (
+	file_vllm_engine_proto_rawDescOnce sync.Once
+	file_vllm_engine_proto_rawDescData []byte
+)
+
+func file_vllm_engine_proto_rawDescGZIP() []byte {
+	file_vllm_engine_proto_rawDescOnce.Do(func() {
+		file_vllm_engine_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_vllm_engine_proto_rawDesc), len(file_vllm_engine_proto_rawDesc)))
+	})
+	return file_vllm_engine_proto_rawDescData
+}
+
+var file_vllm_engine_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_vllm_engine_proto_goTypes = []any{
+	(*GetLoadsRequest)(nil),  // 0: vllm.grpc.engine.GetLoadsRequest
+	(*GetLoadsResponse)(nil), // 1: vllm.grpc.engine.GetLoadsResponse
+	(*SchedulerLoad)(nil),    // 2: vllm.grpc.engine.SchedulerLoad
+}
+var file_vllm_engine_proto_depIdxs = []int32{
+	2, // 0: vllm.grpc.engine.GetLoadsResponse.loads:type_name -> vllm.grpc.engine.SchedulerLoad
+	0, // 1: vllm.grpc.engine.VllmEngine.GetLoads:input_type -> vllm.grpc.engine.GetLoadsRequest
+	1, // 2: vllm.grpc.engine.VllmEngine.GetLoads:output_type -> vllm.grpc.engine.GetLoadsResponse
+	2, // [2:3] is the sub-list for method output_type
+	1, // [1:2] is the sub-list for method input_type
+	1, // [1:1] is the sub-list for extension type_name
+	1, // [1:1] is the sub-list for extension extendee
+	0, // [0:1] is the sub-list for field type_name
+}
+
+func init() { file_vllm_engine_proto_init() }
+func file_vllm_engine_proto_init() {
+	if File_vllm_engine_proto != nil {
+		return
+	}
+	file_vllm_engine_proto_msgTypes[0].OneofWrappers = []any{}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: unsafe.Slice(unsafe.StringData(file_vllm_engine_proto_rawDesc), len(file_vllm_engine_proto_rawDesc)),
+			NumEnums:      0,
+			NumMessages:   3,
+			NumExtensions: 0,
+			NumServices:   1,
+		},
+		GoTypes:           file_vllm_engine_proto_goTypes,
+		DependencyIndexes: file_vllm_engine_proto_depIdxs,
+		MessageInfos:      file_vllm_engine_proto_msgTypes,
+	}.Build()
+	File_vllm_engine_proto = out.File
+	file_vllm_engine_proto_goTypes = nil
+	file_vllm_engine_proto_depIdxs = nil
+}
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.proto b/pkg/adapters/engine/vllmengine/vllm_engine.proto
new file mode 100644
index 00000000..ebf7ec08
--- /dev/null
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.proto
@@ -0,0 +1,49 @@
+syntax = "proto3";
+
+// Minimal vendored subset of the SMG vLLM engine gRPC contract
+// (machxai/smg: crates/grpc_client/proto/vllm_engine.proto) — ONLY the GetLoads
+// RPC and its messages. The kvevent-subscriber uses this to read engine load
+// over gRPC as an alternative to scraping HTTP /metrics (which an SMG gRPC
+// engine does not expose).
+//
+// The package name MUST stay `vllm.grpc.engine` to match the upstream service
+// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
+// field numbers mirror upstream. Kept out of the `proto/` module on purpose —
+// it is a vendored external contract, not part of IC's own gRPC API, so it is
+// not subject to IC's buf-lint conventions or the proto drift check.
+package vllm.grpc.engine;
+
+option go_package = "github.com/cachebox-project/inference-cache/pkg/adapters/engine/vllmengine;vllmengine";
+
+service VllmEngine {
+  // Scheduler load metrics, per DP rank.
+  rpc GetLoads(GetLoadsRequest) returns (GetLoadsResponse);
+}
+
+message GetLoadsRequest {
+  optional int32 dp_rank = 1;
+  repeated string include = 2;
+}
+
+message GetLoadsResponse {
+  string timestamp = 1;
+  string version = 2;
+  int32 dp_rank_count = 3;
+  repeated SchedulerLoad loads = 4;
+}
+
+// Per-DP-rank scheduler load snapshot. Mirrors upstream field numbers; unset
+// fields are 0.
+message SchedulerLoad {
+  int32 dp_rank = 1;
+  int32 num_running_reqs = 2;
+  int32 num_waiting_reqs = 3;
+  int32 num_total_reqs = 4;
+  int32 num_used_tokens = 5;
+  int32 max_total_num_tokens = 6;
+  double token_usage = 7;  // KV-cache utilization in [0,1)
+  double gen_throughput = 8;
+  double cache_hit_rate = 9;
+  double utilization = 10;
+  int32 max_running_requests = 11;
+}
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
new file mode 100644
index 00000000..191f4e65
--- /dev/null
+++ b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
@@ -0,0 +1,135 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.5.1
+// - protoc             v7.34.1
+// source: vllm_engine.proto
+
+// Minimal vendored subset of the SMG vLLM engine gRPC contract
+// (machxai/smg: crates/grpc_client/proto/vllm_engine.proto) — ONLY the GetLoads
+// RPC and its messages. The kvevent-subscriber uses this to read engine load
+// over gRPC as an alternative to scraping HTTP /metrics (which an SMG gRPC
+// engine does not expose).
+//
+// The package name MUST stay `vllm.grpc.engine` to match the upstream service
+// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
+// field numbers mirror upstream. Kept out of the `proto/` module on purpose —
+// it is a vendored external contract, not part of IC's own gRPC API, so it is
+// not subject to IC's buf-lint conventions or the proto drift check.
+
+package vllmengine
+
+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 (
+	VllmEngine_GetLoads_FullMethodName = "/vllm.grpc.engine.VllmEngine/GetLoads"
+)
+
+// VllmEngineClient is the client API for VllmEngine 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.
+type VllmEngineClient interface {
+	// Scheduler load metrics, per DP rank.
+	GetLoads(ctx context.Context, in *GetLoadsRequest, opts ...grpc.CallOption) (*GetLoadsResponse, error)
+}
+
+type vllmEngineClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewVllmEngineClient(cc grpc.ClientConnInterface) VllmEngineClient {
+	return &vllmEngineClient{cc}
+}
+
+func (c *vllmEngineClient) GetLoads(ctx context.Context, in *GetLoadsRequest, opts ...grpc.CallOption) (*GetLoadsResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(GetLoadsResponse)
+	err := c.cc.Invoke(ctx, VllmEngine_GetLoads_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// VllmEngineServer is the server API for VllmEngine service.
+// All implementations must embed UnimplementedVllmEngineServer
+// for forward compatibility.
+type VllmEngineServer interface {
+	// Scheduler load metrics, per DP rank.
+	GetLoads(context.Context, *GetLoadsRequest) (*GetLoadsResponse, error)
+	mustEmbedUnimplementedVllmEngineServer()
+}
+
+// UnimplementedVllmEngineServer 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 UnimplementedVllmEngineServer struct{}
+
+func (UnimplementedVllmEngineServer) GetLoads(context.Context, *GetLoadsRequest) (*GetLoadsResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method GetLoads not implemented")
+}
+func (UnimplementedVllmEngineServer) mustEmbedUnimplementedVllmEngineServer() {}
+func (UnimplementedVllmEngineServer) testEmbeddedByValue()                    {}
+
+// UnsafeVllmEngineServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to VllmEngineServer will
+// result in compilation errors.
+type UnsafeVllmEngineServer interface {
+	mustEmbedUnimplementedVllmEngineServer()
+}
+
+func RegisterVllmEngineServer(s grpc.ServiceRegistrar, srv VllmEngineServer) {
+	// If the following call pancis, it indicates UnimplementedVllmEngineServer 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(&VllmEngine_ServiceDesc, srv)
+}
+
+func _VllmEngine_GetLoads_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(GetLoadsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VllmEngineServer).GetLoads(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: VllmEngine_GetLoads_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VllmEngineServer).GetLoads(ctx, req.(*GetLoadsRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// VllmEngine_ServiceDesc is the grpc.ServiceDesc for VllmEngine service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var VllmEngine_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "vllm.grpc.engine.VllmEngine",
+	HandlerType: (*VllmEngineServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "GetLoads",
+			Handler:    _VllmEngine_GetLoads_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "vllm_engine.proto",
+}

From c5dff783665b3d908676a79923adef7b11781914 Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Wed, 22 Jul 2026 17:35:52 -0700
Subject: [PATCH 2/9] fix: sanitize GetLoads values, testable load-source
 selection, reproducible vendored proto

- Clamp external token_usage/cache_hit_rate to finite [0,1] so garbage from the
  engine can't overflow cache_memory_bytes or poison hit_rate (NaN-safe).
- Extract buildStatsScraper so the gRPC-vs-HTTP load-source selection is unit
  tested; add the control-flow test.
- Rename GRPCLoadsScraper* to the Go initialism; log the selected load source.
- Add proto-gen-vendored (run by proto-gen) + go:generate + CI drift check so the
  vendored engine stubs are reproducible and can't silently drift.
- Document the selectable gRPC stats path and the deferred sidecar auto-inject of
  --engine-loads-grpc (gated on the engine GetLoads floor) in the wiring doc.
---
 .github/workflows/ci.yml                      |   4 +-
 Makefile                                      |  12 +-
 cmd/kvevent-subscriber/main.go                | 120 ++++++++++++------
 cmd/kvevent-subscriber/main_test.go           |  50 ++++++++
 docs/design/kvevent-subscriber-wiring.md      |  24 +++-
 pkg/adapters/engine/grpc_loads_scraper.go     |  46 ++++---
 .../engine/grpc_loads_scraper_test.go         |  74 ++++++++---
 pkg/adapters/engine/vllmengine/generate.go    |  11 ++
 .../engine/vllmengine/vllm_engine.pb.go       |   6 +-
 .../engine/vllmengine/vllm_engine.proto       |   6 +-
 .../engine/vllmengine/vllm_engine_grpc.pb.go  |   6 +-
 11 files changed, 277 insertions(+), 82 deletions(-)
 create mode 100644 cmd/kvevent-subscriber/main_test.go
 create mode 100644 pkg/adapters/engine/vllmengine/generate.go

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index ea26d7e6..30f5b43a 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -106,8 +106,8 @@ jobs:
     - name: Verify protobufs up to date
       run: |
         make proto-gen
-        if [ -n "$(git status --porcelain -- pkg/server/proto)" ]; then
-          git status --short -- pkg/server/proto
+        if [ -n "$(git status --porcelain -- pkg/server/proto pkg/adapters/engine/vllmengine)" ]; then
+          git status --short -- pkg/server/proto pkg/adapters/engine/vllmengine
           exit 1
         fi
 
diff --git a/Makefile b/Makefile
index 7f7a445d..11ed5c61 100644
--- a/Makefile
+++ b/Makefile
@@ -220,9 +220,19 @@ manifests: controller-gen ## Generate CRD, RBAC, and webhook manifests.
 	$(CONTROLLER_GEN) rbac:roleName=inference-cache-manager-role crd webhook paths="./..." output:crd:artifacts:config=config/crd/bases output:rbac:artifacts:config=config/rbac output:webhook:artifacts:config=config/webhook
 
 .PHONY: proto-gen
-proto-gen: protoc-gen-go ## Generate protobuf Go code.
+proto-gen: protoc-gen-go proto-gen-vendored ## Generate protobuf Go code (IC contract + vendored external stubs).
 	PATH="$(LOCALBIN):$$PATH" protoc -I proto --go_out=. --go_opt=module=$(MODULE) --go-grpc_out=. --go-grpc_opt=module=$(MODULE) $$(find proto -name '*.proto' | sort)
 
+# The vendored external engine contract (GetLoads only) lives OUTSIDE proto/ on
+# purpose — it is not part of IC's own gRPC API and is exempt from buf lint — so
+# `proto-gen` above (which only scans proto/) does not cover it. This target
+# regenerates its stubs from the same pinned generators; it is a prerequisite of
+# proto-gen so a plain `make proto-gen` keeps both in sync, and CI diffs the
+# output the same way. See pkg/adapters/engine/vllmengine/vllm_engine.proto.
+.PHONY: proto-gen-vendored
+proto-gen-vendored: protoc-gen-go ## Regenerate the vendored external vLLM engine stubs (pkg/adapters/engine/vllmengine).
+	PATH="$(LOCALBIN):$$PATH" protoc -I pkg/adapters/engine/vllmengine --go_out=. --go_opt=module=$(MODULE) --go-grpc_out=. --go-grpc_opt=module=$(MODULE) pkg/adapters/engine/vllmengine/vllm_engine.proto
+
 .PHONY: proto-lint
 proto-lint: buf ## Lint the gRPC contract with buf (lint-only; codegen stays on protoc).
 	$(BUF) lint
diff --git a/cmd/kvevent-subscriber/main.go b/cmd/kvevent-subscriber/main.go
index b57c826c..4e3015a6 100644
--- a/cmd/kvevent-subscriber/main.go
+++ b/cmd/kvevent-subscriber/main.go
@@ -6,10 +6,13 @@
 // Two independent paths share the gRPC client:
 //   - Event path: ZMQ → decoded EventBatch → ReportCacheState (prefix adds) +
 //     PublishEvent (removals/clears). Debounced on a short window.
-//   - Stats path: HTTP GET against the engine's Prometheus /metrics → derived
-//     ReplicaStats → ReportCacheState (stats-only CSU). Ticks on its own
-//     cadence (~10s), so the snapshot/CR status surface (cache_memory_bytes,
-//     hit_rate, pressure) lights up regardless of event rate.
+//   - Stats path: engine load → derived ReplicaStats → ReportCacheState
+//     (stats-only CSU). The load source is selectable: HTTP GET against the
+//     engine's Prometheus /metrics (default), or the VllmEngine GetLoads gRPC
+//     RPC when --engine-loads-grpc is set (SMG gRPC engines expose no HTTP
+//     /metrics). Ticks on its own cadence (~10s), so the snapshot/CR status
+//     surface (cache_memory_bytes, hit_rate, pressure) lights up regardless of
+//     event rate.
 //
 // The two paths are independent failure domains — a scrape failure never
 // blocks the event stream, and an event-stream drop never delays a stats tick.
@@ -19,6 +22,7 @@ import (
 	"context"
 	"errors"
 	"flag"
+	"io"
 	"log/slog"
 	"net/http"
 	"os"
@@ -33,6 +37,50 @@ import (
 	icpb "github.com/cachebox-project/inference-cache/pkg/server/proto/inferencecache/v1alpha1"
 )
 
+// engineScraper is the load-source contract both scrapers satisfy — a local
+// mirror of the engine package's unexported statsScraper. NewStatsReporter
+// accepts either concrete type through it.
+type engineScraper interface {
+	Scrape(context.Context) (*icpb.ReplicaStats, error)
+}
+
+// scraperParams carries the load-source selection inputs for buildStatsScraper.
+type scraperParams struct {
+	loadsGRPC      string
+	metricsURL     string
+	engineModel    string
+	tier           engine.CacheTier
+	cacheSizeBytes int64
+	ceiling        int
+}
+
+// buildStatsScraper selects the engine load source: the GetLoads gRPC scraper
+// when loadsGRPC is non-empty (preferred for SMG gRPC engines, which expose no
+// HTTP /metrics), else the HTTP /metrics scraper. It returns the scraper, an
+// optional closer (the gRPC scraper owns a client conn — the HTTP one owns
+// nothing, so closer is nil), and an error only from gRPC dial setup.
+func buildStatsScraper(p scraperParams, httpClient *http.Client, logger *slog.Logger) (engineScraper, io.Closer, error) {
+	if p.loadsGRPC != "" {
+		gs, err := engine.NewGRPCLoadsScraper(engine.GRPCLoadsScraperConfig{
+			Addr:                  p.loadsGRPC,
+			CacheSizeBytes:        p.cacheSizeBytes,
+			MaxConcurrencyCeiling: p.ceiling,
+		})
+		if err != nil {
+			return nil, nil, err
+		}
+		return gs, gs, nil
+	}
+	ms := engine.NewMetricsScraper(httpClient, engine.ScraperConfig{
+		URL:                   p.metricsURL,
+		Tier:                  p.tier,
+		ModelLabel:            p.engineModel,
+		CacheSizeBytes:        p.cacheSizeBytes,
+		MaxConcurrencyCeiling: p.ceiling,
+	}, logger)
+	return ms, nil, nil
+}
+
 func main() {
 	var (
 		endpoint           = flag.String("engine-endpoint", "tcp://127.0.0.1:5557", "engine KV-event ZMQ PUB endpoint")
@@ -44,7 +92,7 @@ func main() {
 		scheme             = flag.String("hash-scheme", "vllm", "engine prefix-hash scheme (required, non-empty)")
 		window             = flag.Duration("window", 100*time.Millisecond, "add-batching/debounce flush window")
 		metricsURL         = flag.String("engine-metrics-url", "http://127.0.0.1:8000/metrics", "engine Prometheus /metrics URL")
-		loadsGrpc          = flag.String("engine-loads-grpc", "", "engine VllmEngine gRPC address (host:port) to read load via the GetLoads RPC instead of scraping --engine-metrics-url. Preferred for SMG gRPC engines, which expose no HTTP /metrics. Empty = use the HTTP scrape.")
+		loadsGRPC          = flag.String("engine-loads-grpc", "", "engine VllmEngine gRPC address (host:port) to read load via the GetLoads RPC instead of scraping --engine-metrics-url. Preferred for SMG gRPC engines, which expose no HTTP /metrics. Empty = use the HTTP scrape.")
 		statsInterval      = flag.Duration("stats-interval", 10*time.Second, "ReplicaStats scrape/emit cadence")
 		cacheSizeBytes     = flag.Int64("engine-cache-size-bytes", 0, "engine total KV-cache capacity in bytes (multiplied by usage_perc to derive cacheMemoryBytes; 0 emits cacheMemoryBytes=0)")
 		ceiling            = flag.Int("max-concurrency-ceiling", 256, "denominator for the pressure proxy = clamp01((num_requests_running+num_requests_waiting)/ceiling)")
@@ -95,42 +143,33 @@ func main() {
 	// Load source: gRPC GetLoads (preferred for SMG gRPC engines, which expose no
 	// HTTP /metrics) when --engine-loads-grpc is set, else the HTTP /metrics scrape.
 	// Both satisfy the same statsScraper, so the StatsReporter is identical.
-	var statsReporter *engine.StatsReporter
-	if *loadsGrpc != "" {
-		gs, gerr := engine.NewGrpcLoadsScraper(engine.GrpcLoadsScraperConfig{
-			Addr:                  *loadsGrpc,
-			CacheSizeBytes:        *cacheSizeBytes,
-			MaxConcurrencyCeiling: *ceiling,
-		})
-		if gerr != nil {
-			logger.Error("grpc loads scraper", "addr", *loadsGrpc, "err", gerr)
-			os.Exit(1)
-		}
-		defer func() { _ = gs.Close() }()
-		logger.Info("engine load source: GetLoads gRPC", "addr", *loadsGrpc)
-		statsReporter = engine.NewStatsReporter(client, gs, cfg,
-			engine.WithStatsInterval(*statsInterval),
-			engine.WithStatsLogger(logger),
-		)
-	} else {
-		scraper := engine.NewMetricsScraper(
-			&http.Client{Timeout: 5 * time.Second},
-			engine.ScraperConfig{
-				URL:                   *metricsURL,
-				Tier:                  tier,
-				ModelLabel:            *engineModel,
-				CacheSizeBytes:        *cacheSizeBytes,
-				MaxConcurrencyCeiling: *ceiling,
-			},
-			logger,
-		)
-		logger.Info("engine load source: HTTP /metrics scrape", "url", *metricsURL)
-		statsReporter = engine.NewStatsReporter(client, scraper, cfg,
-			engine.WithStatsInterval(*statsInterval),
-			engine.WithStatsLogger(logger),
-		)
+	scraper, scraperCloser, serr := buildStatsScraper(scraperParams{
+		loadsGRPC:      *loadsGRPC,
+		metricsURL:     *metricsURL,
+		engineModel:    *engineModel,
+		tier:           tier,
+		cacheSizeBytes: *cacheSizeBytes,
+		ceiling:        *ceiling,
+	}, &http.Client{Timeout: 5 * time.Second}, logger)
+	if serr != nil {
+		logger.Error("build stats scraper", "err", serr)
+		os.Exit(1)
+	}
+	if scraperCloser != nil {
+		defer func() { _ = scraperCloser.Close() }()
 	}
 
+	loadSource, loadTarget := "HTTP /metrics scrape", *metricsURL
+	if *loadsGRPC != "" {
+		loadSource, loadTarget = "GetLoads gRPC", *loadsGRPC
+	}
+	logger.Info("engine load source", "source", loadSource, "target", loadTarget)
+
+	statsReporter := engine.NewStatsReporter(client, scraper, cfg,
+		engine.WithStatsInterval(*statsInterval),
+		engine.WithStatsLogger(logger),
+	)
+
 	out := make(chan *engine.EventBatch, 256)
 
 	// The reporter stops by draining a closed channel, not by signal — so on
@@ -159,7 +198,8 @@ func main() {
 		"server", *server,
 		"replica_id", *replica,
 		"model_id", *model,
-		"engine_metrics_url", *metricsURL,
+		"load_source", loadSource,
+		"load_target", loadTarget,
 		"stats_interval", statsInterval.String(),
 		"cache_tier", *cacheTier,
 		"engine_model_name", *engineModel,
diff --git a/cmd/kvevent-subscriber/main_test.go b/cmd/kvevent-subscriber/main_test.go
new file mode 100644
index 00000000..e162139e
--- /dev/null
+++ b/cmd/kvevent-subscriber/main_test.go
@@ -0,0 +1,50 @@
+package main
+
+import (
+	"io"
+	"log/slog"
+	"net/http"
+	"testing"
+
+	"github.com/cachebox-project/inference-cache/pkg/adapters/engine"
+)
+
+// TestBuildStatsScraperSelectsSource pins the load-source control flow: a
+// non-empty --engine-loads-grpc must select the GetLoads gRPC scraper (and hand
+// back a closer for its conn), while an empty flag must preserve the HTTP
+// /metrics scraper (which owns no conn, so no closer). This is the crux of
+// Option B — the wrong branch silently leaves an SMG gRPC engine reporting no
+// load — so it gets an explicit test.
+func TestBuildStatsScraperSelectsSource(t *testing.T) {
+	hc := &http.Client{}
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	t.Run("nonempty flag selects gRPC", func(t *testing.T) {
+		s, closer, err := buildStatsScraper(scraperParams{loadsGRPC: "127.0.0.1:50051"}, hc, logger)
+		if err != nil {
+			t.Fatalf("buildStatsScraper: %v", err)
+		}
+		if _, ok := s.(*engine.GRPCLoadsScraper); !ok {
+			t.Fatalf("selected %T, want *engine.GRPCLoadsScraper", s)
+		}
+		if closer == nil {
+			t.Fatal("gRPC scraper must return a closer to release its client conn")
+		}
+		if err := closer.Close(); err != nil {
+			t.Errorf("closer.Close: %v", err)
+		}
+	})
+
+	t.Run("empty flag preserves HTTP", func(t *testing.T) {
+		s, closer, err := buildStatsScraper(scraperParams{metricsURL: "http://127.0.0.1:8000/metrics"}, hc, logger)
+		if err != nil {
+			t.Fatalf("buildStatsScraper: %v", err)
+		}
+		if _, ok := s.(*engine.MetricsScraper); !ok {
+			t.Fatalf("selected %T, want *engine.MetricsScraper", s)
+		}
+		if closer != nil {
+			t.Fatal("HTTP scraper owns no conn; closer must be nil")
+		}
+	})
+}
diff --git a/docs/design/kvevent-subscriber-wiring.md b/docs/design/kvevent-subscriber-wiring.md
index 19d437c9..d91599c1 100644
--- a/docs/design/kvevent-subscriber-wiring.md
+++ b/docs/design/kvevent-subscriber-wiring.md
@@ -66,11 +66,15 @@ Concretely:
   admission picks it up once the operator sets the key), `--hash-scheme` ← the
   adapter's runtime convention (`"vllm"` or `"sglang"`), `--server` ← the policy-server
   in-cluster Service DNS (operator-configurable via a controller flag),
-  `--engine-endpoint` ← `tcp://127.0.0.1:<engine ZMQ port>`. The stats-path flags
-  (`--engine-metrics-url`, `--stats-interval`, etc.) are added by the adapter when the
-  shipped subscriber binary learns to scrape and emit `ReplicaStats`; passing flags the
-  binary doesn't recognise would crash the sidecar at startup. No operator-supplied
-  `--replica-id` / `--model-id` on the demo path.
+  `--engine-endpoint` ← `tcp://127.0.0.1:<engine ZMQ port>`. The stats path runs on
+  the subscriber binary's built-in defaults: it scrapes the engine's HTTP Prometheus
+  `/metrics` (the `--engine-metrics-url` default) and emits `ReplicaStats` on its own
+  cadence. Explicit stats-path flags — `--stats-interval`, the cache-size/ceiling
+  hints, and `--engine-loads-grpc` (read load via the engine's GetLoads gRPC RPC
+  instead of HTTP; see the deferred note below) — are a follow-up enrichment the
+  sidecar does not set yet; passing a flag the binary doesn't recognise would crash
+  the sidecar at startup. No operator-supplied `--replica-id` / `--model-id` on the
+  demo path.
 
 ### Why this combination
 
@@ -101,6 +105,16 @@ Concretely:
   ceilings, cache-size hint) — not added in this PR. The sidecar passes the
   `kvevent-subscriber` binary's flag defaults and can be enriched in a follow-up when an
   operator needs the knobs.
+* **Auto-injecting `--engine-loads-grpc` (GetLoads load source)** — the subscriber binary
+  can now read engine load over the VllmEngine `GetLoads` gRPC RPC instead of scraping HTTP
+  `/metrics` (an SMG gRPC engine exposes no `/metrics` endpoint at all), selected by the
+  `--engine-loads-grpc` flag. The sidecar renderer does **not** pass that flag yet, so
+  auto-injected sidecars keep the HTTP default. Deferred deliberately: `GetLoads` requires a
+  newer engine floor (SMG gRPC servicer ≥ 0.5.2 / vLLM ≥ 0.19) than the engine currently
+  deployed, so auto-wiring it now would point every sidecar at an unimplemented RPC and flip
+  the load signal to "stale" (see the `StatsReporter` stale-escalation path). It gets wired
+  alongside the stats-path knobs above once the engine floor moves; until then it is opt-in
+  via the binary flag for gRPC-only engine deployments.
 * **TLS subscriber → policy-server** — separate ticket.
 * **HA / multi-server policy-server target** — separate ticket.
 * **Readiness gating on first KV event observed** — natural follow-up once this lands.
diff --git a/pkg/adapters/engine/grpc_loads_scraper.go b/pkg/adapters/engine/grpc_loads_scraper.go
index 9ed2661a..813413ca 100644
--- a/pkg/adapters/engine/grpc_loads_scraper.go
+++ b/pkg/adapters/engine/grpc_loads_scraper.go
@@ -3,6 +3,7 @@ package engine
 import (
 	"context"
 	"fmt"
+	"math"
 	"time"
 
 	"google.golang.org/grpc"
@@ -12,8 +13,8 @@ import (
 	icpb "github.com/cachebox-project/inference-cache/pkg/server/proto/inferencecache/v1alpha1"
 )
 
-// GrpcLoadsScraperConfig tunes the gRPC GetLoads scraper. Addr is required.
-type GrpcLoadsScraperConfig struct {
+// GRPCLoadsScraperConfig tunes the gRPC GetLoads scraper. Addr is required.
+type GRPCLoadsScraperConfig struct {
 	// Addr is the engine's VllmEngine gRPC address (host:port), e.g.
 	// 127.0.0.1:50051.
 	Addr string
@@ -29,7 +30,7 @@ type GrpcLoadsScraperConfig struct {
 	Timeout time.Duration
 }
 
-// GrpcLoadsScraper reads engine load via the SMG engine's GetLoads gRPC RPC and
+// GRPCLoadsScraper reads engine load via the SMG engine's GetLoads gRPC RPC and
 // projects it into a ReplicaStats — the gRPC-native alternative to scraping the
 // engine's HTTP /metrics. An SMG gRPC engine (vllm.entrypoints.grpc_server)
 // exposes no HTTP metrics endpoint; live load is available only over GetLoads.
@@ -40,16 +41,16 @@ type GrpcLoadsScraperConfig struct {
 // T2HitTokens/T2QueryTokens are left 0 here — those remain HTTP-/metrics-only.
 // The load signals the ranker actually uses (pressure, cache-usage, hit-rate)
 // are all provided.
-type GrpcLoadsScraper struct {
+type GRPCLoadsScraper struct {
 	client vpb.VllmEngineClient
-	cfg    GrpcLoadsScraperConfig
+	cfg    GRPCLoadsScraperConfig
 	closer func() error
 }
 
-// NewGrpcLoadsScraper dials the engine (lazily — grpc.NewClient does not connect
+// NewGRPCLoadsScraper dials the engine (lazily — grpc.NewClient does not connect
 // until the first RPC, so a down engine doesn't fail construction) and returns a
 // scraper. Call Close to release the connection.
-func NewGrpcLoadsScraper(cfg GrpcLoadsScraperConfig) (*GrpcLoadsScraper, error) {
+func NewGRPCLoadsScraper(cfg GRPCLoadsScraperConfig) (*GRPCLoadsScraper, error) {
 	if cfg.Addr == "" {
 		return nil, fmt.Errorf("grpc loads scraper: Addr is required")
 	}
@@ -60,24 +61,24 @@ func NewGrpcLoadsScraper(cfg GrpcLoadsScraperConfig) (*GrpcLoadsScraper, error)
 	if err != nil {
 		return nil, fmt.Errorf("grpc loads scraper: dial %s: %w", cfg.Addr, err)
 	}
-	return &GrpcLoadsScraper{client: vpb.NewVllmEngineClient(conn), cfg: cfg, closer: conn.Close}, nil
+	return &GRPCLoadsScraper{client: vpb.NewVllmEngineClient(conn), cfg: cfg, closer: conn.Close}, nil
 }
 
-// newGrpcLoadsScraperWithClient injects a client (tests) and never owns a conn.
-func newGrpcLoadsScraperWithClient(c vpb.VllmEngineClient, cfg GrpcLoadsScraperConfig) *GrpcLoadsScraper {
+// newGRPCLoadsScraperWithClient injects a client (tests) and never owns a conn.
+func newGRPCLoadsScraperWithClient(c vpb.VllmEngineClient, cfg GRPCLoadsScraperConfig) *GRPCLoadsScraper {
 	if cfg.Timeout <= 0 {
 		cfg.Timeout = defaultScrapeTimeout
 	}
-	return &GrpcLoadsScraper{client: c, cfg: cfg, closer: func() error { return nil }}
+	return &GRPCLoadsScraper{client: c, cfg: cfg, closer: func() error { return nil }}
 }
 
 // Close releases the underlying gRPC connection.
-func (s *GrpcLoadsScraper) Close() error { return s.closer() }
+func (s *GRPCLoadsScraper) Close() error { return s.closer() }
 
 // Scrape calls GetLoads once and projects the per-DP-rank SchedulerLoad into a
 // ReplicaStats. On error a zero-valued *icpb.ReplicaStats is returned alongside
 // the error — the StatsReporter logs and skips the tick, same as the HTTP path.
-func (s *GrpcLoadsScraper) Scrape(ctx context.Context) (*icpb.ReplicaStats, error) {
+func (s *GRPCLoadsScraper) Scrape(ctx context.Context) (*icpb.ReplicaStats, error) {
 	rctx, cancel := context.WithTimeout(ctx, s.cfg.Timeout)
 	defer cancel()
 
@@ -95,10 +96,14 @@ func (s *GrpcLoadsScraper) Scrape(ctx context.Context) (*icpb.ReplicaStats, erro
 	for _, l := range resp.GetLoads() {
 		running += int64(l.GetNumRunningReqs())
 		waiting += int64(l.GetNumWaitingReqs())
-		if u := l.GetTokenUsage(); u > usage {
+		// token_usage and cache_hit_rate come from an external process and are
+		// contractually ratios in [0,1]. Sanitize before use: an out-of-range or
+		// non-finite value must not overflow cache_memory_bytes (usage *
+		// CacheSizeBytes) or push hit_rate outside [0,1].
+		if u := finite01(l.GetTokenUsage()); u > usage {
 			usage = u
 		}
-		hitRateSum += l.GetCacheHitRate()
+		hitRateSum += finite01(l.GetCacheHitRate())
 		hitRateN++
 	}
 	hitRate := 0.0
@@ -117,3 +122,14 @@ func (s *GrpcLoadsScraper) Scrape(ctx context.Context) (*icpb.ReplicaStats, erro
 		Pressure:         float32(pressure),
 	}, nil
 }
+
+// finite01 clamps a raw engine-reported ratio into [0,1], mapping non-finite
+// (NaN / ±Inf) values to 0. clamp01 alone would pass NaN through (NaN compares
+// false against both bounds), so this guards the external GetLoads values that
+// feed cache_memory_bytes and hit_rate.
+func finite01(v float64) float64 {
+	if math.IsNaN(v) || math.IsInf(v, 0) {
+		return 0
+	}
+	return clamp01(v)
+}
diff --git a/pkg/adapters/engine/grpc_loads_scraper_test.go b/pkg/adapters/engine/grpc_loads_scraper_test.go
index d2efeaee..deffa94f 100644
--- a/pkg/adapters/engine/grpc_loads_scraper_test.go
+++ b/pkg/adapters/engine/grpc_loads_scraper_test.go
@@ -3,6 +3,7 @@ package engine
 import (
 	"context"
 	"errors"
+	"math"
 	"net"
 	"testing"
 
@@ -23,12 +24,12 @@ func (f *fakeLoadsClient) GetLoads(_ context.Context, _ *vpb.GetLoadsRequest, _
 	return f.resp, f.err
 }
 
-func TestGrpcLoadsScraperMapsSingleRank(t *testing.T) {
+func TestGRPCLoadsScraperMapsSingleRank(t *testing.T) {
 	resp := &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
 		{NumRunningReqs: 6, NumWaitingReqs: 2, TokenUsage: 0.5, CacheHitRate: 0.8},
 	}}
-	s := newGrpcLoadsScraperWithClient(&fakeLoadsClient{resp: resp},
-		GrpcLoadsScraperConfig{CacheSizeBytes: 1 << 30, MaxConcurrencyCeiling: 16})
+	s := newGRPCLoadsScraperWithClient(&fakeLoadsClient{resp: resp},
+		GRPCLoadsScraperConfig{CacheSizeBytes: 1 << 30, MaxConcurrencyCeiling: 16})
 
 	st, err := s.Scrape(context.Background())
 	if err != nil {
@@ -45,14 +46,14 @@ func TestGrpcLoadsScraperMapsSingleRank(t *testing.T) {
 	}
 }
 
-func TestGrpcLoadsScraperAggregatesRanks(t *testing.T) {
+func TestGRPCLoadsScraperAggregatesRanks(t *testing.T) {
 	// two DP ranks: running/waiting SUM, usage MAX, hit-rate MEAN
 	resp := &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
 		{NumRunningReqs: 2, NumWaitingReqs: 1, TokenUsage: 0.3, CacheHitRate: 0.5},
 		{NumRunningReqs: 5, NumWaitingReqs: 4, TokenUsage: 0.7, CacheHitRate: 0.9},
 	}}
-	s := newGrpcLoadsScraperWithClient(&fakeLoadsClient{resp: resp},
-		GrpcLoadsScraperConfig{CacheSizeBytes: 1000, MaxConcurrencyCeiling: 24})
+	s := newGRPCLoadsScraperWithClient(&fakeLoadsClient{resp: resp},
+		GRPCLoadsScraperConfig{CacheSizeBytes: 1000, MaxConcurrencyCeiling: 24})
 
 	st, err := s.Scrape(context.Background())
 	if err != nil {
@@ -69,12 +70,12 @@ func TestGrpcLoadsScraperAggregatesRanks(t *testing.T) {
 	}
 }
 
-func TestGrpcLoadsScraperZeroConfigIsHonest(t *testing.T) {
+func TestGRPCLoadsScraperZeroConfigIsHonest(t *testing.T) {
 	// CacheSizeBytes=0 -> 0 bytes (not fabricated); ceiling=0 -> pressure 0
 	resp := &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
 		{NumRunningReqs: 9, NumWaitingReqs: 9, TokenUsage: 0.9},
 	}}
-	s := newGrpcLoadsScraperWithClient(&fakeLoadsClient{resp: resp}, GrpcLoadsScraperConfig{})
+	s := newGRPCLoadsScraperWithClient(&fakeLoadsClient{resp: resp}, GRPCLoadsScraperConfig{})
 	st, err := s.Scrape(context.Background())
 	if err != nil {
 		t.Fatalf("Scrape: %v", err)
@@ -85,9 +86,43 @@ func TestGrpcLoadsScraperZeroConfigIsHonest(t *testing.T) {
 	}
 }
 
-func TestGrpcLoadsScraperErrorIsSurfaced(t *testing.T) {
-	s := newGrpcLoadsScraperWithClient(&fakeLoadsClient{err: errors.New("unavailable")},
-		GrpcLoadsScraperConfig{Addr: "x:1"})
+func TestGRPCLoadsScraperSanitizesOutOfRange(t *testing.T) {
+	// token_usage and cache_hit_rate are contractually [0,1] ratios, but come
+	// from an external engine process. Feed garbage across ranks: over 1, NaN,
+	// and negative. Sanitization must clamp/reject each so cache_memory_bytes
+	// never overflows past CacheSizeBytes and hit_rate stays finite in [0,1].
+	resp := &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
+		{TokenUsage: 5.0, CacheHitRate: 9.0},               // both way over 1
+		{TokenUsage: math.NaN(), CacheHitRate: math.NaN()}, // non-finite
+		{TokenUsage: -1.0, CacheHitRate: -0.5},             // negative
+	}}
+	s := newGRPCLoadsScraperWithClient(&fakeLoadsClient{resp: resp},
+		GRPCLoadsScraperConfig{CacheSizeBytes: 1000, MaxConcurrencyCeiling: 8})
+
+	st, err := s.Scrape(context.Background())
+	if err != nil {
+		t.Fatalf("Scrape: %v", err)
+	}
+	// usage clamps per-rank to [0,1]; MAX(1,0,0)=1 → bytes = 1*1000, never 5000/negative.
+	if got := st.GetCacheMemoryBytes(); got != 1000 {
+		t.Errorf("cache_memory_bytes = %d, want 1000 (usage clamped to 1, not overflowed)", got)
+	}
+	// hit_rate clamps per-rank to [0,1]; MEAN(1,0,0)=1/3, finite (no NaN poison).
+	hr := float64(st.GetHitRate())
+	if math.IsNaN(hr) || math.IsInf(hr, 0) {
+		t.Fatalf("hit_rate = %v, want a finite value (NaN input must not propagate)", hr)
+	}
+	if hr < 0 || hr > 1 {
+		t.Errorf("hit_rate = %v, want within [0,1]", hr)
+	}
+	if math.Abs(hr-1.0/3.0) > 1e-6 {
+		t.Errorf("hit_rate = %v, want ~0.3333 (mean of clamped 1,0,0)", hr)
+	}
+}
+
+func TestGRPCLoadsScraperErrorIsSurfaced(t *testing.T) {
+	s := newGRPCLoadsScraperWithClient(&fakeLoadsClient{err: errors.New("unavailable")},
+		GRPCLoadsScraperConfig{Addr: "x:1"})
 	st, err := s.Scrape(context.Background())
 	if err == nil {
 		t.Fatal("want error, got nil")
@@ -108,7 +143,7 @@ func (m *mockEngineServer) GetLoads(context.Context, *vpb.GetLoadsRequest) (*vpb
 	return m.resp, nil
 }
 
-func TestGrpcLoadsScraperEndToEndOverBufconn(t *testing.T) {
+func TestGRPCLoadsScraperEndToEndOverBufconn(t *testing.T) {
 	lis := bufconn.Listen(1 << 20)
 	srv := grpc.NewServer()
 	vpb.RegisterVllmEngineServer(srv, &mockEngineServer{
@@ -116,8 +151,15 @@ func TestGrpcLoadsScraperEndToEndOverBufconn(t *testing.T) {
 			{NumRunningReqs: 3, NumWaitingReqs: 1, TokenUsage: 0.25, CacheHitRate: 0.6},
 		}},
 	})
-	go func() { _ = srv.Serve(lis) }()
-	defer srv.Stop()
+	serveErr := make(chan error, 1)
+	go func() { serveErr <- srv.Serve(lis) }()
+	t.Cleanup(func() {
+		srv.Stop()
+		// Serve returns ErrServerStopped on Stop(); anything else is a real fault.
+		if err := <-serveErr; err != nil && !errors.Is(err, grpc.ErrServerStopped) {
+			t.Errorf("bufconn Serve terminated unexpectedly: %v", err)
+		}
+	})
 
 	conn, err := grpc.NewClient("passthrough:///bufnet",
 		grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }),
@@ -127,8 +169,8 @@ func TestGrpcLoadsScraperEndToEndOverBufconn(t *testing.T) {
 	}
 	defer func() { _ = conn.Close() }()
 
-	s := newGrpcLoadsScraperWithClient(vpb.NewVllmEngineClient(conn),
-		GrpcLoadsScraperConfig{CacheSizeBytes: 1 << 30, MaxConcurrencyCeiling: 8})
+	s := newGRPCLoadsScraperWithClient(vpb.NewVllmEngineClient(conn),
+		GRPCLoadsScraperConfig{CacheSizeBytes: 1 << 30, MaxConcurrencyCeiling: 8})
 	st, err := s.Scrape(context.Background())
 	if err != nil {
 		t.Fatalf("Scrape: %v", err)
diff --git a/pkg/adapters/engine/vllmengine/generate.go b/pkg/adapters/engine/vllmengine/generate.go
new file mode 100644
index 00000000..3af77307
--- /dev/null
+++ b/pkg/adapters/engine/vllmengine/generate.go
@@ -0,0 +1,11 @@
+// Package vllmengine holds a minimal vendored subset of the SMG vLLM engine gRPC
+// contract (GetLoads only), used by the kvevent-subscriber to read engine load
+// over gRPC instead of scraping HTTP /metrics. The .pb.go files are generated,
+// not hand-edited.
+//
+// Regenerate with `make proto-gen-vendored` (also run as part of `make
+// proto-gen`); CI diffs the output to catch drift. See vllm_engine.proto for the
+// contract and why it is kept outside the proto/ module.
+package vllmengine
+
+//go:generate make -C ../../../.. proto-gen-vendored
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
index abf05374..8d17fdd8 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
@@ -14,7 +14,11 @@
 // so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
 // field numbers mirror upstream. Kept out of the `proto/` module on purpose —
 // it is a vendored external contract, not part of IC's own gRPC API, so it is
-// not subject to IC's buf-lint conventions or the proto drift check.
+// exempt from IC's buf-lint conventions.
+//
+// Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
+// proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
+// two committed stubs stay reproducible from this file.
 
 package vllmengine
 
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.proto b/pkg/adapters/engine/vllmengine/vllm_engine.proto
index ebf7ec08..0d34b4c9 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine.proto
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.proto
@@ -10,7 +10,11 @@ syntax = "proto3";
 // so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
 // field numbers mirror upstream. Kept out of the `proto/` module on purpose —
 // it is a vendored external contract, not part of IC's own gRPC API, so it is
-// not subject to IC's buf-lint conventions or the proto drift check.
+// exempt from IC's buf-lint conventions.
+//
+// Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
+// proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
+// two committed stubs stay reproducible from this file.
 package vllm.grpc.engine;
 
 option go_package = "github.com/cachebox-project/inference-cache/pkg/adapters/engine/vllmengine;vllmengine";
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
index 191f4e65..2d2401b2 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
+++ b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
@@ -14,7 +14,11 @@
 // so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
 // field numbers mirror upstream. Kept out of the `proto/` module on purpose —
 // it is a vendored external contract, not part of IC's own gRPC API, so it is
-// not subject to IC's buf-lint conventions or the proto drift check.
+// exempt from IC's buf-lint conventions.
+//
+// Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
+// proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
+// two committed stubs stay reproducible from this file.
 
 package vllmengine
 

From d4ba51c544af64346008157b3e3e9826bcf3a38d Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Wed, 22 Jul 2026 17:46:36 -0700
Subject: [PATCH 3/9] fix: exclude non-finite hit-rate ranks from the mean

Mapping a non-finite cache_hit_rate to 0 and still counting it dragged the
mean down and contradicted the 'ranks that report one' comment. Exclude
non-finite ranks from the average instead; finite out-of-range values are
still clamped and kept. token_usage keeps finite01 (0 never wins its max).
---
 pkg/adapters/engine/grpc_loads_scraper.go     | 19 ++++++++++++-------
 .../engine/grpc_loads_scraper_test.go         |  7 ++++---
 2 files changed, 16 insertions(+), 10 deletions(-)

diff --git a/pkg/adapters/engine/grpc_loads_scraper.go b/pkg/adapters/engine/grpc_loads_scraper.go
index 813413ca..e9d07eab 100644
--- a/pkg/adapters/engine/grpc_loads_scraper.go
+++ b/pkg/adapters/engine/grpc_loads_scraper.go
@@ -89,22 +89,27 @@ func (s *GRPCLoadsScraper) Scrape(ctx context.Context) (*icpb.ReplicaStats, erro
 
 	// Aggregate across DP ranks: request counts SUM (total in-flight across the
 	// engine), KV-cache usage is the MAX (worst-case pressure proxy), hit-rate is
-	// the mean of ranks that report one.
+	// the mean over ranks that report a finite value.
+	//
+	// token_usage and cache_hit_rate come from an external process and are
+	// contractually ratios in [0,1] — sanitize both. token_usage feeds the MAX and
+	// then cache_memory_bytes, so a non-finite value maps to 0 (finite01) and just
+	// never wins the max. cache_hit_rate feeds a mean, so a non-finite (unavailable)
+	// rank is EXCLUDED rather than counted as 0 — otherwise one garbage sample would
+	// drag the average down; finite-but-out-of-range values are clamped and kept.
 	var running, waiting int64
 	var usage, hitRateSum float64
 	var hitRateN int
 	for _, l := range resp.GetLoads() {
 		running += int64(l.GetNumRunningReqs())
 		waiting += int64(l.GetNumWaitingReqs())
-		// token_usage and cache_hit_rate come from an external process and are
-		// contractually ratios in [0,1]. Sanitize before use: an out-of-range or
-		// non-finite value must not overflow cache_memory_bytes (usage *
-		// CacheSizeBytes) or push hit_rate outside [0,1].
 		if u := finite01(l.GetTokenUsage()); u > usage {
 			usage = u
 		}
-		hitRateSum += finite01(l.GetCacheHitRate())
-		hitRateN++
+		if hr := l.GetCacheHitRate(); !math.IsNaN(hr) && !math.IsInf(hr, 0) {
+			hitRateSum += clamp01(hr)
+			hitRateN++
+		}
 	}
 	hitRate := 0.0
 	if hitRateN > 0 {
diff --git a/pkg/adapters/engine/grpc_loads_scraper_test.go b/pkg/adapters/engine/grpc_loads_scraper_test.go
index deffa94f..160509bc 100644
--- a/pkg/adapters/engine/grpc_loads_scraper_test.go
+++ b/pkg/adapters/engine/grpc_loads_scraper_test.go
@@ -107,7 +107,8 @@ func TestGRPCLoadsScraperSanitizesOutOfRange(t *testing.T) {
 	if got := st.GetCacheMemoryBytes(); got != 1000 {
 		t.Errorf("cache_memory_bytes = %d, want 1000 (usage clamped to 1, not overflowed)", got)
 	}
-	// hit_rate clamps per-rank to [0,1]; MEAN(1,0,0)=1/3, finite (no NaN poison).
+	// hit_rate: the NaN rank is EXCLUDED (unavailable), the over-1 rank clamps to 1
+	// and the negative to 0 → MEAN(1,0)=0.5, finite (no NaN poison).
 	hr := float64(st.GetHitRate())
 	if math.IsNaN(hr) || math.IsInf(hr, 0) {
 		t.Fatalf("hit_rate = %v, want a finite value (NaN input must not propagate)", hr)
@@ -115,8 +116,8 @@ func TestGRPCLoadsScraperSanitizesOutOfRange(t *testing.T) {
 	if hr < 0 || hr > 1 {
 		t.Errorf("hit_rate = %v, want within [0,1]", hr)
 	}
-	if math.Abs(hr-1.0/3.0) > 1e-6 {
-		t.Errorf("hit_rate = %v, want ~0.3333 (mean of clamped 1,0,0)", hr)
+	if math.Abs(hr-0.5) > 1e-6 {
+		t.Errorf("hit_rate = %v, want 0.5 (mean of clamped 1 and 0; NaN rank excluded)", hr)
 	}
 }
 

From abd07c7f39fd1bccc6df67925d54bd0cb52b57ce Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Thu, 23 Jul 2026 11:32:23 -0700
Subject: [PATCH 4/9] chore: exclude vendored engine stubs from coverage gate;
 tighten proto provenance
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The generated vllmengine/*.pb.go were counted as hand-written by cover-check;
add the package to COVER_EXCLUDE (it is entirely generated). Also strengthen the
proto provenance note: the fields are copied verbatim from the SMG source and a
re-sync must pin the upstream revision — the bufconn test proves self-consistency
with the generated stubs, not a match to the deployed engine.
---
 Makefile                                         |  2 +-
 pkg/adapters/engine/vllmengine/vllm_engine.pb.go | 16 ++++++++++++----
 pkg/adapters/engine/vllmengine/vllm_engine.proto | 16 ++++++++++++----
 .../engine/vllmengine/vllm_engine_grpc.pb.go     | 16 ++++++++++++----
 4 files changed, 37 insertions(+), 13 deletions(-)

diff --git a/Makefile b/Makefile
index 11ed5c61..47925c01 100644
--- a/Makefile
+++ b/Makefile
@@ -286,7 +286,7 @@ vulncheck: $(LOCALBIN) ## Scan dependencies + reachable code for known Go vulner
 COVER_MIN ?= 90
 COVER_PROFILE ?= cover.out
 COVER_PROFILE_LOGIC ?= cover.logic.out
-COVER_EXCLUDE := pkg/server/proto/|zz_generated|/cmd/|/hack/|pkg/testing/
+COVER_EXCLUDE := pkg/server/proto/|pkg/adapters/engine/vllmengine/|zz_generated|/cmd/|/hack/|pkg/testing/
 
 .PHONY: cover
 cover: ## Run tests with coverage and print the per-function report (logic packages, cross-package counted).
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
index 8d17fdd8..ceea3f35 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
@@ -11,10 +11,18 @@
 // engine does not expose).
 //
 // The package name MUST stay `vllm.grpc.engine` to match the upstream service
-// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
-// field numbers mirror upstream. Kept out of the `proto/` module on purpose —
-// it is a vendored external contract, not part of IC's own gRPC API, so it is
-// exempt from IC's buf-lint conventions.
+// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up; the
+// message names, field numbers, and types below are copied verbatim from the
+// source file so the wire encoding is byte-compatible. Kept out of the `proto/`
+// module on purpose — it is a vendored external contract, not part of IC's own
+// gRPC API, so it is exempt from IC's buf-lint conventions.
+//
+// Provenance: this is a hand-copied subset, so it is only as correct as the
+// upstream file it was taken from. When re-syncing against a new SMG release,
+// copy from the pinned revision and record it here (repo + commit), and confirm
+// the GetLoads request/response field numbers still match — the bufconn test
+// (grpc_loads_scraper_test.go) proves this file round-trips against its own
+// generated stubs, NOT that it matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.proto b/pkg/adapters/engine/vllmengine/vllm_engine.proto
index 0d34b4c9..831e9bb0 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine.proto
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.proto
@@ -7,10 +7,18 @@ syntax = "proto3";
 // engine does not expose).
 //
 // The package name MUST stay `vllm.grpc.engine` to match the upstream service
-// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
-// field numbers mirror upstream. Kept out of the `proto/` module on purpose —
-// it is a vendored external contract, not part of IC's own gRPC API, so it is
-// exempt from IC's buf-lint conventions.
+// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up; the
+// message names, field numbers, and types below are copied verbatim from the
+// source file so the wire encoding is byte-compatible. Kept out of the `proto/`
+// module on purpose — it is a vendored external contract, not part of IC's own
+// gRPC API, so it is exempt from IC's buf-lint conventions.
+//
+// Provenance: this is a hand-copied subset, so it is only as correct as the
+// upstream file it was taken from. When re-syncing against a new SMG release,
+// copy from the pinned revision and record it here (repo + commit), and confirm
+// the GetLoads request/response field numbers still match — the bufconn test
+// (grpc_loads_scraper_test.go) proves this file round-trips against its own
+// generated stubs, NOT that it matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
index 2d2401b2..017fa75a 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
+++ b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
@@ -11,10 +11,18 @@
 // engine does not expose).
 //
 // The package name MUST stay `vllm.grpc.engine` to match the upstream service
-// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up;
-// field numbers mirror upstream. Kept out of the `proto/` module on purpose —
-// it is a vendored external contract, not part of IC's own gRPC API, so it is
-// exempt from IC's buf-lint conventions.
+// so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up; the
+// message names, field numbers, and types below are copied verbatim from the
+// source file so the wire encoding is byte-compatible. Kept out of the `proto/`
+// module on purpose — it is a vendored external contract, not part of IC's own
+// gRPC API, so it is exempt from IC's buf-lint conventions.
+//
+// Provenance: this is a hand-copied subset, so it is only as correct as the
+// upstream file it was taken from. When re-syncing against a new SMG release,
+// copy from the pinned revision and record it here (repo + commit), and confirm
+// the GetLoads request/response field numbers still match — the bufconn test
+// (grpc_loads_scraper_test.go) proves this file round-trips against its own
+// generated stubs, NOT that it matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the

From 95e8fb9baf2a09cf445dac73636551f311e99d4a Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Thu, 23 Jul 2026 11:43:00 -0700
Subject: [PATCH 5/9] chore: pin verified SMG proto revision; extend pre-pr
 drift gate; fix stale doc note

- Record the exact machxai/smg commit (6d63ec8, 2026-07-09) the vendored proto
  was copied from, verified field-for-field against upstream.
- Add pkg/adapters/engine/vllmengine to the pre-pr generated-drift paths so the
  local gate inspects the stubs it now regenerates (matches the CI check).
- Reword the wiring-doc note: the enumerated stats flags are recognised by the
  binary; they are simply not set by the sidecar yet.
---
 Makefile                                           |  2 +-
 docs/design/kvevent-subscriber-wiring.md           |  8 ++++----
 pkg/adapters/engine/vllmengine/vllm_engine.pb.go   | 14 ++++++++------
 pkg/adapters/engine/vllmengine/vllm_engine.proto   | 14 ++++++++------
 .../engine/vllmengine/vllm_engine_grpc.pb.go       | 14 ++++++++------
 5 files changed, 29 insertions(+), 23 deletions(-)

diff --git a/Makefile b/Makefile
index 47925c01..d243b1dc 100644
--- a/Makefile
+++ b/Makefile
@@ -551,7 +551,7 @@ ci: verify-naming verify-no-internal-refs verify-syft-pin fmt-check vet ci-lint
 .PHONY: pre-pr
 pre-pr: ci ## Pre-PR gate: CI gate + generated-code drift check + sample admission check + review checklist.
 	@$(MAKE) --no-print-directory manifests generate proto-gen >/dev/null
-	@gen='config/crd config/rbac/role.yaml config/webhook/manifests.yaml api/v1alpha1/zz_generated.deepcopy.go pkg/server/proto'; \
+	@gen='config/crd config/rbac/role.yaml config/webhook/manifests.yaml api/v1alpha1/zz_generated.deepcopy.go pkg/server/proto pkg/adapters/engine/vllmengine'; \
 	if ! git diff --quiet -- $$gen; then \
 		echo "✗ generated-code drift — regenerate and commit these files:"; \
 		git --no-pager diff --name-only -- $$gen; \
diff --git a/docs/design/kvevent-subscriber-wiring.md b/docs/design/kvevent-subscriber-wiring.md
index d91599c1..6880cf5b 100644
--- a/docs/design/kvevent-subscriber-wiring.md
+++ b/docs/design/kvevent-subscriber-wiring.md
@@ -71,10 +71,10 @@ Concretely:
   `/metrics` (the `--engine-metrics-url` default) and emits `ReplicaStats` on its own
   cadence. Explicit stats-path flags — `--stats-interval`, the cache-size/ceiling
   hints, and `--engine-loads-grpc` (read load via the engine's GetLoads gRPC RPC
-  instead of HTTP; see the deferred note below) — are a follow-up enrichment the
-  sidecar does not set yet; passing a flag the binary doesn't recognise would crash
-  the sidecar at startup. No operator-supplied `--replica-id` / `--model-id` on the
-  demo path.
+  instead of HTTP; see the deferred note below) — are all recognised by the binary
+  but not yet set by the sidecar; wiring them (and deriving their values from the
+  CR) is the follow-up enrichment tracked below. No operator-supplied
+  `--replica-id` / `--model-id` on the demo path.
 
 ### Why this combination
 
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
index ceea3f35..dc616e3a 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
@@ -17,12 +17,14 @@
 // module on purpose — it is a vendored external contract, not part of IC's own
 // gRPC API, so it is exempt from IC's buf-lint conventions.
 //
-// Provenance: this is a hand-copied subset, so it is only as correct as the
-// upstream file it was taken from. When re-syncing against a new SMG release,
-// copy from the pinned revision and record it here (repo + commit), and confirm
-// the GetLoads request/response field numbers still match — the bufconn test
-// (grpc_loads_scraper_test.go) proves this file round-trips against its own
-// generated stubs, NOT that it matches the engine actually deployed.
+// Provenance: vendored from machxai/smg crates/grpc_client/proto/vllm_engine.proto
+// at commit 6d63ec83f052af42649a330d91f39aa56471aceb (2026-07-09). The GetLoads
+// RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages below
+// were verified field-for-field (numbers and types) against that revision. When
+// re-syncing against a newer SMG release, re-copy from the pinned commit, update
+// this line, and re-verify the field numbers. The bufconn test
+// (grpc_loads_scraper_test.go) only proves this file round-trips against its own
+// generated stubs — NOT that it matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.proto b/pkg/adapters/engine/vllmengine/vllm_engine.proto
index 831e9bb0..79a9e2fd 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine.proto
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.proto
@@ -13,12 +13,14 @@ syntax = "proto3";
 // module on purpose — it is a vendored external contract, not part of IC's own
 // gRPC API, so it is exempt from IC's buf-lint conventions.
 //
-// Provenance: this is a hand-copied subset, so it is only as correct as the
-// upstream file it was taken from. When re-syncing against a new SMG release,
-// copy from the pinned revision and record it here (repo + commit), and confirm
-// the GetLoads request/response field numbers still match — the bufconn test
-// (grpc_loads_scraper_test.go) proves this file round-trips against its own
-// generated stubs, NOT that it matches the engine actually deployed.
+// Provenance: vendored from machxai/smg crates/grpc_client/proto/vllm_engine.proto
+// at commit 6d63ec83f052af42649a330d91f39aa56471aceb (2026-07-09). The GetLoads
+// RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages below
+// were verified field-for-field (numbers and types) against that revision. When
+// re-syncing against a newer SMG release, re-copy from the pinned commit, update
+// this line, and re-verify the field numbers. The bufconn test
+// (grpc_loads_scraper_test.go) only proves this file round-trips against its own
+// generated stubs — NOT that it matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
index 017fa75a..bae98a5c 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
+++ b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
@@ -17,12 +17,14 @@
 // module on purpose — it is a vendored external contract, not part of IC's own
 // gRPC API, so it is exempt from IC's buf-lint conventions.
 //
-// Provenance: this is a hand-copied subset, so it is only as correct as the
-// upstream file it was taken from. When re-syncing against a new SMG release,
-// copy from the pinned revision and record it here (repo + commit), and confirm
-// the GetLoads request/response field numbers still match — the bufconn test
-// (grpc_loads_scraper_test.go) proves this file round-trips against its own
-// generated stubs, NOT that it matches the engine actually deployed.
+// Provenance: vendored from machxai/smg crates/grpc_client/proto/vllm_engine.proto
+// at commit 6d63ec83f052af42649a330d91f39aa56471aceb (2026-07-09). The GetLoads
+// RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages below
+// were verified field-for-field (numbers and types) against that revision. When
+// re-syncing against a newer SMG release, re-copy from the pinned commit, update
+// this line, and re-verify the field numbers. The bufconn test
+// (grpc_loads_scraper_test.go) only proves this file round-trips against its own
+// generated stubs — NOT that it matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the

From 139796a4271ce5dc990c7b03f0e72f1b51f24bd2 Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Thu, 23 Jul 2026 11:55:35 -0700
Subject: [PATCH 6/9] chore: log the load-scraper connection Close error
 instead of discarding it

---
 cmd/kvevent-subscriber/main.go | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/cmd/kvevent-subscriber/main.go b/cmd/kvevent-subscriber/main.go
index 4e3015a6..8cfb1d32 100644
--- a/cmd/kvevent-subscriber/main.go
+++ b/cmd/kvevent-subscriber/main.go
@@ -156,7 +156,11 @@ func main() {
 		os.Exit(1)
 	}
 	if scraperCloser != nil {
-		defer func() { _ = scraperCloser.Close() }()
+		defer func() {
+			if err := scraperCloser.Close(); err != nil {
+				logger.Warn("closing load-scraper connection", "err", err)
+			}
+		}()
 	}
 
 	loadSource, loadTarget := "HTTP /metrics scrape", *metricsURL

From a11b95253f4893672bcad81224fc497758b50715 Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Thu, 23 Jul 2026 12:01:25 -0700
Subject: [PATCH 7/9] fix: bound cache_memory_bytes conversion against int64
 overflow

float64(CacheSizeBytes) can round above the int64 range at extreme capacities,
so int64(usage*float64(cap)) could overflow to a negative. Clamp in float space
before the conversion and return the exact int64 capacity at the top end. Adds a
MaxInt64 boundary test.
---
 pkg/adapters/engine/grpc_loads_scraper.go     | 14 +++++++++++-
 .../engine/grpc_loads_scraper_test.go         | 22 +++++++++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/pkg/adapters/engine/grpc_loads_scraper.go b/pkg/adapters/engine/grpc_loads_scraper.go
index e9d07eab..ed14975d 100644
--- a/pkg/adapters/engine/grpc_loads_scraper.go
+++ b/pkg/adapters/engine/grpc_loads_scraper.go
@@ -119,7 +119,19 @@ func (s *GRPCLoadsScraper) Scrape(ctx context.Context) (*icpb.ReplicaStats, erro
 	pressure := pressureFrom(float64(running+waiting), s.cfg.MaxConcurrencyCeiling)
 	var cacheBytes int64
 	if s.cfg.CacheSizeBytes > 0 {
-		cacheBytes = int64(usage * float64(s.cfg.CacheSizeBytes))
+		// usage ∈ [0,1] ⇒ bytes ∈ [0, CacheSizeBytes]. Clamp in float space BEFORE
+		// the int64 conversion: float64(CacheSizeBytes) can round above the int64
+		// range at extreme capacities, so int64(usage*float64(CacheSizeBytes)) could
+		// overflow to a negative. Returning the exact int64 capacity at the top end
+		// sidesteps the lossy conversion.
+		switch b := usage * float64(s.cfg.CacheSizeBytes); {
+		case b <= 0:
+			cacheBytes = 0
+		case b >= float64(s.cfg.CacheSizeBytes):
+			cacheBytes = s.cfg.CacheSizeBytes
+		default:
+			cacheBytes = int64(b)
+		}
 	}
 	return &icpb.ReplicaStats{
 		CacheMemoryBytes: cacheBytes,
diff --git a/pkg/adapters/engine/grpc_loads_scraper_test.go b/pkg/adapters/engine/grpc_loads_scraper_test.go
index 160509bc..bd84b6ea 100644
--- a/pkg/adapters/engine/grpc_loads_scraper_test.go
+++ b/pkg/adapters/engine/grpc_loads_scraper_test.go
@@ -121,6 +121,28 @@ func TestGRPCLoadsScraperSanitizesOutOfRange(t *testing.T) {
 	}
 }
 
+func TestGRPCLoadsScraperCacheBytesNoOverflow(t *testing.T) {
+	// At an extreme capacity, float64(CacheSizeBytes) rounds above the int64 range,
+	// so a naive int64(usage*float64(cap)) overflows to a negative. cache_memory_bytes
+	// must stay in [0, CacheSizeBytes].
+	resp := &vpb.GetLoadsResponse{Loads: []*vpb.SchedulerLoad{
+		{TokenUsage: 1.0}, // full cache
+	}}
+	s := newGRPCLoadsScraperWithClient(&fakeLoadsClient{resp: resp},
+		GRPCLoadsScraperConfig{CacheSizeBytes: math.MaxInt64})
+
+	st, err := s.Scrape(context.Background())
+	if err != nil {
+		t.Fatalf("Scrape: %v", err)
+	}
+	if got := st.GetCacheMemoryBytes(); got < 0 || got > math.MaxInt64 {
+		t.Fatalf("cache_memory_bytes = %d, want within [0, MaxInt64] (no overflow)", got)
+	}
+	if got := st.GetCacheMemoryBytes(); got != math.MaxInt64 {
+		t.Errorf("cache_memory_bytes = %d, want MaxInt64 at usage=1.0 full capacity", got)
+	}
+}
+
 func TestGRPCLoadsScraperErrorIsSurfaced(t *testing.T) {
 	s := newGRPCLoadsScraperWithClient(&fakeLoadsClient{err: errors.New("unavailable")},
 		GRPCLoadsScraperConfig{Addr: "x:1"})

From 3e3a3b5ee8827aca8b122963bf790afd3736ba26 Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Thu, 23 Jul 2026 19:40:00 -0700
Subject: [PATCH 8/9] fix: drop tautological MaxInt64 comparison (staticcheck
 SA4003); make stats-path docs source-neutral
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The overflow test's 'got > math.MaxInt64' is always false for an int64 (SA4003,
failing ci-lint) — the meaningful guard is got < 0 (overflow wraps negative), and
the next assertion already pins the exact MaxInt64 value. Also update doc.go +
StatsReporter doc to describe the selectable load source (HTTP /metrics or GetLoads
gRPC), not HTTP-only (review nit).
---
 pkg/adapters/engine/doc.go                     | 10 ++++++----
 pkg/adapters/engine/grpc_loads_scraper_test.go |  4 ++--
 pkg/adapters/engine/stats_reporter.go          | 13 +++++++------
 3 files changed, 15 insertions(+), 12 deletions(-)

diff --git a/pkg/adapters/engine/doc.go b/pkg/adapters/engine/doc.go
index 234ca072..9b0537a6 100644
--- a/pkg/adapters/engine/doc.go
+++ b/pkg/adapters/engine/doc.go
@@ -5,10 +5,12 @@
 // Two independent paths share one gRPC client:
 //   - Event path: ZMQ → EventBatch → ReportCacheState (prefix adds) +
 //     PublishEvent (removals/clears), debounced on a short window.
-//   - Stats path: HTTP GET against the engine's Prometheus /metrics →
-//     MetricsScraper → StatsReporter → ReportCacheState (stats-only
-//     CacheStateUpdate populating cacheMemoryBytes / hitRate / pressure on its
-//     own cadence, default ~10s).
+//   - Stats path: engine load → a statsScraper → StatsReporter → ReportCacheState
+//     (stats-only CacheStateUpdate populating cacheMemoryBytes / hitRate / pressure
+//     on its own cadence, default ~10s). The load source is selectable: an HTTP GET
+//     against the engine's Prometheus /metrics (MetricsScraper, the default), or the
+//     VllmEngine GetLoads gRPC RPC (GRPCLoadsScraper) for gRPC engines that expose
+//     no /metrics endpoint.
 //
 // Metadata only — never KV tensors or prompt text. Fail-soft on both paths:
 // neither a ZMQ drop nor a scrape failure can stall the engine. The package is
diff --git a/pkg/adapters/engine/grpc_loads_scraper_test.go b/pkg/adapters/engine/grpc_loads_scraper_test.go
index bd84b6ea..86d05646 100644
--- a/pkg/adapters/engine/grpc_loads_scraper_test.go
+++ b/pkg/adapters/engine/grpc_loads_scraper_test.go
@@ -135,8 +135,8 @@ func TestGRPCLoadsScraperCacheBytesNoOverflow(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Scrape: %v", err)
 	}
-	if got := st.GetCacheMemoryBytes(); got < 0 || got > math.MaxInt64 {
-		t.Fatalf("cache_memory_bytes = %d, want within [0, MaxInt64] (no overflow)", got)
+	if got := st.GetCacheMemoryBytes(); got < 0 {
+		t.Fatalf("cache_memory_bytes = %d, want non-negative (overflow would wrap to negative)", got)
 	}
 	if got := st.GetCacheMemoryBytes(); got != math.MaxInt64 {
 		t.Errorf("cache_memory_bytes = %d, want MaxInt64 at usage=1.0 full capacity", got)
diff --git a/pkg/adapters/engine/stats_reporter.go b/pkg/adapters/engine/stats_reporter.go
index fc08d36c..4b0a484d 100644
--- a/pkg/adapters/engine/stats_reporter.go
+++ b/pkg/adapters/engine/stats_reporter.go
@@ -15,14 +15,15 @@ type statsScraper interface {
 	Scrape(ctx context.Context) (*icpb.ReplicaStats, error)
 }
 
-// StatsReporter periodically scrapes engine /metrics and emits a stats-only
-// CacheStateUpdate via ReportCacheState. It runs alongside the event Reporter
+// StatsReporter periodically pulls engine load from a statsScraper and emits a
+// stats-only CacheStateUpdate via ReportCacheState. The scraper is the load source
+// — MetricsScraper (HTTP /metrics) or GRPCLoadsScraper (the GetLoads gRPC RPC) — so
+// the reporter itself is source-neutral. It runs alongside the event Reporter
 // (different cadence, different data source) and shares the same gRPC client.
 //
-// Failure independence is load-bearing: a scrape failure (engine /metrics down,
-// HTTP timeout, parse error) logs and skips the tick — it never blocks the
-// event path or kills the subscriber. The two paths are independent failure
-// domains.
+// Failure independence is load-bearing: a scrape failure (engine load unavailable,
+// timeout, parse error) logs and skips the tick — it never blocks the event path or
+// kills the subscriber. The two paths are independent failure domains.
 type StatsReporter struct {
 	client     icpb.InferenceCacheClient
 	scraper    statsScraper

From 65b498405d3e19edadd87bd14d38fdfa64089071 Mon Sep 17 00:00:00 2001
From: Ed Sun <sunxu.edward@gmail.com>
Date: Fri, 24 Jul 2026 09:23:35 -0700
Subject: [PATCH 9/9] docs: describe the load source generically as the vLLM
 engine gRPC contract (drop SMG references)

inference-cache is a general repo, so its comments shouldn't name our downstream
gateway fork. Reframe the GetLoads load source + vendored proto as the vLLM engine
gRPC contract (vllm.grpc.engine.VllmEngine / vllm.entrypoints.grpc_server); drop
the machxai/smg provenance + the SMG-servicer version floor (keep the GetLoads +
vLLM >= 0.19 requirement). Regenerated the vendored stubs from the updated header.
---
 cmd/kvevent-subscriber/main.go                |  8 ++---
 cmd/kvevent-subscriber/main_test.go           |  2 +-
 docs/design/kvevent-subscriber-wiring.md      |  4 +--
 pkg/adapters/engine/grpc_loads_scraper.go     |  4 +--
 pkg/adapters/engine/vllmengine/generate.go    |  2 +-
 .../engine/vllmengine/vllm_engine.pb.go       | 31 +++++++++----------
 .../engine/vllmengine/vllm_engine.proto       | 31 +++++++++----------
 .../engine/vllmengine/vllm_engine_grpc.pb.go  | 31 +++++++++----------
 8 files changed, 55 insertions(+), 58 deletions(-)

diff --git a/cmd/kvevent-subscriber/main.go b/cmd/kvevent-subscriber/main.go
index 8cfb1d32..4a41c126 100644
--- a/cmd/kvevent-subscriber/main.go
+++ b/cmd/kvevent-subscriber/main.go
@@ -9,7 +9,7 @@
 //   - Stats path: engine load → derived ReplicaStats → ReportCacheState
 //     (stats-only CSU). The load source is selectable: HTTP GET against the
 //     engine's Prometheus /metrics (default), or the VllmEngine GetLoads gRPC
-//     RPC when --engine-loads-grpc is set (SMG gRPC engines expose no HTTP
+//     RPC when --engine-loads-grpc is set (vLLM gRPC engines expose no HTTP
 //     /metrics). Ticks on its own cadence (~10s), so the snapshot/CR status
 //     surface (cache_memory_bytes, hit_rate, pressure) lights up regardless of
 //     event rate.
@@ -55,7 +55,7 @@ type scraperParams struct {
 }
 
 // buildStatsScraper selects the engine load source: the GetLoads gRPC scraper
-// when loadsGRPC is non-empty (preferred for SMG gRPC engines, which expose no
+// when loadsGRPC is non-empty (preferred for vLLM gRPC engines, which expose no
 // HTTP /metrics), else the HTTP /metrics scraper. It returns the scraper, an
 // optional closer (the gRPC scraper owns a client conn — the HTTP one owns
 // nothing, so closer is nil), and an error only from gRPC dial setup.
@@ -92,7 +92,7 @@ func main() {
 		scheme             = flag.String("hash-scheme", "vllm", "engine prefix-hash scheme (required, non-empty)")
 		window             = flag.Duration("window", 100*time.Millisecond, "add-batching/debounce flush window")
 		metricsURL         = flag.String("engine-metrics-url", "http://127.0.0.1:8000/metrics", "engine Prometheus /metrics URL")
-		loadsGRPC          = flag.String("engine-loads-grpc", "", "engine VllmEngine gRPC address (host:port) to read load via the GetLoads RPC instead of scraping --engine-metrics-url. Preferred for SMG gRPC engines, which expose no HTTP /metrics. Empty = use the HTTP scrape.")
+		loadsGRPC          = flag.String("engine-loads-grpc", "", "engine VllmEngine gRPC address (host:port) to read load via the GetLoads RPC instead of scraping --engine-metrics-url. Preferred for vLLM gRPC engines, which expose no HTTP /metrics. Empty = use the HTTP scrape.")
 		statsInterval      = flag.Duration("stats-interval", 10*time.Second, "ReplicaStats scrape/emit cadence")
 		cacheSizeBytes     = flag.Int64("engine-cache-size-bytes", 0, "engine total KV-cache capacity in bytes (multiplied by usage_perc to derive cacheMemoryBytes; 0 emits cacheMemoryBytes=0)")
 		ceiling            = flag.Int("max-concurrency-ceiling", 256, "denominator for the pressure proxy = clamp01((num_requests_running+num_requests_waiting)/ceiling)")
@@ -140,7 +140,7 @@ func main() {
 		engine.WithIgnoreBlockRemoved(*ignoreBlockRemoved))
 	sub := engine.NewSubscriber(*endpoint, *topic, engine.WithSubscriberLogger(logger))
 
-	// Load source: gRPC GetLoads (preferred for SMG gRPC engines, which expose no
+	// Load source: gRPC GetLoads (preferred for vLLM gRPC engines, which expose no
 	// HTTP /metrics) when --engine-loads-grpc is set, else the HTTP /metrics scrape.
 	// Both satisfy the same statsScraper, so the StatsReporter is identical.
 	scraper, scraperCloser, serr := buildStatsScraper(scraperParams{
diff --git a/cmd/kvevent-subscriber/main_test.go b/cmd/kvevent-subscriber/main_test.go
index e162139e..a3b9e4a1 100644
--- a/cmd/kvevent-subscriber/main_test.go
+++ b/cmd/kvevent-subscriber/main_test.go
@@ -13,7 +13,7 @@ import (
 // non-empty --engine-loads-grpc must select the GetLoads gRPC scraper (and hand
 // back a closer for its conn), while an empty flag must preserve the HTTP
 // /metrics scraper (which owns no conn, so no closer). This is the crux of
-// Option B — the wrong branch silently leaves an SMG gRPC engine reporting no
+// Option B — the wrong branch silently leaves a vLLM engine served over gRPC reporting no
 // load — so it gets an explicit test.
 func TestBuildStatsScraperSelectsSource(t *testing.T) {
 	hc := &http.Client{}
diff --git a/docs/design/kvevent-subscriber-wiring.md b/docs/design/kvevent-subscriber-wiring.md
index 6880cf5b..c52154c4 100644
--- a/docs/design/kvevent-subscriber-wiring.md
+++ b/docs/design/kvevent-subscriber-wiring.md
@@ -107,10 +107,10 @@ Concretely:
   operator needs the knobs.
 * **Auto-injecting `--engine-loads-grpc` (GetLoads load source)** — the subscriber binary
   can now read engine load over the VllmEngine `GetLoads` gRPC RPC instead of scraping HTTP
-  `/metrics` (an SMG gRPC engine exposes no `/metrics` endpoint at all), selected by the
+  `/metrics` (a vLLM engine served over gRPC exposes no `/metrics` endpoint at all), selected by the
   `--engine-loads-grpc` flag. The sidecar renderer does **not** pass that flag yet, so
   auto-injected sidecars keep the HTTP default. Deferred deliberately: `GetLoads` requires a
-  newer engine floor (SMG gRPC servicer ≥ 0.5.2 / vLLM ≥ 0.19) than the engine currently
+  newer engine floor (a gRPC server that implements `GetLoads`; vLLM ≥ 0.19) than the engine currently
   deployed, so auto-wiring it now would point every sidecar at an unimplemented RPC and flip
   the load signal to "stale" (see the `StatsReporter` stale-escalation path). It gets wired
   alongside the stats-path knobs above once the engine floor moves; until then it is opt-in
diff --git a/pkg/adapters/engine/grpc_loads_scraper.go b/pkg/adapters/engine/grpc_loads_scraper.go
index ed14975d..eebd8efa 100644
--- a/pkg/adapters/engine/grpc_loads_scraper.go
+++ b/pkg/adapters/engine/grpc_loads_scraper.go
@@ -30,9 +30,9 @@ type GRPCLoadsScraperConfig struct {
 	Timeout time.Duration
 }
 
-// GRPCLoadsScraper reads engine load via the SMG engine's GetLoads gRPC RPC and
+// GRPCLoadsScraper reads engine load via the vLLM engine's GetLoads gRPC RPC and
 // projects it into a ReplicaStats — the gRPC-native alternative to scraping the
-// engine's HTTP /metrics. An SMG gRPC engine (vllm.entrypoints.grpc_server)
+// engine's HTTP /metrics. A vLLM engine served over gRPC (vllm.entrypoints.grpc_server)
 // exposes no HTTP metrics endpoint; live load is available only over GetLoads.
 // It implements the same statsScraper interface as MetricsScraper, so the
 // StatsReporter uses whichever is configured, unchanged.
diff --git a/pkg/adapters/engine/vllmengine/generate.go b/pkg/adapters/engine/vllmengine/generate.go
index 3af77307..c5a37b6d 100644
--- a/pkg/adapters/engine/vllmengine/generate.go
+++ b/pkg/adapters/engine/vllmengine/generate.go
@@ -1,4 +1,4 @@
-// Package vllmengine holds a minimal vendored subset of the SMG vLLM engine gRPC
+// Package vllmengine holds a minimal vendored subset of the vLLM engine gRPC
 // contract (GetLoads only), used by the kvevent-subscriber to read engine load
 // over gRPC instead of scraping HTTP /metrics. The .pb.go files are generated,
 // not hand-edited.
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
index dc616e3a..9d2d39c0 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.pb.go
@@ -4,27 +4,26 @@
 // 	protoc        v7.34.1
 // source: vllm_engine.proto
 
-// Minimal vendored subset of the SMG vLLM engine gRPC contract
-// (machxai/smg: crates/grpc_client/proto/vllm_engine.proto) — ONLY the GetLoads
-// RPC and its messages. The kvevent-subscriber uses this to read engine load
-// over gRPC as an alternative to scraping HTTP /metrics (which an SMG gRPC
-// engine does not expose).
+// Minimal vendored subset of the vLLM engine gRPC contract (the
+// `vllm.grpc.engine.VllmEngine` service exposed by vLLM's
+// `vllm.entrypoints.grpc_server`) — ONLY the GetLoads RPC and its messages. The
+// kvevent-subscriber uses this to read engine load over gRPC as an alternative to
+// scraping HTTP /metrics (which a vLLM engine served over gRPC does not expose).
 //
 // The package name MUST stay `vllm.grpc.engine` to match the upstream service
 // so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up; the
 // message names, field numbers, and types below are copied verbatim from the
-// source file so the wire encoding is byte-compatible. Kept out of the `proto/`
-// module on purpose — it is a vendored external contract, not part of IC's own
-// gRPC API, so it is exempt from IC's buf-lint conventions.
+// upstream contract so the wire encoding is byte-compatible. Kept out of the
+// `proto/` module on purpose — it is a vendored external contract, not part of
+// IC's own gRPC API, so it is exempt from IC's buf-lint conventions.
 //
-// Provenance: vendored from machxai/smg crates/grpc_client/proto/vllm_engine.proto
-// at commit 6d63ec83f052af42649a330d91f39aa56471aceb (2026-07-09). The GetLoads
-// RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages below
-// were verified field-for-field (numbers and types) against that revision. When
-// re-syncing against a newer SMG release, re-copy from the pinned commit, update
-// this line, and re-verify the field numbers. The bufconn test
-// (grpc_loads_scraper_test.go) only proves this file round-trips against its own
-// generated stubs — NOT that it matches the engine actually deployed.
+// Provenance: a hand-copied subset of the upstream vLLM engine gRPC contract; the
+// GetLoads RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages
+// below were verified field-for-field (numbers and types) against it. When
+// re-syncing against a newer engine release, re-copy from the upstream proto and
+// re-verify the field numbers. The bufconn test (grpc_loads_scraper_test.go) only
+// proves this file round-trips against its own generated stubs — NOT that it
+// matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine.proto b/pkg/adapters/engine/vllmengine/vllm_engine.proto
index 79a9e2fd..4e67829e 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine.proto
+++ b/pkg/adapters/engine/vllmengine/vllm_engine.proto
@@ -1,26 +1,25 @@
 syntax = "proto3";
 
-// Minimal vendored subset of the SMG vLLM engine gRPC contract
-// (machxai/smg: crates/grpc_client/proto/vllm_engine.proto) — ONLY the GetLoads
-// RPC and its messages. The kvevent-subscriber uses this to read engine load
-// over gRPC as an alternative to scraping HTTP /metrics (which an SMG gRPC
-// engine does not expose).
+// Minimal vendored subset of the vLLM engine gRPC contract (the
+// `vllm.grpc.engine.VllmEngine` service exposed by vLLM's
+// `vllm.entrypoints.grpc_server`) — ONLY the GetLoads RPC and its messages. The
+// kvevent-subscriber uses this to read engine load over gRPC as an alternative to
+// scraping HTTP /metrics (which a vLLM engine served over gRPC does not expose).
 //
 // The package name MUST stay `vllm.grpc.engine` to match the upstream service
 // so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up; the
 // message names, field numbers, and types below are copied verbatim from the
-// source file so the wire encoding is byte-compatible. Kept out of the `proto/`
-// module on purpose — it is a vendored external contract, not part of IC's own
-// gRPC API, so it is exempt from IC's buf-lint conventions.
+// upstream contract so the wire encoding is byte-compatible. Kept out of the
+// `proto/` module on purpose — it is a vendored external contract, not part of
+// IC's own gRPC API, so it is exempt from IC's buf-lint conventions.
 //
-// Provenance: vendored from machxai/smg crates/grpc_client/proto/vllm_engine.proto
-// at commit 6d63ec83f052af42649a330d91f39aa56471aceb (2026-07-09). The GetLoads
-// RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages below
-// were verified field-for-field (numbers and types) against that revision. When
-// re-syncing against a newer SMG release, re-copy from the pinned commit, update
-// this line, and re-verify the field numbers. The bufconn test
-// (grpc_loads_scraper_test.go) only proves this file round-trips against its own
-// generated stubs — NOT that it matches the engine actually deployed.
+// Provenance: a hand-copied subset of the upstream vLLM engine gRPC contract; the
+// GetLoads RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages
+// below were verified field-for-field (numbers and types) against it. When
+// re-syncing against a newer engine release, re-copy from the upstream proto and
+// re-verify the field numbers. The bufconn test (grpc_loads_scraper_test.go) only
+// proves this file round-trips against its own generated stubs — NOT that it
+// matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the
diff --git a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
index bae98a5c..cbc2b63d 100644
--- a/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
+++ b/pkg/adapters/engine/vllmengine/vllm_engine_grpc.pb.go
@@ -4,27 +4,26 @@
 // - protoc             v7.34.1
 // source: vllm_engine.proto
 
-// Minimal vendored subset of the SMG vLLM engine gRPC contract
-// (machxai/smg: crates/grpc_client/proto/vllm_engine.proto) — ONLY the GetLoads
-// RPC and its messages. The kvevent-subscriber uses this to read engine load
-// over gRPC as an alternative to scraping HTTP /metrics (which an SMG gRPC
-// engine does not expose).
+// Minimal vendored subset of the vLLM engine gRPC contract (the
+// `vllm.grpc.engine.VllmEngine` service exposed by vLLM's
+// `vllm.entrypoints.grpc_server`) — ONLY the GetLoads RPC and its messages. The
+// kvevent-subscriber uses this to read engine load over gRPC as an alternative to
+// scraping HTTP /metrics (which a vLLM engine served over gRPC does not expose).
 //
 // The package name MUST stay `vllm.grpc.engine` to match the upstream service
 // so the wire method path `/vllm.grpc.engine.VllmEngine/GetLoads` lines up; the
 // message names, field numbers, and types below are copied verbatim from the
-// source file so the wire encoding is byte-compatible. Kept out of the `proto/`
-// module on purpose — it is a vendored external contract, not part of IC's own
-// gRPC API, so it is exempt from IC's buf-lint conventions.
+// upstream contract so the wire encoding is byte-compatible. Kept out of the
+// `proto/` module on purpose — it is a vendored external contract, not part of
+// IC's own gRPC API, so it is exempt from IC's buf-lint conventions.
 //
-// Provenance: vendored from machxai/smg crates/grpc_client/proto/vllm_engine.proto
-// at commit 6d63ec83f052af42649a330d91f39aa56471aceb (2026-07-09). The GetLoads
-// RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages below
-// were verified field-for-field (numbers and types) against that revision. When
-// re-syncing against a newer SMG release, re-copy from the pinned commit, update
-// this line, and re-verify the field numbers. The bufconn test
-// (grpc_loads_scraper_test.go) only proves this file round-trips against its own
-// generated stubs — NOT that it matches the engine actually deployed.
+// Provenance: a hand-copied subset of the upstream vLLM engine gRPC contract; the
+// GetLoads RPC and the GetLoadsRequest / GetLoadsResponse / SchedulerLoad messages
+// below were verified field-for-field (numbers and types) against it. When
+// re-syncing against a newer engine release, re-copy from the upstream proto and
+// re-verify the field numbers. The bufconn test (grpc_loads_scraper_test.go) only
+// proves this file round-trips against its own generated stubs — NOT that it
+// matches the engine actually deployed.
 //
 // Regenerate the .pb.go stubs with `make proto-gen-vendored` (also run by `make
 // proto-gen`); CI diffs pkg/adapters/engine/vllmengine to catch drift, so the