Fix comments and add missing tests.

Signed-off-by: Mohammad <mohammad.nassar@ibm.com>

Author: Mohammad-nassar10

pkg/epp/framework/plugins/requestcontrol/dataproducer/README.md

@@ -35,7 +35,7 @@ The framework resolves a DAG from each plugin's `Produces` and `Consumes` declar
 - `inflight-load-producer` **optionally** consumes `PrefixCacheMatchInfo` from an approx or precise prefix producer; prefix-discounting is applied automatically when the attribute is present.
 - `p2p-source-producer` **requires** `PrefixCacheMatchInfo` from a prefix producer; set `prefixMatchInfoProducerName` to select a non-default producer instance. Omitting it binds the default key, which auto-wires the approximate producer (no error) — set it explicitly for precise-only deployments. Set `prefillProfileName` to match a renamed `disagg-profile-handler` prefill profile.
 - `predicted-latency-producer` **optionally** consumes `PrefixCacheMatchInfo`; set `prefixMatchInfoProducerName` in its config to the name of the prefix producer instance.
-- `latency-observer-producer` **requires** `InFlightLoad`, so `inflight-load-producer` is ordered ahead of it and auto-created when absent. It must itself be listed under `dataLayer.sources`, which is what drives its recompute; auto-creation alone leaves every endpoint reading cold.
+- `latency-observer-producer` **requires** `InFlightLoad`, so `inflight-load-producer` is ordered ahead of it and auto-created when absent. It must itself be listed under `dataLayer.sources`; auto-creation from `latency-observation-scorer`'s required data key only wires the attribute, not the periodic tick that publishes it. See the [producer README](latencyobserver/README.md#configuration).
 
 ## Related documentation
 

pkg/epp/framework/plugins/requestcontrol/dataproducer/latencyobserver/README.md

@@ -30,7 +30,7 @@ It builds on the request-control hooks and publishes statistics as a datalayer a
 
 | hook | what it does |
 |---|---|
-| `Produce` | Update inflight load at the time of current request |
+| `Produce` | Read and pin each candidate's in-flight load, before this request joins it |
 | `PreRequest` | Record the endpoint decision to enable capturing endpoint statistics |
 | `ResponseBody`, first chunk | Capture TTFT and append it to that endpoint's observation window |
 | `Dispatch` | Every `intervalDuration`, recompute the percentiles and publish |

pkg/epp/framework/plugins/requestcontrol/dataproducer/latencyobserver/flush_test.go

