[review-fix 1] Preserve baseline selections across windows

- Replace expired one-shot contexts during admission so the next baseline window cannot lose an identical selection before flush cleanup.
- Remove the E2E package-install fallback and verify the local test service before generating traffic with the existing Python runtime.
- Use explicit Dynamic Tests state and consistent empty metric tags for clearer test and telemetry semantics.

Source: review feedback and full-diff self-review
Validation: Bazel pathteststore tests and netpath dynamic-tests build

Author: AlexandreYang

comp/networkpath/npcollector/impl/config_test.go

@@ -28,7 +28,7 @@ func TestNetworkPathCollectorEnabled(t *testing.T) {
 	}
 	assert.True(t, config.networkPathCollectorEnabled())
 
-	config.dynamicTestsState = 0
+	config.dynamicTestsState = npconfig.DynamicTestsOff
 	assert.False(t, config.networkPathCollectorEnabled())
 
 	config.netflowMonitoringEnabled = true

comp/networkpath/npcollector/impl/pathteststore/pathteststore.go

@@ -164,7 +164,7 @@ func (f *Store) Flush() []*PathtestContext {
 				f.statsdClient.Incr(networkPathStoreMetricPrefix+"pathtest_never_run", []string{}, 1) //nolint:errcheck
 			}
 			if ptConfigCtx.Pathtest.DynamicTestProfile == payload.DynamicTestProfileBaseline {
-				f.statsdClient.Incr(networkPathStoreMetricPrefix+"baseline_expired", nil, 1) //nolint:errcheck
+				f.statsdClient.Incr(networkPathStoreMetricPrefix+"baseline_expired", []string{}, 1) //nolint:errcheck
 			}
 			continue
 		}
