feat(harness): add subject entrypoint overrides

cmd/harness/main.go

@@ -744,18 +744,25 @@ func versionCmd() *cobra.Command {
 	}
 }
 
-// applyCasePin overlays a case's subject_image/subject_version pin onto a
-// registry-resolved Subject. The global --image/--version CLI flags are applied
-// later in runner.applySubjectOverrides, so an explicit CLI flag still wins:
+// applyCasePin overlays a case's subject_image/subject_version/subject_entrypoint
+// pin onto a registry-resolved Subject. The global --image/--version CLI flags are
+// applied later in runner.applySubjectOverrides, so an explicit CLI flag still
+// wins:
 //
 //	CLI --image/--version  >  case subject_image/subject_version  >  registry default
+//
+// subject_entrypoint has no CLI counterpart: it is a property of what the case
+// tests (see TestCase.SubjectEntrypoint), not of which build is under test.
 func applyCasePin(s config.Subject, tc *config.TestCase) config.Subject {
 	if tc.SubjectImage != "" {
 		s = s.WithImage(tc.SubjectImage)
 	}
 	if tc.SubjectVersion != "" {
 		s = s.WithVersion(tc.SubjectVersion)
 	}
+	if len(tc.SubjectEntrypoint) > 0 {
+		s = s.WithEntrypoint(tc.SubjectEntrypoint)
+	}
 	return s
 }
 

internal/config/case.go