@@ -0,0 +1,140 @@
+/*
+Copyright 2026 The llm-d Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package latencyobserver
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// testEndpointID is the namespaced name of the endpoint the tests observe.
+const testEndpointID = "default/a"
+
+// flushConfig is tuned for tests: a short bucket so the floor path runs, and a
+// low minRequests so a handful of observations is enough to be trusted.
+func flushConfig() Config {
+	cfg := DefaultConfig
+	cfg.WindowSize, cfg.MaxRequests, cfg.MinRequests = 64, 16, 4
+	cfg.BucketDuration, cfg.BucketHistorySize = "1s", 4
+	return cfg
+}
+
+// flushAt recomputes the endpoint's snapshot at a controlled time. Dispatch
+// reads time.Now(), so tests drive publish directly to steer the bucket clock.
+func flushAt(p *Observer, now time.Time) {
+	p.publish(context.Background(), testEndpointID, p.stateFor(testEndpointID), now)
+}
+
+func TestResolveConfig(t *testing.T) {
+	resolved, err := DefaultConfig.resolve()
+	require.NoError(t, err)
+	assert.Equal(t, time.Second, resolved.interval)
+	assert.Equal(t, 100, resolved.maxRequests)
+	assert.InDelta(t, 0.25, resolved.lowPercentile, 1e-9, "percentiles resolve to fractions")
+
+	invalid := map[string]func(*Config){
+		"zero window":              func(c *Config) { c.WindowSize = 0 },
+		"maxRequests above window": func(c *Config) { c.MaxRequests = c.WindowSize + 1 },
+		"zero minRequests":         func(c *Config) { c.MinRequests = 0 },
+		"bucket history below two": func(c *Config) { c.BucketHistorySize = 1 },
+		"percentiles inverted":     func(c *Config) { c.LowPercentile, c.TypicalPercentile = 50, 25 },
+		"unparsable duration":      func(c *Config) { c.IntervalDuration = "soon" },
+		"non-positive duration":    func(c *Config) { c.BucketDuration = "-1m" },
+	}
+	for name, mutate := range invalid {
+		t.Run(name, func(t *testing.T) {
+			cfg := DefaultConfig
+			mutate(&cfg)
+			_, err := cfg.resolve()
+			require.Error(t, err)
+		})
+	}
+}
+
+func TestFlush(t *testing.T) {
+	now := time.Now()
+
+	t.Run("publishes the load anchors from the short window", func(t *testing.T) {
+		p := newObserverWithConfig(t, flushConfig())
+		// TTFT rises with the in-flight count it was dispatched at, which is the
+		// relationship the scorer's curve captures.
+		for i := range 10 {
+			p.record(testEndpointID, 0.1+float64(i)*0.1, int64(i), now.Add(time.Duration(i)*time.Millisecond))
+		}
+
+		flushAt(p, now.Add(time.Second))
+
+		s := p.stateFor(testEndpointID).published.Load()
+		require.NotNil(t, s)
+		assert.Equal(t, 10, s.RecentRequestCount)
+		assert.Equal(t, 4, s.CalibrationThreshold)
+		assert.Less(t, s.LowLoadTTFT, s.TypicalLoadTTFT, "P25 must sit below P50")
+		assert.Less(t, s.InflightAtLowLoad, s.InflightAtTypicalLoad, "the faster band must be less loaded")
+	})
+
+	t.Run("the floor arrives only once a bucket closes", func(t *testing.T) {
+		p := newObserverWithConfig(t, flushConfig())
+		for i := range 10 {
+			ttft := 1.0
+			if i < 2 {
+				ttft = 0.1 // a fast tail the floor must follow
+			}
+			p.record(testEndpointID, ttft, 1, now.Add(time.Duration(i)*time.Millisecond))
+		}
+
+		flushAt(p, now) // starts the bucket clock
+		assert.Zero(t, p.stateFor(testEndpointID).published.Load().FloorTTFT)
+
+		// One bucket later exactly: the bucket window looks back bucketDuration,
+		// so flushing further out would close an empty bucket.
+		flushAt(p, now.Add(time.Second))
+		s := p.stateFor(testEndpointID).published.Load()
+		assert.Greater(t, s.FloorTTFT, 0.0)
+		assert.Less(t, s.FloorTTFT, 1.0, "tracks the fast requests, not the bulk")
+	})
+
+	t.Run("Floor withholds the value below minRequests", func(t *testing.T) {
+		p := newObserverWithConfig(t, flushConfig()) // minRequests 4
+		for i := range 2 {
+			p.record(testEndpointID, 0.1, 1, now.Add(time.Duration(i)*time.Millisecond))
+		}
+		flushAt(p, now)
+		flushAt(p, now.Add(time.Second))
+
+		s := p.stateFor(testEndpointID).published.Load()
+		assert.Greater(t, s.FloorTTFT, 0.0, "the raw floor is computed")
+		assert.Zero(t, s.Floor(), "but Floor() withholds it")
+	})
+}
+
+// The datalayer drives the recompute: one Dispatch per endpoint publishes that
+// endpoint's snapshot, and the plugin accepts no extractors.
+func TestDispatch(t *testing.T) {
+	ctx := context.Background()
+	p := newObserverWithConfig(t, flushConfig())
+	p.record(testEndpointID, 0.3, 1, time.Now())
+
+	require.NoError(t, p.Dispatch(ctx, newDataEndpoint()))
+	assert.NotNil(t, p.stateFor(testEndpointID).published.Load())
+	assert.Equal(t, time.Second, p.Interval(), "the configured interval is the dispatch cadence")
+	require.NoError(t, p.Dispatch(ctx, nil), "a nil endpoint is a no-op")
+	assert.Error(t, p.AppendExtractor(p), "this dispatcher sources nothing")
+}

pkg/epp/framework/plugins/requestcontrol/dataproducer/latencyobserver/hooks.go

@@ -18,6 +18,7 @@ package latencyobserver
 
 import (
 	"context"
+	"maps"
 	"time"
 
 	"sigs.k8s.io/controller-runtime/pkg/log"
@@ -53,13 +54,21 @@ func (d *dispatchInfo) Clone() fwkplugin.StateData {
 	return &cp
 }
 
-// Produces declares the percentile snapshot this producer publishes, and the
-// request-scoped key Produce pins: the endpoint scope rejects undeclared writes.
+// inflightAtDispatch is what each candidate was carrying when Produce ran, keyed
+// by endpoint ID. It lives in PluginState, not on the endpoint: the data-parallel
+// and disaggregated profile handlers rebuild endpoints with empty attribute maps.
+type inflightAtDispatch map[string]int64
+
+// Clone implements [fwkplugin.StateData].
+func (m inflightAtDispatch) Clone() fwkplugin.StateData {
+	cp := make(inflightAtDispatch, len(m))
+	maps.Copy(cp, m)
+	return cp
+}
+
+// Produces declares the percentile snapshot this producer publishes.
 func (p *Observer) Produces() map[fwkplugin.DataKey]any {
-	return map[fwkplugin.DataKey]any{
-		p.percentilesDataKey:        attrlatency.TTFTPercentiles{},
-		p.inflightAtDispatchDataKey: attrconcurrency.InFlightLoad{},
-	}
+	return map[fwkplugin.DataKey]any{p.percentilesDataKey: attrlatency.TTFTPercentiles{}}
 }
 
 // Consumes declares the in-flight load as Required, so the DAG runs
@@ -70,19 +79,22 @@ func (p *Observer) Consumes() fwkplugin.DataDependencies {
 	}
 }
 
-// Produce pins each candidate's in-flight load under a request-scoped key, so
-// PreRequest sees the load carried before this request landed. Reading it in
-// PreRequest instead would race: inflight-load-producer increments its counters
-// there too, and hook order between plugins is undefined. Produce is DAG-ordered.
-func (p *Observer) Produce(_ context.Context, _ *fwksched.InferenceRequest, endpoints []fwksched.Endpoint) error {
+// Produce pins every candidate's in-flight load, so PreRequest sees the load
+// carried before this request landed. Reading it there instead would race:
+// inflight-load-producer increments its counters in its own PreRequest, and hook
+// order between plugins is undefined. Produce is DAG-ordered.
+func (p *Observer) Produce(_ context.Context, request *fwksched.InferenceRequest, endpoints []fwksched.Endpoint) error {
+	if request == nil || request.RequestID == "" || p.PluginState == nil {
+		return nil
+	}
+	pinned := make(inflightAtDispatch, len(endpoints))
 	for _, endpoint := range endpoints {
 		if endpoint == nil || endpoint.GetMetadata() == nil {
 			continue
 		}
-		endpoint.Put(p.inflightAtDispatchDataKey, &attrconcurrency.InFlightLoad{
-			Requests: readInFlightRequests(endpoint, p.inFlightLoadDataKey),
-		})
+		pinned[endpoint.GetMetadata().ID.String()] = readInFlightRequests(endpoint, p.inFlightLoadDataKey)
 	}
+	p.PluginState.Write(request.RequestID, inflightStateKey, pinned)
 	return nil
 }
 
@@ -108,14 +120,24 @@ func (p *Observer) PreRequest(ctx context.Context, request *fwksched.InferenceRe
 		log.FromContext(ctx).V(logutil.DEBUG).Info("Skipping TTFT tracking: no request ID or no primary target")
 		return nil
 	}
+	endpointID := endpoint.GetMetadata().ID.String()
 	p.PluginState.Write(request.RequestID, dispatchStateKey, &dispatchInfo{
-		endpointID:   endpoint.GetMetadata().ID.String(),
-		inflight:     readInFlightRequests(endpoint, p.inflightAtDispatchDataKey),
+		endpointID:   endpointID,
+		inflight:     p.pinnedInflight(request.RequestID, endpointID),
 		dispatchedAt: time.Now(),
 	})
 	return nil
 }
 
+// pinnedInflight returns what Produce recorded, or zero if it never ran.
+func (p *Observer) pinnedInflight(requestID, endpointID string) int64 {
+	pinned, err := fwkplugin.ReadPluginStateKey[inflightAtDispatch](p.PluginState, requestID, inflightStateKey)
+	if err != nil {
+		return 0
+	}
+	return pinned[endpointID]
+}
+
 // primaryTarget returns the endpoint the primary profile selected, or nil. TTFT
 // belongs to whichever endpoint produced the first token.
 func primaryTarget(result *fwksched.SchedulingResult) fwksched.Endpoint {