@@ -178,7 +178,7 @@ func (f *Store) Flush() []*PathtestContext {
 		pathtestsToFlush = append(pathtestsToFlush, ptConfigCtx.snapshot())
 		if ptConfigCtx.Pathtest.OneShot {
 			if ptConfigCtx.Pathtest.DynamicTestProfile == payload.DynamicTestProfileBaseline {
-				f.statsdClient.Incr(networkPathStoreMetricPrefix+"baseline_dispatched", nil, 1) //nolint:errcheck
+				f.statsdClient.Incr(networkPathStoreMetricPrefix+"baseline_dispatched", []string{}, 1) //nolint:errcheck
 			}
 			delete(f.contexts, key)
 			continue
@@ -205,16 +205,30 @@ func (f *Store) Add(pathtestToAdd *common.Pathtest) {
 	if ok {
 		if pathtestToAdd.OneShot {
 			// A baseline slot is consumed before admission. Deduplication must not
-			// extend or recur an already selected one-shot context.
+			// extend or recur a live one-shot context. Replace an expired context
+			// immediately so a flush racing with the next window cannot discard the
+			// new selection along with the old one.
+			now := f.timeNowFn()
+			expired := pathtestCtx.runUntil.Before(now) || (pathtestCtx.Pathtest.OneShot && pathtestCtx.runUntil.Equal(now))
+			if !expired {
+				return
+			}
+			delete(f.contexts, hash)
+			if pathtestCtx.lastFlushTime.IsZero() {
+				f.statsdClient.Incr(networkPathStoreMetricPrefix+"pathtest_never_run", []string{}, 1) //nolint:errcheck
+			}
+			if pathtestCtx.Pathtest.DynamicTestProfile == payload.DynamicTestProfileBaseline {
+				f.statsdClient.Incr(networkPathStoreMetricPrefix+"baseline_expired", []string{}, 1) //nolint:errcheck
+			}
+		} else {
+			// Refresh attribution from the latest admission without creating a second
+			// context for the same path.
+			pathtestCtx.Pathtest.TestConfigID = pathtestToAdd.TestConfigID
+			pathtestCtx.Pathtest.TestConfigSource = pathtestToAdd.TestConfigSource
+			pathtestCtx.Pathtest.Tags = slices.Clone(pathtestToAdd.Tags)
+			pathtestCtx.runUntil = f.timeNowFn().Add(f.config.TTL)
 			return
 		}
-		// Refresh attribution from the latest admission without creating a second
-		// context for the same path.
-		pathtestCtx.Pathtest.TestConfigID = pathtestToAdd.TestConfigID
-		pathtestCtx.Pathtest.TestConfigSource = pathtestToAdd.TestConfigSource
-		pathtestCtx.Pathtest.Tags = slices.Clone(pathtestToAdd.Tags)
-		pathtestCtx.runUntil = f.timeNowFn().Add(f.config.TTL)
-		return
 	}
 
 	if len(f.contexts) >= f.config.ContextsLimit {

comp/networkpath/npcollector/impl/pathteststore/pathteststore_test.go

@@ -224,6 +224,24 @@ func TestPathtestStoreOneShotExpiresBeforeDispatch(t *testing.T) {
 	assert.Equal(t, int64(1), stats.GetCountSummaries()[networkPathStoreMetricPrefix+"baseline_expired"].Sum)
 }
 
+func TestPathtestStoreReplacesExpiredOneShotBeforeFlush(t *testing.T) {
+	now := time.Date(2026, 8, 15, 12, 0, 0, 0, time.UTC)
+	stats := &teststatsd.Client{}
+	store := NewPathtestStore(Config{ContextsLimit: 10}, logmock.New(t), stats, func() time.Time { return now })
+	store.Add(&common.Pathtest{Hostname: "baseline", OneShot: true, ExecutionDeadline: now.Add(30 * time.Minute), DynamicTestProfile: payload.DynamicTestProfileBaseline})
+	now = now.Add(30 * time.Minute)
+	newDeadline := now.Add(30 * time.Minute)
+	store.Add(&common.Pathtest{Hostname: "baseline", OneShot: true, ExecutionDeadline: newDeadline, DynamicTestProfile: payload.DynamicTestProfileBaseline})
+
+	replacement := store.contexts[(&common.Pathtest{Hostname: "baseline"}).GetHash()]
+	require.NotNil(t, replacement)
+	assert.Equal(t, newDeadline, replacement.runUntil)
+	assert.Len(t, store.Flush(), 1, "the next window's replacement must remain dispatchable")
+	assert.Zero(t, store.GetContextsCount())
+	assert.Equal(t, int64(1), stats.GetCountSummaries()[networkPathStoreMetricPrefix+"baseline_expired"].Sum)
+	assert.Equal(t, int64(1), stats.GetCountSummaries()[networkPathStoreMetricPrefix+"baseline_dispatched"].Sum)
+}
+
 func TestPathtestStoreRecurringDispatchesAtTTLBoundary(t *testing.T) {
 	now := time.Date(2026, 8, 15, 12, 0, 0, 0, time.UTC)
 	store := NewPathtestStore(Config{

test/new-e2e/tests/netpath/dynamic-tests/host_traffic_dynamic_path_common_test.go

@@ -40,8 +40,8 @@ const (
 	hostTrafficDNSPIDPath         = "/tmp/host_traffic_dns.pid"
 	hostTrafficResolverBackupPath = "/tmp/host_traffic_resolv.conf.backup"
 	hostTrafficResolverLinkPath   = "/tmp/host_traffic_resolv.conf.link"
-	hostTrafficCurlLogPath        = "/tmp/host_traffic_dynamic_path_curl.log"
-	hostTrafficCurlPIDPath        = "/tmp/host_traffic_dynamic_path_curl.pid"
+	hostTrafficGeneratorLogPath   = "/tmp/host_traffic_dynamic_path_generator.log"
+	hostTrafficGeneratorPIDPath   = "/tmp/host_traffic_dynamic_path_generator.pid"
 	hostTrafficHTTPBinComposeYAML = `version: '3.9'
 services:
   httpbin:
@@ -124,8 +124,8 @@ func hostTrafficHTTPBinCompose() docker.ComposeInlineManifest {
 }
 
 func (s *hostTrafficDynamicPathBaseSuite) setupHostTraffic() {
-	s.ensureCurlInstalled()
 	s.startHostTrafficDNSServer()
+	s.assertHostTrafficServiceReady()
 	s.configureAgentResolver()
 	s.assertHostTrafficDomainResolves()
 }
@@ -139,13 +139,18 @@ func (s *hostTrafficDynamicPathBaseSuite) tearDownHostTraffic() {
 func (s *hostTrafficDynamicPathBaseSuite) AfterTest(suiteName, testName string) {
 	if s.T().Failed() {
 		s.logRemoteFile(s.Env().HTTPBinHost, hostTrafficDNSLogPath)
-		s.logRemoteFile(s.Env().RemoteHost, hostTrafficCurlLogPath)
+		s.logRemoteFile(s.Env().RemoteHost, hostTrafficGeneratorLogPath)
 	}
 	s.BaseSuite.AfterTest(suiteName, testName)
 }
 
-func (s *hostTrafficDynamicPathBaseSuite) ensureCurlInstalled() {
-	s.Env().RemoteHost.MustExecute("if ! command -v curl >/dev/null 2>&1; then sudo apt-get update && sudo apt-get install -y curl; fi")
+func (s *hostTrafficDynamicPathBaseSuite) assertHostTrafficServiceReady() {
+	s.Env().HTTPBinHost.MustExecute(`i=0; while [ "$i" -lt 30 ]; do
+  python3 -c 'import urllib.request; urllib.request.urlopen("http://127.0.0.1/", timeout=5).read()' && exit 0
+  sleep 2
+  i=$((i+1))
+done
+exit 1`)
 }
 
 func (s *hostTrafficDynamicPathBaseSuite) startHostTrafficDNSServer() {
@@ -234,23 +239,23 @@ func (s *hostTrafficDynamicPathBaseSuite) assertHostTrafficDomainResolves() {
 
 func (s *hostTrafficDynamicPathBaseSuite) startHostTrafficGenerator(duration time.Duration) {
 	seconds := int(duration.Seconds())
-	trafficCommand := fmt.Sprintf(
-		"i=0; while [ \"$i\" -lt %d ]; do curl -4 -fsS --max-time 5 %s >/dev/null || true; sleep 2; i=$((i+2)); done",
+	trafficScript := fmt.Sprintf(
+		"import time, urllib.request\nurl = %q\nend = time.monotonic() + %d\nwhile time.monotonic() < end:\n    try:\n        urllib.request.urlopen(url, timeout=5).read()\n    except Exception:\n        pass\n    time.sleep(2)\n",
+		hostTrafficURL(hostTrafficRemoteConfigDomain),
 		seconds,
-		shellQuote(hostTrafficURL(hostTrafficRemoteConfigDomain)),
 	)
-	s.Env().RemoteHost.MustExecute(fmt.Sprintf("nohup sh -c %s >%s 2>&1 & echo $! >%s",
-		shellQuote(trafficCommand),
-		shellQuote(hostTrafficCurlLogPath),
-		shellQuote(hostTrafficCurlPIDPath),
+	s.Env().RemoteHost.MustExecute(fmt.Sprintf("nohup python3 -c %s >%s 2>&1 & echo $! >%s",
+		shellQuote(trafficScript),
+		shellQuote(hostTrafficGeneratorLogPath),
+		shellQuote(hostTrafficGeneratorPIDPath),
 	))
 }
 
 func (s *hostTrafficDynamicPathBaseSuite) stopHostTrafficGenerator() {
 	if s.Env().RemoteHost == nil {
 		return
 	}
-	_, err := s.Env().RemoteHost.Execute(fmt.Sprintf(`if [ -f %s ]; then kill "$(cat %s)" || true; fi`, shellQuote(hostTrafficCurlPIDPath), shellQuote(hostTrafficCurlPIDPath)))
+	_, err := s.Env().RemoteHost.Execute(fmt.Sprintf(`if [ -f %s ]; then kill "$(cat %s)" || true; fi`, shellQuote(hostTrafficGeneratorPIDPath), shellQuote(hostTrafficGeneratorPIDPath)))
 	if err != nil {
 		s.T().Logf("failed to stop host traffic generator: %v", err)
 	}