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,13 @@ 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.
+	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)",
+			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,18 @@ 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
+
 // 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 +6605,29 @@ 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 {
+		if err := sleepCtx(r.ctx, endpointSourceCeilingQuietPeriod); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
+		if stable := r.ccfWaitStable(metricsPort, time.Now().Add(10*time.Second)); stable > finalCount {
+			finalCount = stable
+		}
+		if finalCount > int64(es.ExpectMax) {
+			errs = append(errs, fmt.Sprintf("over-delivery: %d records from the source (expected ≤ %d) — more arrived than the case asserts", finalCount, es.ExpectMax))
+		} else {
+			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)
 }