@@ -135,6 +135,17 @@ type TestCase struct {
 	SubjectImage   string `yaml:"subject_image"`
 	SubjectVersion string `yaml:"subject_version"`
 
+	// SubjectEntrypoint replaces the registry's entrypoint for this case, with
+	// the registry Command still appended as arguments. Same precedence and the
+	// same non-strict-decode tolerance as SubjectImage.
+	//
+	// Needed where the case is about the subject's own process lifecycle: the
+	// director self-update cases stop and restart the service they are updating,
+	// which ends the container when the service is PID 1. Pointing the entrypoint
+	// at a wrapper that runs the service as a CHILD keeps the container alive
+	// across that restart, so "the service came back" is observable at all.
+	SubjectEntrypoint []string `yaml:"subject_entrypoint"`
+
 	Subjects       []string                 `yaml:"subjects"`
 	Configurations map[string]Configuration `yaml:"configurations"`
 	Correctness    CorrectnessConfig        `yaml:"correctness"`
@@ -2063,7 +2074,13 @@ type EndpointSourceConfig struct {
 	// source must NOT deliver. The driver waits the settle window then asserts the
 	// receiver count stays <= ExpectMax. Proves auth/validation is load-bearing.
 	Reject bool `yaml:"reject"`
-	// ExpectMax is the ceiling for a reject case (default 0 — nothing delivered).
+	// ExpectMax is the ceiling. For a reject case it is the whole assertion
+	// (default 0 — nothing delivered). On a POSITIVE case it turns the count into a
+	// window: the driver waits past the floor and then fails on over-delivery,
+	// which is what a case needs when its records ARE its assertion (e.g. the
+	// director_update_* family, where one record stands for one proven outcome) —
+	// a floor alone cannot notice a subject that forwarded more than the case
+	// meant.
 	ExpectMax int `yaml:"expect_max"`
 }
 
@@ -2097,6 +2114,16 @@ func (tc *TestCase) validateEndpointSource() error {
 	if tc.EndpointSource.Reject && tc.EndpointSource.ExpectMax < 0 {
 		return fmt.Errorf("case %q: endpoint_source.expect_max must be >= 0", tc.Name)
 	}
+	// A ceiling below the floor can never be satisfied, so refuse it at load time
+	// rather than at the end of a ten-minute run. Every NONZERO value is checked, not
+	// just positive ones: zero is the documented "unset", and the runner enables the
+	// ceiling on the same `> 0` test, so `expect_max: -1` would otherwise pass
+	// validation AND silently disable the assertion it looks like it is making.
+	if !tc.EndpointSource.Reject && tc.EndpointSource.ExpectMax != 0 &&
+		tc.EndpointSource.ExpectMax < tc.EndpointSource.ExpectMin {
+		return fmt.Errorf("case %q: endpoint_source.expect_max (%d) must be >= expect_min (%d), or 0 for no ceiling",
+			tc.Name, tc.EndpointSource.ExpectMax, tc.EndpointSource.ExpectMin)
+	}
 	want := tc.EndpointSource.SenderContainerOrDefault()[len("bench-"):]
 	found := false
 	for _, e := range tc.Endpoints {

internal/config/subject.go

@@ -336,3 +336,11 @@ func (s Subject) WithImage(img string) Subject {
 	s.Image = img
 	return s
 }
+
+// WithEntrypoint returns a copy of the Subject with the entrypoint overridden.
+// Command is left alone, so the registry's arguments still reach the new
+// entrypoint — a wrapper script gets them as "$@".
+func (s Subject) WithEntrypoint(entrypoint []string) Subject {
+	s.Entrypoint = entrypoint
+	return s
+}

internal/runner/runner.go

@@ -6498,11 +6498,43 @@ func (r *Runner) runRedisSourceCorrectness(tc *config.TestCase, subject config.S
 	return r.saveAuxResult(tc, subject, configName, "redis source", startTime, finalCount, passed, errs, pubContainer, subjectContainer)
 }
 
+// endpointSourceCeilingQuietPeriod is how long a case that states an
+// endpoint_source.expect_max keeps watching after its floor is met, before the
+// ceiling is judged. Short on purpose: excess records arrive in a burst, so this
+// only has to outlast one, and a longer wait would tax every ceiling case.
+const endpointSourceCeilingQuietPeriod = 30 * time.Second
+
+// earliest returns whichever instant comes first — used to keep a step's own
+// deadline inside the run budget.
+func earliest(a, b time.Time) time.Time {
+	if a.Before(b) {
+		return a
+	}
+
+	return b
+}
+
+// budgetedTimeout clamps a step's timeout to the budget left before deadline, never
+// returning a non-positive value: a zero timeout would read as "no deadline" to the
+// callers here, which is the opposite of what a spent budget means.
+func budgetedTimeout(deadline time.Time, want time.Duration) time.Duration {
+	if remaining := time.Until(deadline); remaining < want {
+		if remaining < time.Second {
+			return time.Second
+		}
+
+		return remaining
+	}
+
+	return want
+}
+
 // runEndpointSourceCorrectness drives a director source the bench generator can't
 // feed (snmptrap, tftp, smtp, …) via a generic CLI-sender endpoint and counts at
 // the receiver. The sender is an `endpoints:` container in the case; the driver
 // just waits for the receiver to reach endpoint_source.expect_min. Tolerant of the
-// best-effort loss of UDP senders (expect_min, not exact).
+// best-effort loss of UDP senders (expect_min, not exact) unless the case also
+// states an expect_max, which makes the count a window — see the ceiling below.
 func (r *Runner) runEndpointSourceCorrectness(tc *config.TestCase, subject config.Subject) (results.RunResult, error) {
 	configName := r.opts.ConfigName
 	subject = r.applySubjectOverrides(subject)
@@ -6598,6 +6630,67 @@ func (r *Runner) runEndpointSourceCorrectness(tc *config.TestCase, subject confi
 		fmt.Printf("  source delivered %d records (≥ %d) ✓\n", got, expectMin)
 	}
 
+	// Ceiling, when the case states one. A floor alone cannot fail a case whose
+	// records ARE its assertion: a subject that forwards more than it was asked to
+	// (a filter that silently stopped matching, a duplicated route, an evidence line
+	// the case did not intend) clears any expect_min. Cases that encode content in
+	// the count set expect_max == expect_min and become exact.
+	//
+	// The extra wait is deliberately short rather than the full settle window: a
+	// runaway forwarder delivers its excess within a second or two, and charging
+	// every ceiling case the whole window would add minutes per case for nothing.
+	if reached && es.ExpectMax > 0 {
+		// Bounded by the run budget: the floor above is capped at runDeadline, and a
+		// case that reached it late must not push the run past --timeout on the way
+		// out. No budget left is a timeout, not a pass.
+		quietUntil := time.Now().Add(endpointSourceCeilingQuietPeriod)
+		if quietUntil.After(runDeadline) {
+			quietUntil = runDeadline
+		}
+
+		remaining := time.Until(quietUntil)
+		if remaining <= 0 {
+			// No budget means the window was never watched, and an unwatched window is
+			// not a pass — the same rule the metrics-outage branch below applies. A case
+			// that reaches its floor in the last seconds of --timeout fails as
+			// unverified, which is the honest outcome for a case that asked for a ceiling.
+			errs = append(errs, fmt.Sprintf("ceiling not verified: the run budget (%s) ran out before the ≤ %d window could be observed", r.opts.Timeout, es.ExpectMax))
+			passed := len(errs) == 0
+
+			return r.saveAuxResult(tc, subject, configName, "endpoint source", startTime, finalCount, passed, errs, senderContainer, subjectContainer)
+		}
+
+		if err := sleepCtx(r.ctx, remaining); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
+
+		// The ceiling needs a FRESH count, and it must fail when it cannot get one:
+		// ccfWaitStable reports 0 or the last value it saw when the metrics endpoint
+		// stops answering, and treating that as "within the ceiling" would pass a case
+		// whose delivery nobody measured.
+		//
+		// Both this settle and the authoritative sample below stay inside runDeadline:
+		// the whole point of capping the quiet period is lost if the two steps after it
+		// can push the run past --timeout on their own.
+		if stable := r.ccfWaitStable(metricsPort, earliest(time.Now().Add(10*time.Second), runDeadline)); stable > finalCount {
+			finalCount = stable
+		}
+
+		rm, err := r.queryReceiverMetrics(metricsPort, budgetedTimeout(runDeadline, 5*time.Second))
+		switch {
+		case err != nil:
+			errs = append(errs, fmt.Sprintf("ceiling not verified: the receiver metrics endpoint stopped answering (%v) — a ≤ %d assertion cannot be made on an unmeasured window", err, es.ExpectMax))
+		case rm.LinesReceived > int64(es.ExpectMax):
+			finalCount = rm.LinesReceived
+			errs = append(errs, fmt.Sprintf("over-delivery: %d records from the source (expected ≤ %d) — more arrived than the case asserts", finalCount, es.ExpectMax))
+		default:
+			if rm.LinesReceived > finalCount {
+				finalCount = rm.LinesReceived
+			}
+			fmt.Printf("  source stayed within the ceiling: %d ≤ %d ✓\n", finalCount, es.ExpectMax)
+		}
+	}
+
 	passed := len(errs) == 0
 	return r.saveAuxResult(tc, subject, configName, "endpoint source", startTime, finalCount, passed, errs, senderContainer, subjectContainer)
 }