From 7560fa5471d50af641db2042d30ae3ad43c8702e Mon Sep 17 00:00:00 2001
From: Eren Aslan <16862833+erenaslandev@users.noreply.github.com>
Date: Fri, 28 Aug 2026 23:42:26 +0300
Subject: [PATCH 1/3] feat(harness): S3 seed/verdict knobs for windowed and
 poison-object poll cases

- aws.seed_objects[]: delay_seconds (real LastModified separation
  between
  seed groups inside the LocalStack init hook), extension (key suffix,
  default .log; a .gz name over a plain-text body yields a poison object
  whose decode fails), and the $TODAY prefix token substituted in Go at
  render time (UTC YYYY/MM/DD).
- LocalStack healthcheck retries scale with the total seed delay so a
  delayed init hook is not reported unhealthy.
- Subject config templating: {{@.T0Plus N@}} renders T0+N seconds as
  RFC3339 (negative N allowed); device-side % tokens pass through.
- correctness.forbidden_substring: the verdict fails if any received
  line
  carries the token (receiver counts hits and keeps samples);
  correctness.max_received: upper bound on lines received.
---
 containers/receiver/main.go                  |  59 +++++++--
 internal/config/case.go                      |  14 +++
 internal/config/cloud.go                     |  46 ++++++-
 internal/config/cloud_test.go                |  40 ++++++
 internal/orchestrator/awsinit.go             |  20 ++-
 internal/orchestrator/awsinit_test.go        |  59 +++++++++
 internal/orchestrator/compose_render_test.go | 121 +++++++++++++++++++
 internal/orchestrator/docker.go              |  93 +++++++++-----
 internal/runner/runner.go                    |  15 +++
 9 files changed, 426 insertions(+), 41 deletions(-)
 create mode 100644 internal/orchestrator/awsinit_test.go

diff --git a/containers/receiver/main.go b/containers/receiver/main.go
index c218f2d..d53a50a 100644
--- a/containers/receiver/main.go
+++ b/containers/receiver/main.go
@@ -27,11 +27,12 @@ type config struct {
 	Timeout     time.Duration
 
 	// Correctness validation (all optional, off by default)
-	ValidateDedup     bool
-	ValidateContent   bool   // O(1) per line, no heap map — safe for high-volume tests
-	ExpectedLines     int64  // 0 = don't check
-	RequiredSubstring string // empty = don't check; protocol-agnostic decode check
-	ValidateJSON      bool   // every emitted line must parse as JSON
+	ValidateDedup      bool
+	ValidateContent    bool   // O(1) per line, no heap map — safe for high-volume tests
+	ExpectedLines      int64  // 0 = don't check
+	RequiredSubstring  string // empty = don't check; protocol-agnostic decode check
+	ForbiddenSubstring string // empty = don't check; the verdict fails if any line contains it
+	ValidateJSON       bool   // every emitted line must parse as JSON
 
 	// RecordArrivalTimes, when true, has the receiver capture the
 	// wall-clock arrival nanosecond of every record into an in-memory
@@ -207,6 +208,12 @@ type validator struct {
 	missingSubstr     atomic.Int64
 	missingSubstrSamp []string
 
+	// Forbidden-substring check: no line may contain the configured token
+	// (e.g. the marker of objects a lookback / end_date / prefix window must
+	// exclude). Counts + up to 10 sample violators.
+	forbiddenHit     atomic.Int64
+	forbiddenHitSamp []string
+
 	// JSON-validity check: every emitted line must parse as a JSON value.
 	// Enabled by ValidateJSON. Without this, a JSON test could be passed
 	// by a subject that emits the right line count of garbage. Counts +
@@ -288,6 +295,23 @@ func (v *validator) recordLine(line []byte, cfg config) {
 		}
 	}
 
+	// Negative content check: a line carrying the forbidden token proves an
+	// exclusion (lookback / end_date / prefix pruning) was NOT enforced.
+	if cfg.ForbiddenSubstring != "" {
+		if bytes.Contains(line, []byte(cfg.ForbiddenSubstring)) {
+			v.forbiddenHit.Add(1)
+			v.mu.Lock()
+			if len(v.forbiddenHitSamp) < 10 {
+				s := string(line)
+				if len(s) > 120 {
+					s = s[:120] + "…"
+				}
+				v.forbiddenHitSamp = append(v.forbiddenHitSamp, s)
+			}
+			v.mu.Unlock()
+		}
+	}
+
 	// JSON-validity check: every line must parse as JSON. json.Valid is
 	// streaming-friendly (no allocation for the parse tree) so it's cheap
 	// enough to run per-line at multi-million lines/s.
@@ -484,6 +508,18 @@ func (v *validator) validate(cfg config, totalLines int64) (bool, []string) {
 		}
 	}
 
+	// Forbidden-substring — no emitted line may carry the configured token.
+	if cfg.ForbiddenSubstring != "" {
+		if m := v.forbiddenHit.Load(); m > 0 {
+			msg := fmt.Sprintf("forbidden substring %q present in %d line(s)", cfg.ForbiddenSubstring, m)
+			if len(v.forbiddenHitSamp) > 0 {
+				msg += "; samples: " + strings.Join(v.forbiddenHitSamp, " | ")
+			}
+			errors = append(errors, msg)
+			passed = false
+		}
+	}
+
 	return passed, errors
 }
 
@@ -547,7 +583,7 @@ func main() {
 		otlpShard := cnt.newShard()
 		onLine := func(line []byte) {
 			otlpShard.recordLine(int64(len(line)) + 1)
-			if cfg.ValidateDedup || cfg.ValidateContent || cfg.RequiredSubstring != "" {
+			if cfg.ValidateDedup || cfg.ValidateContent || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != "" {
 				val.recordLine(line, cfg)
 			}
 		}
@@ -671,7 +707,7 @@ func handleConn(conn net.Conn, shard *connStats, val *validator, cfg config) {
 	// time, inflating the receive window when the sender idles before closing.
 	scanner := bufio.NewScanner(&stampingReader{r: conn, shard: shard})
 	scanner.Buffer(make([]byte, 1024*1024), 1024*1024)
-	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
+	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != ""
 	// Pass the scanner's internal slice directly. shard.recordLine + validator
 	// only read len() and copy via `string(line)` / hash when they need to
 	// keep bytes — they never retain the slice past return. Skipping the
@@ -695,7 +731,7 @@ func handleConn(conn net.Conn, shard *connStats, val *validator, cfg config) {
 func receiveFile(cfg config, cnt *counters, val *validator) error {
 	shard := cnt.newShard()
 	defer shard.finish()
-	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
+	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != ""
 
 	deadline := time.Now().Add(cfg.Timeout + 5*time.Minute) // generous for file tests
 	f, err := os.Open(cfg.Listen)
@@ -742,7 +778,7 @@ func receiveHTTP(cfg config, cnt *counters, val *validator) error {
 	// per-shard ceiling so contention here doesn't matter, and a single
 	// shard keeps the per-request bookkeeping trivial.
 	httpShard := cnt.newShard()
-	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
+	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != ""
 	mux := http.NewServeMux()
 
 	// Generic POST handler — counts every line in the body
@@ -831,7 +867,7 @@ func receiveHTTP(cfg config, cnt *counters, val *validator) error {
 // POST. It handles both plain line-per-request and ES /_bulk NDJSON format.
 func startHTTPDataEndpoint(port string, cnt *counters, val *validator, cfg config) error {
 	httpShard := cnt.newShard()
-	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
+	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != ""
 	lineCallback := func(line []byte) {
 		httpShard.recordLine(int64(len(line)) + 1)
 		if needsValidation {
@@ -942,7 +978,7 @@ func serveMetrics(port string, cnt *counters, val *validator, cfg config) {
 			"last_received_ns":  lastNs,
 		}
 		// Include correctness data if validation is enabled
-		if cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.ExpectedLines > 0 || cfg.RequiredSubstring != "" {
+		if cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.ExpectedLines > 0 || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != "" {
 			passed, errors := val.validate(cfg, totalLines)
 			resp["passed"] = passed
 			if len(errors) > 0 {
@@ -1020,6 +1056,7 @@ func loadConfig() config {
 		ValidateContent:    getEnvBool("RECEIVER_VALIDATE_CONTENT", false),
 		ExpectedLines:      int64(getEnvInt("RECEIVER_EXPECTED_LINES", 0)),
 		RequiredSubstring:  getEnv("RECEIVER_REQUIRED_SUBSTRING", ""),
+		ForbiddenSubstring: getEnv("RECEIVER_FORBIDDEN_SUBSTRING", ""),
 		ValidateJSON:       getEnvBool("RECEIVER_VALIDATE_JSON", false),
 		RecordArrivalTimes: getEnvBool("RECEIVER_RECORD_ARRIVAL_TIMES", false),
 	}
diff --git a/internal/config/case.go b/internal/config/case.go
index dcfab7e..d427cc3 100644
--- a/internal/config/case.go
+++ b/internal/config/case.go
@@ -1204,6 +1204,9 @@ func (tc *TestCase) Validate() error {
 	if tc.Correctness.MaxOverDeliveryPct < 0 {
 		return fmt.Errorf("case %q: max_overdelivery_pct must be non-negative, got %.2f", tc.Name, tc.Correctness.MaxOverDeliveryPct)
 	}
+	if tc.Correctness.MaxReceived < 0 {
+		return fmt.Errorf("case %q: max_received must be non-negative (0 = disabled), got %d", tc.Name, tc.Correctness.MaxReceived)
+	}
 	if err := tc.validateVault(); err != nil {
 		return err
 	}
@@ -2641,6 +2644,11 @@ type CorrectnessConfig struct {
 	// successful decode is proven by the presence of a value the generator
 	// embedded in every record. Empty = check disabled.
 	RequiredSubstring string `yaml:"required_substring"`
+	// ForbiddenSubstring is the inverse of RequiredSubstring: the verdict FAILS
+	// if ANY received line contains it. It proves an exclusion (a lookback or
+	// end_date cutoff, prefix pruning) is enforced, not just that inclusion
+	// works. Generic — any correctness case may use it. Empty = disabled.
+	ForbiddenSubstring string `yaml:"forbidden_substring"`
 	// ValidateJSON, when true, requires every emitted line to parse as a
 	// JSON object. Without this, a subject can pass a JSON-shape test by
 	// truncating to a matching line count or by re-emitting binary garbage —
@@ -2664,6 +2672,12 @@ type CorrectnessConfig struct {
 	// "LinesReceived >= MinReceived" (default 1) AND the receiver didn't flag a
 	// content failure. Ignored when the case has a generator.
 	MinReceived int64 `yaml:"min_received"`
+	// MaxReceived, when > 0, fails the verdict if LinesReceived exceeds it —
+	// independent of expect_failure/loss/over-delivery logic. Use it to prove
+	// an upper bound (e.g. an end_date cutoff) where the exact count is
+	// timing-sensitive and only a ceiling is safe to assert. Generic — any
+	// correctness case may use it. 0 = disabled.
+	MaxReceived int64 `yaml:"max_received"`
 
 	// DrainSeconds extends how long the harness waits for backlog to
 	// arrive after the generator(s) stop. The case still finishes early if
diff --git a/internal/config/cloud.go b/internal/config/cloud.go
index 1090526..1e77bd0 100644
--- a/internal/config/cloud.go
+++ b/internal/config/cloud.go
@@ -101,8 +101,46 @@ type AWSSeedObjects struct {
 	// against the cloud-name charset so it cannot inject into the init shell
 	// script.
 	Marker string `yaml:"marker"`
+	// DelaySeconds sleeps that many seconds inside the LocalStack init script
+	// immediately before this group's upload loop, giving consecutive seed
+	// groups real wall-clock separation in their S3 LastModified (LocalStack
+	// cannot backdate objects, so back-to-back groups otherwise share one
+	// second). 0 = no delay (default, existing behaviour).
+	DelaySeconds int `yaml:"delay_seconds"`
+	// Extension is the key suffix of every seeded object (default ".log").
+	// Bodies are always plain text, so a compressed-looking extension such
+	// as ".gz" deliberately produces a poison object whose decode fails —
+	// the fixture for retry-cap / skip-on-error cases. Must match
+	// SeedExtensionPattern.
+	Extension string `yaml:"extension"`
 }
 
+// SeedExtensionPattern bounds seed_objects[].extension: a leading dot then
+// lowercase alphanumerics/dots, so it can neither inject into the init shell
+// script nor produce a key the subject's extension dispatch cannot classify.
+var SeedExtensionPattern = regexp.MustCompile(`^\.[a-z0-9.]{1,15}$`)
+
+// TotalSeedDelaySeconds sums seed_objects[].delay_seconds — the wall-clock
+// the LocalStack init hook sleeps before it can report "completed", which the
+// compose healthcheck budget must cover.
+func (a *AWSConfig) TotalSeedDelaySeconds() int {
+	if a == nil {
+		return 0
+	}
+	total := 0
+	for _, so := range a.SeedObjects {
+		if so.DelaySeconds > 0 {
+			total += so.DelaySeconds
+		}
+	}
+	return total
+}
+
+// SeedTodayToken is the one prefix placeholder the harness substitutes itself
+// (UTC YYYY/MM/DD at render time) so a case can seed "today's" partition. It is
+// stripped before the charset check and never reaches the shell unexpanded.
+const SeedTodayToken = "$TODAY"
+
 // AWSStream declares a Kinesis stream created at init.
 type AWSStream struct {
 	Name string `yaml:"name"`
@@ -449,10 +487,13 @@ func (tc *TestCase) validateAWS() error {
 			return fmt.Errorf("case %q: seed_objects references undeclared bucket %q", tc.Name, so.Bucket)
 		}
 		if so.Prefix != "" {
-			if err := validateCloudName(tc.Name, "aws seed prefix", so.Prefix); err != nil {
+			if err := validateCloudName(tc.Name, "aws seed prefix", strings.ReplaceAll(so.Prefix, SeedTodayToken, "")); err != nil {
 				return err
 			}
 		}
+		if so.DelaySeconds < 0 {
+			return fmt.Errorf("case %q: seed_objects for bucket %q delay_seconds must be non-negative, got %d", tc.Name, so.Bucket, so.DelaySeconds)
+		}
 		if so.Marker != "" {
 			if err := validateCloudName(tc.Name, "aws seed marker", so.Marker); err != nil {
 				return err
@@ -464,6 +505,9 @@ func (tc *TestCase) validateAWS() error {
 		if so.Lines <= 0 {
 			return fmt.Errorf("case %q: seed_objects for bucket %q requires lines > 0, got %d", tc.Name, so.Bucket, so.Lines)
 		}
+		if so.Extension != "" && !SeedExtensionPattern.MatchString(so.Extension) {
+			return fmt.Errorf("case %q: seed_objects for bucket %q extension %q must match %s", tc.Name, so.Bucket, so.Extension, SeedExtensionPattern)
+		}
 	}
 	return nil
 }
diff --git a/internal/config/cloud_test.go b/internal/config/cloud_test.go
index 8089061..9c9f6e0 100644
--- a/internal/config/cloud_test.go
+++ b/internal/config/cloud_test.go
@@ -33,6 +33,46 @@ func TestValidateCloud(t *testing.T) {
 				Generator: GeneratorConfig{Mode: "s3", Target: "http://localstack:4566"},
 			},
 		},
+		{
+			name: "seed group with negative delay",
+			tc: TestCase{
+				Name: "c",
+				AWS: &AWSConfig{
+					Buckets:     []string{"bench-in"},
+					SeedObjects: []AWSSeedObjects{{Bucket: "bench-in", Objects: 1, Lines: 1, DelaySeconds: -1}},
+				},
+			},
+			wantErr: "delay_seconds must be non-negative",
+		},
+		{
+			name: "seed prefix with $TODAY token and delay",
+			tc: TestCase{
+				Name: "c",
+				AWS: &AWSConfig{
+					Buckets:     []string{"bench-in"},
+					SeedObjects: []AWSSeedObjects{{Bucket: "bench-in", Prefix: "logs/$TODAY/", Objects: 1, Lines: 1, DelaySeconds: 45}},
+				},
+			},
+		},
+		{
+			name: "seed prefix with other shell token rejected",
+			tc: TestCase{
+				Name: "c",
+				AWS: &AWSConfig{
+					Buckets:     []string{"bench-in"},
+					SeedObjects: []AWSSeedObjects{{Bucket: "bench-in", Prefix: "logs/$HOME/", Objects: 1, Lines: 1}},
+				},
+			},
+			wantErr: "aws seed prefix",
+		},
+		{
+			name: "negative max_received rejected",
+			tc: TestCase{
+				Name:        "c",
+				Correctness: CorrectnessConfig{MaxReceived: -1},
+			},
+			wantErr: "max_received must be non-negative",
+		},
 		{
 			name: "azure block with container",
 			tc: TestCase{
diff --git a/internal/orchestrator/awsinit.go b/internal/orchestrator/awsinit.go
index 19b5e54..c574dd7 100644
--- a/internal/orchestrator/awsinit.go
+++ b/internal/orchestrator/awsinit.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/VirtualMetric/PipeBench/internal/config"
 )
@@ -17,7 +18,10 @@ import (
 // Every name interpolated here passed config.Validate()'s charset check
 // ([a-zA-Z0-9._/-]), so single-quoting is belt-and-braces, not the only
 // injection defense.
-func writeAWSInit(path string, aws *config.AWSConfig) error {
+//
+// now is the harness render instant; it resolves the $TODAY seed-prefix token
+// (UTC YYYY/MM/DD) so it shares T0 with configTemplateContext.T0Plus.
+func writeAWSInit(path string, aws *config.AWSConfig, now time.Time) error {
 	var b strings.Builder
 	b.WriteString("#!/bin/sh\n")
 	b.WriteString("# Auto-generated by harness — creates bench resources inside LocalStack.\n")
@@ -61,17 +65,29 @@ func writeAWSInit(path string, aws *config.AWSConfig) error {
 		if prefix == "" {
 			prefix = "seed/"
 		}
+		// $TODAY is substituted here, in Go — the upload command below is
+		// single-quoted, so the shell would never expand it.
+		prefix = strings.ReplaceAll(prefix, config.SeedTodayToken, now.UTC().Format("2006/01/02"))
 		marker := so.Marker
 		if marker == "" {
 			marker = "SEED"
 		}
+		ext := so.Extension
+		if ext == "" {
+			ext = ".log"
+		}
+		// A delayed group sleeps before uploading so its LastModified is
+		// separated from the previous group by real wall-clock time.
+		if so.DelaySeconds > 0 {
+			fmt.Fprintf(&b, "sleep %d\n", so.DelaySeconds)
+		}
 		// Build each object body with awk (busybox ships it), then upload —
 		// so the objects exist before LocalStack reports init "completed" and
 		// the subject's depends_on gate releases. All values are int-formatted
 		// or charset-validated, so single-quoting is defense in depth.
 		fmt.Fprintf(&b, "i=0; while [ \"$i\" -lt %d ]; do\n", so.Objects)
 		fmt.Fprintf(&b, "  awk -v o=\"$i\" 'BEGIN{for(l=0;l<%d;l++) printf \"%s-OBJ%%d-LINE%%d\\n\", o, l}' > /tmp/pb-seed-obj\n", so.Lines, marker)
-		fmt.Fprintf(&b, "  awslocal s3 cp /tmp/pb-seed-obj 's3://%s/%sobj-'\"$i\"'.log'\n", so.Bucket, prefix)
+		fmt.Fprintf(&b, "  awslocal s3 cp /tmp/pb-seed-obj 's3://%s/%sobj-'\"$i\"'%s'\n", so.Bucket, prefix, ext)
 		b.WriteString("  i=$((i+1))\n")
 		b.WriteString("done\n")
 	}
diff --git a/internal/orchestrator/awsinit_test.go b/internal/orchestrator/awsinit_test.go
new file mode 100644
index 0000000..85c7541
--- /dev/null
+++ b/internal/orchestrator/awsinit_test.go
@@ -0,0 +1,59 @@
+package orchestrator
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/VirtualMetric/PipeBench/internal/config"
+)
+
+// TestWriteAWSInitSeedDelayAndToday verifies that a seed group's delay_seconds
+// renders a sleep before its upload loop and that the $TODAY prefix token is
+// substituted (UTC YYYY/MM/DD) by the harness, not left for the shell.
+func TestWriteAWSInitSeedDelayAndToday(t *testing.T) {
+	tmp := t.TempDir()
+	path := filepath.Join(tmp, "aws-init.sh")
+	now := time.Date(2026, 8, 27, 12, 0, 0, 0, time.UTC)
+	aws := &config.AWSConfig{
+		Buckets: []string{"bench-in"},
+		SeedObjects: []config.AWSSeedObjects{
+			{Bucket: "bench-in", Prefix: "old/", Objects: 2, Lines: 3, Marker: "OLD"},
+			{Bucket: "bench-in", Prefix: "logs/$TODAY/", Objects: 2, Lines: 3, Marker: "NEW", DelaySeconds: 45},
+			{Bucket: "bench-in", Prefix: "poison/", Objects: 1, Lines: 3, Marker: "BAD", Extension: ".gz"},
+		},
+	}
+	if err := writeAWSInit(path, aws, now); err != nil {
+		t.Fatalf("writeAWSInit: %v", err)
+	}
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := string(data)
+	if !strings.Contains(out, "s3://bench-in/logs/2026/08/27/obj-") {
+		t.Fatalf("$TODAY not substituted:\n%s", out)
+	}
+	if strings.Contains(out, "$TODAY") {
+		t.Fatalf("$TODAY leaked into the script:\n%s", out)
+	}
+	sleepIdx := strings.Index(out, "sleep 45\n")
+	newIdx := strings.Index(out, "logs/2026/08/27/")
+	oldIdx := strings.Index(out, "s3://bench-in/old/")
+	if sleepIdx < 0 || newIdx < 0 || oldIdx < 0 || !(oldIdx < sleepIdx && sleepIdx < newIdx) {
+		t.Fatalf("expected old loop, then sleep 45, then new loop; got old=%d sleep=%d new=%d\n%s", oldIdx, sleepIdx, newIdx, out)
+	}
+	if strings.Count(out, "sleep ") != 1 {
+		t.Fatalf("only the delayed group may sleep:\n%s", out)
+	}
+	// The default extension stays .log; an explicit extension renames the key
+	// only (the body is still plain text, i.e. a poison object for .gz).
+	if !strings.Contains(out, "s3://bench-in/old/obj-'\"$i\"'.log'") {
+		t.Fatalf("default .log extension missing:\n%s", out)
+	}
+	if !strings.Contains(out, "s3://bench-in/poison/obj-'\"$i\"'.gz'") {
+		t.Fatalf("explicit .gz extension missing:\n%s", out)
+	}
+}
diff --git a/internal/orchestrator/compose_render_test.go b/internal/orchestrator/compose_render_test.go
index bcbd8ea..46053dd 100644
--- a/internal/orchestrator/compose_render_test.go
+++ b/internal/orchestrator/compose_render_test.go
@@ -5,6 +5,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/VirtualMetric/PipeBench/internal/config"
 	"gopkg.in/yaml.v3"
@@ -970,3 +971,123 @@ func mustNotContain(t *testing.T, hay, needle string) {
 		t.Errorf("rendered compose unexpectedly contains %q:\n%s", needle, hay)
 	}
 }
+
+// TestComposeRendersForbiddenSubstring verifies the negative-content
+// assertion reaches the receiver container as RECEIVER_FORBIDDEN_SUBSTRING.
+func TestComposeRendersForbiddenSubstring(t *testing.T) {
+	tc := &config.TestCase{
+		Name:     "forbid",
+		Type:     "correctness",
+		Duration: "10s",
+		Warmup:   "5s",
+		Generator: config.GeneratorConfig{
+			Mode:     "tcp",
+			Target:   "subject:9000",
+			Rate:     100,
+			LineSize: 256,
+			Format:   "raw",
+		},
+		Receiver: config.ReceiverConfig{Mode: "tcp", Listen: ":9001"},
+		Correctness: config.CorrectnessConfig{
+			RequiredSubstring:  "NEW-",
+			ForbiddenSubstring: "OLD-",
+		},
+	}
+	subj := config.Subject{Name: "vmetric", Image: "vmetric/director", Version: "2.0.3", ConfigPath: "/config.yml"}
+	tmp := t.TempDir()
+	composePath := filepath.Join(tmp, "compose.yaml")
+	cfg := RunConfig{
+		TestCase:         tc,
+		Subject:          subj,
+		ConfigName:       "default",
+		ConfigSrcPath:    composePath,
+		TmpDir:           tmp,
+		GeneratorImage:   "img-gen",
+		ReceiverImage:    "img-recv",
+		CollectorImage:   "img-coll",
+		ReceiverHostPort: 19001,
+	}
+	if err := writeCompose(composePath, cfg); err != nil {
+		t.Fatalf("writeCompose: %v", err)
+	}
+	data, err := os.ReadFile(composePath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := string(data)
+	mustContain(t, out, "RECEIVER_REQUIRED_SUBSTRING: \"NEW-\"")
+	mustContain(t, out, "RECEIVER_FORBIDDEN_SUBSTRING: \"OLD-\"")
+}
+
+// TestRenderSubjectConfigT0Plus verifies a subject config can anchor an
+// absolute timestamp to the harness render instant via {{@.T0Plus N@}}.
+func TestRenderSubjectConfigT0Plus(t *testing.T) {
+	tmp := t.TempDir()
+	src := filepath.Join(tmp, "vmetric.yml")
+	if err := os.WriteFile(src, []byte("properties:\n  end_date: \"{{@.T0Plus 90@}}\"\n  start_date: \"{{@.T0Plus -86400@}}\"\n  prefix: \"logs/%Y/\"\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	outDir := filepath.Join(tmp, "out")
+	if err := os.Mkdir(outDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	ctx := configTemplateContext{CPUs: 4, T0: time.Date(2026, 8, 27, 12, 0, 0, 0, time.UTC)}
+	outPath, err := renderSubjectConfig(src, outDir, ctx)
+	if err != nil {
+		t.Fatalf("renderSubjectConfig: %v", err)
+	}
+	data, err := os.ReadFile(outPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := string(data)
+	mustContain(t, out, "end_date: \"2026-08-27T12:01:30Z\"")
+	// A negative offset (start_date one day before render) must render too.
+	mustContain(t, out, "start_date: \"2026-08-26T12:00:00Z\"")
+	// vmetric's own strftime prefix tokens must pass through untouched.
+	mustContain(t, out, "prefix: \"logs/%Y/\"")
+}
+
+// TestComposeScalesLocalStackHealthRetries verifies that seed_objects delays
+// extend the LocalStack healthcheck budget (the init hook sleeps before
+// "completed" flips true) instead of tripping the fixed 40-retry ceiling.
+func TestComposeScalesLocalStackHealthRetries(t *testing.T) {
+	render := func(t *testing.T, aws *config.AWSConfig) string {
+		tc := &config.TestCase{
+			Name:     "seeded",
+			Type:     "correctness",
+			Duration: "10s",
+			Warmup:   "5s",
+			AWS:      aws,
+			Receiver: config.ReceiverConfig{Mode: "tcp", Listen: ":9001"},
+		}
+		subj := config.Subject{Name: "vmetric", Image: "vmetric/director", Version: "2.0.3", ConfigPath: "/config.yml"}
+		tmp := t.TempDir()
+		composePath := filepath.Join(tmp, "compose.yaml")
+		cfg := RunConfig{
+			TestCase: tc, Subject: subj, ConfigName: "default", ConfigSrcPath: composePath, TmpDir: tmp,
+			GeneratorImage: "img-gen", ReceiverImage: "img-recv", CollectorImage: "img-coll", ReceiverHostPort: 19001,
+		}
+		if err := writeCompose(composePath, cfg); err != nil {
+			t.Fatalf("writeCompose: %v", err)
+		}
+		data, err := os.ReadFile(composePath)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return string(data)
+	}
+
+	plain := render(t, &config.AWSConfig{Buckets: []string{"bench-in"}})
+	mustContain(t, plain, "retries: 40\n")
+
+	delayed := render(t, &config.AWSConfig{
+		Buckets: []string{"bench-in"},
+		SeedObjects: []config.AWSSeedObjects{
+			{Bucket: "bench-in", Objects: 1, Lines: 1},
+			{Bucket: "bench-in", Prefix: "new/", Objects: 1, Lines: 1, DelaySeconds: 90},
+		},
+	})
+	// 40 base + ceil(90/3)=30 for the sleep + 10 slack.
+	mustContain(t, delayed, "retries: 80\n")
+}
diff --git a/internal/orchestrator/docker.go b/internal/orchestrator/docker.go
index 609cb77..22e8f16 100644
--- a/internal/orchestrator/docker.go
+++ b/internal/orchestrator/docker.go
@@ -49,6 +49,17 @@ type configTemplateContext struct {
 	// case's config once per node ({{@.NodeID@}} → director.id / node.name so each
 	// node self-identifies against the cluster's nodes list). 0 for non-cluster.
 	NodeID int
+	// T0 is the wall-clock instant the harness rendered this config, captured
+	// just before compose up — the closest proxy for "run start" a case can
+	// anchor an absolute timestamp to (see T0Plus). Containers take seconds to
+	// start after T0; a case must budget that latency into its offsets.
+	T0 time.Time
+}
+
+// T0Plus returns T0 + sec seconds as RFC3339 UTC, for absolute device
+// properties such as `end_date: "{{@.T0Plus 90@}}"`.
+func (c configTemplateContext) T0Plus(sec int) string {
+	return c.T0.Add(time.Duration(sec) * time.Second).UTC().Format(time.RFC3339)
 }
 
 // Harness template delimiters. Subject configs carry their OWN native
@@ -530,6 +541,9 @@ services:
 {{- if $.RecvRequiredSubstring }}
       RECEIVER_REQUIRED_SUBSTRING: "{{ $.RecvRequiredSubstring }}"
 {{- end }}
+{{- if $.RecvForbiddenSubstring }}
+      RECEIVER_FORBIDDEN_SUBSTRING: "{{ $.RecvForbiddenSubstring }}"
+{{- end }}
 {{- if $.RecvValidateJSON }}
       RECEIVER_VALIDATE_JSON: "true"
 {{- end }}
@@ -576,6 +590,9 @@ services:
 {{- if .RecvRequiredSubstring }}
       RECEIVER_REQUIRED_SUBSTRING: "{{ .RecvRequiredSubstring }}"
 {{- end }}
+{{- if .RecvForbiddenSubstring }}
+      RECEIVER_FORBIDDEN_SUBSTRING: "{{ .RecvForbiddenSubstring }}"
+{{- end }}
 {{- if .RecvValidateJSON }}
       RECEIVER_VALIDATE_JSON: "true"
 {{- end }}
@@ -1099,7 +1116,7 @@ services:
       test: ["CMD-SHELL", "curl -sf localhost:4566/_localstack/init/ready | grep -q '\"completed\": true'"]
       interval: 3s
       timeout: 5s
-      retries: 40
+      retries: {{ .AWSHealthRetries }}
       start_period: 10s
     restart: "no"
 {{- end }}
@@ -1883,19 +1900,20 @@ type composeVars struct {
 	DatabaseConfPath       string
 	DatabaseCAHost         string
 
-	RecvMode              string
-	RecvListen            string
-	RecvEnv               map[string]string
-	RecvAWSDep            bool
-	RecvAzureDep          bool
-	RecvMinioDep          bool
-	RecvValidateDedup     string
-	RecvValidateContent   string
-	RecvExpectedLines     int64
-	RecvRequiredSubstring string
-	RecvValidateJSON      bool
-	RecvRecordArrival     bool
-	DockerSocketGID       string
+	RecvMode               string
+	RecvListen             string
+	RecvEnv                map[string]string
+	RecvAWSDep             bool
+	RecvAzureDep           bool
+	RecvMinioDep           bool
+	RecvValidateDedup      string
+	RecvValidateContent    string
+	RecvExpectedLines      int64
+	RecvRequiredSubstring  string
+	RecvForbiddenSubstring string
+	RecvValidateJSON       bool
+	RecvRecordArrival      bool
+	DockerSocketGID        string
 
 	// Verifier (a case's `verifier:` block): a one-shot DuckDB container under
 	// the compose profile "verify", so the initial Up() skips it. The runner
@@ -1922,11 +1940,15 @@ type composeVars struct {
 	// Cloud emulator topology (a case's `aws:` / `azure:` blocks).
 	// AWSInitHost is the host path of the rendered LocalStack init script,
 	// bind-mounted into the emulator's ready.d hook directory.
-	AWSEnabled          bool
-	AWSImage            string
-	AWSServices         string
-	AWSRegion           string
-	AWSInitHost         string
+	AWSEnabled  bool
+	AWSImage    string
+	AWSServices string
+	AWSRegion   string
+	AWSInitHost string
+	// AWSHealthRetries is the LocalStack healthcheck retry count: the base
+	// budget plus one retry per healthcheck interval of seed_objects delay,
+	// so a delayed seed group cannot trip the "unhealthy" gate.
+	AWSHealthRetries    int
 	AzureEnabled        bool
 	AzureImage          string
 	AzureConnString     string
@@ -2000,11 +2022,15 @@ func resolveSampleHost(caseDir, sampleFile string) (host, dst string, err error)
 func writeCompose(path string, cfg RunConfig) error {
 	tc := cfg.TestCase
 	s := cfg.Subject
+	// One render instant shared by the subject config (T0Plus) and the
+	// LocalStack init script ($TODAY) so both anchor to the same clock.
+	t0 := time.Now().UTC()
 
-	// Render the subject config as a template (opt-in via {{...}}) so it can
-	// adapt to the host's CPU count.
+	// Render the subject config as a template (opt-in via {{@...@}}) so it can
+	// adapt to the host's CPU count or anchor timestamps to the run start.
 	renderedConfigSrc, err := renderSubjectConfig(cfg.ConfigSrcPath, cfg.TmpDir, configTemplateContext{
 		CPUs: runtime.NumCPU(),
+		T0:   t0,
 	})
 	if err != nil {
 		return err
@@ -2240,11 +2266,12 @@ func writeCompose(path string, cfg RunConfig) error {
 		DockerSocketGID: cfg.DockerSocketGID,
 		TLSCertsHost:    tlsCertsHost,
 
-		RecvValidateDedup:     boolStr(tc.Correctness.ValidateDedup),
-		RecvValidateContent:   boolStr(tc.Correctness.ValidateContent),
-		RecvExpectedLines:     0,
-		RecvRequiredSubstring: tc.Correctness.RequiredSubstring,
-		RecvValidateJSON:      tc.Correctness.ValidateJSON,
+		RecvValidateDedup:      boolStr(tc.Correctness.ValidateDedup),
+		RecvValidateContent:    boolStr(tc.Correctness.ValidateContent),
+		RecvExpectedLines:      0,
+		RecvRequiredSubstring:  tc.Correctness.RequiredSubstring,
+		RecvForbiddenSubstring: tc.Correctness.ForbiddenSubstring,
+		RecvValidateJSON:       tc.Correctness.ValidateJSON,
 		// Arrival timestamp recording is opt-in via the rate_ceiling
 		// check; flipping it on unconditionally would burn memory in
 		// every performance run.
@@ -2528,10 +2555,11 @@ func writeCompose(path string, cfg RunConfig) error {
 		vars.AWSServices = tc.AWS.ServicesOrDefault()
 		vars.AWSRegion = tc.AWS.RegionOrDefault()
 		initPath := filepath.Join(cfg.TmpDir, "aws-init.sh")
-		if err := writeAWSInit(initPath, tc.AWS); err != nil {
+		if err := writeAWSInit(initPath, tc.AWS, t0); err != nil {
 			return err
 		}
 		vars.AWSInitHost = filepath.ToSlash(initPath)
+		vars.AWSHealthRetries = localStackHealthRetries(tc.AWS.TotalSeedDelaySeconds())
 	}
 	if tc.UsesAzure() {
 		vars.AzureEnabled = true
@@ -2713,3 +2741,14 @@ func applyVersionIfUntagged(image, version string) string {
 
 	return image + ":" + version
 }
+
+// localStackHealthRetries sizes the LocalStack healthcheck retry count (3s
+// interval) so the init hook's seed delays fit inside the budget: 40 base
+// retries plus one per 3s of delay plus 10 slack for the uploads themselves.
+func localStackHealthRetries(seedDelaySeconds int) int {
+	const base, intervalSec, slack = 40, 3, 10
+	if seedDelaySeconds <= 0 {
+		return base
+	}
+	return base + (seedDelaySeconds+intervalSec-1)/intervalSec + slack
+}
diff --git a/internal/runner/runner.go b/internal/runner/runner.go
index 4be2247..54cebb8 100644
--- a/internal/runner/runner.go
+++ b/internal/runner/runner.go
@@ -955,6 +955,21 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 		}
 	}
 
+	// Optional received ceiling, independent of the verdict chain above:
+	// proves an upper bound (e.g. an end_date cutoff) was enforced. 0 =
+	// disabled.
+	if tc.Correctness.MaxReceived > 0 && recvMetrics.LinesReceived > tc.Correctness.MaxReceived {
+		msg := fmt.Sprintf("max_received exceeded: expected <= %s lines, got %s",
+			formatCount(tc.Correctness.MaxReceived), formatCount(recvMetrics.LinesReceived))
+		if result.Passed != nil && !*result.Passed {
+			result.FailReason = result.FailReason + "; " + msg
+		} else {
+			f := false
+			result.Passed = &f
+			result.FailReason = msg
+		}
+	}
+
 	// Optional load-balance fairness check (Feature E). Disabled cases
 	// return Passed=true and the result has no LoadBalance key.
 	if tc.Correctness.LoadBalance.Enabled() && len(perReceiver) > 0 {

From 51ab8a0fdaaefd917a2914e45b7f448d40e6ed18 Mon Sep 17 00:00:00 2001
From: Eren Aslan <16862833+erenaslandev@users.noreply.github.com>
Date: Sat, 29 Aug 2026 01:05:32 +0300
Subject: [PATCH 2/3] fix(harness): review follow-ups for seed/verdict knobs

- accept a seed prefix that is only $TODAY
- pass T0 to cluster node config rendering
- enforce max_received in persistence and mid-delivery runners (shared
  helper)
---
 internal/config/cloud.go                     |  4 +-
 internal/config/cloud_test.go                | 10 +++++
 internal/orchestrator/compose_render_test.go |  6 ++-
 internal/orchestrator/docker.go              |  2 +-
 internal/runner/max_received_test.go         | 42 +++++++++++++++++++
 internal/runner/runner.go                    | 43 +++++++++++++-------
 6 files changed, 90 insertions(+), 17 deletions(-)
 create mode 100644 internal/runner/max_received_test.go

diff --git a/internal/config/cloud.go b/internal/config/cloud.go
index 1e77bd0..025a1e8 100644
--- a/internal/config/cloud.go
+++ b/internal/config/cloud.go
@@ -487,7 +487,9 @@ func (tc *TestCase) validateAWS() error {
 			return fmt.Errorf("case %q: seed_objects references undeclared bucket %q", tc.Name, so.Bucket)
 		}
 		if so.Prefix != "" {
-			if err := validateCloudName(tc.Name, "aws seed prefix", strings.ReplaceAll(so.Prefix, SeedTodayToken, "")); err != nil {
+			// Validate the token as the date shape it expands to, so a prefix that
+			// is only "$TODAY" is accepted like "logs/$TODAY/".
+			if err := validateCloudName(tc.Name, "aws seed prefix", strings.ReplaceAll(so.Prefix, SeedTodayToken, "2006/01/02")); err != nil {
 				return err
 			}
 		}
diff --git a/internal/config/cloud_test.go b/internal/config/cloud_test.go
index 9c9f6e0..908372f 100644
--- a/internal/config/cloud_test.go
+++ b/internal/config/cloud_test.go
@@ -44,6 +44,16 @@ func TestValidateCloud(t *testing.T) {
 			},
 			wantErr: "delay_seconds must be non-negative",
 		},
+		{
+			name: "seed prefix that is only the $TODAY token",
+			tc: TestCase{
+				Name: "x",
+				AWS: &AWSConfig{
+					Buckets:     []string{"bench-in"},
+					SeedObjects: []AWSSeedObjects{{Bucket: "bench-in", Prefix: "$TODAY", Objects: 1, Lines: 1}},
+				},
+			},
+		},
 		{
 			name: "seed prefix with $TODAY token and delay",
 			tc: TestCase{
diff --git a/internal/orchestrator/compose_render_test.go b/internal/orchestrator/compose_render_test.go
index 46053dd..7ed4545 100644
--- a/internal/orchestrator/compose_render_test.go
+++ b/internal/orchestrator/compose_render_test.go
@@ -892,7 +892,7 @@ func TestClusterIPComposeRendersVIPPlumbing(t *testing.T) {
 	}
 	defer os.RemoveAll(srcDir)
 	srcCfg := filepath.Join(srcDir, "vmetric.yml")
-	if err := os.WriteFile(srcCfg, []byte("director:\n  id: {{@.NodeID@}}\n"), 0o644); err != nil {
+	if err := os.WriteFile(srcCfg, []byte("director:\n  id: {{@.NodeID@}}\nproperties:\n  start_date: \"{{@.T0Plus 90@}}\"\n"), 0o644); err != nil {
 		t.Fatal(err)
 	}
 
@@ -955,6 +955,10 @@ func TestClusterIPComposeRendersVIPPlumbing(t *testing.T) {
 			t.Fatalf("read node %s config: %v", n.id, err)
 		}
 		mustContain(t, string(b), n.want)
+		// The cluster path shares the run-start instant with the singular
+		// path: a T0Plus template must render a real timestamp, not year 0001.
+		mustNotContain(t, string(b), "0001-01-01")
+		mustContain(t, string(b), "start_date: \"20")
 	}
 }
 
diff --git a/internal/orchestrator/docker.go b/internal/orchestrator/docker.go
index 22e8f16..3fcf654 100644
--- a/internal/orchestrator/docker.go
+++ b/internal/orchestrator/docker.go
@@ -2051,7 +2051,7 @@ func writeCompose(path string, cfg RunConfig) error {
 	if tc.Cluster != nil && tc.Cluster.Nodes > 0 {
 		for i := 1; i <= tc.Cluster.Nodes; i++ {
 			nodeOut := filepath.Join(cfg.TmpDir, fmt.Sprintf("cluster-node-%d.yml", i))
-			if err := renderConfigToFile(cfg.ConfigSrcPath, nodeOut, configTemplateContext{CPUs: runtime.NumCPU(), NodeID: i}); err != nil {
+			if err := renderConfigToFile(cfg.ConfigSrcPath, nodeOut, configTemplateContext{CPUs: runtime.NumCPU(), NodeID: i, T0: t0}); err != nil {
 				return err
 			}
 			clusterNodes = append(clusterNodes, clusterNode{
diff --git a/internal/runner/max_received_test.go b/internal/runner/max_received_test.go
new file mode 100644
index 0000000..76b530b
--- /dev/null
+++ b/internal/runner/max_received_test.go
@@ -0,0 +1,42 @@
+package runner
+
+import (
+	"testing"
+
+	"github.com/VirtualMetric/PipeBench/internal/config"
+	"github.com/VirtualMetric/PipeBench/internal/results"
+)
+
+func TestApplyMaxReceived(t *testing.T) {
+	t.Parallel()
+
+	passed, failed := true, false
+	tests := []struct {
+		name       string
+		cap        int64
+		received   int64
+		in         results.RunResult
+		wantPassed *bool
+		wantReason string
+	}{
+		{"disabled", 0, 99999, results.RunResult{Passed: &passed}, &passed, ""},
+		{"under cap untouched", 100, 100, results.RunResult{Passed: &passed}, &passed, ""},
+		{"over cap fails a passing result", 100, 101, results.RunResult{Passed: &passed}, &failed, "max_received exceeded: expected <= 100 lines, got 101"},
+		{"over cap appends to a failing result", 100, 250, results.RunResult{Passed: &failed, FailReason: "loss"}, &failed, "loss; max_received exceeded: expected <= 100 lines, got 250"},
+		{"over cap with no verdict yet fails", 100, 101, results.RunResult{}, &failed, "max_received exceeded: expected <= 100 lines, got 101"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			tc := &config.TestCase{Correctness: config.CorrectnessConfig{MaxReceived: tt.cap}}
+			res := tt.in
+			applyMaxReceived(tc, tt.received, &res)
+			if (res.Passed == nil) != (tt.wantPassed == nil) || (res.Passed != nil && *res.Passed != *tt.wantPassed) {
+				t.Fatalf("passed = %v want %v", res.Passed, tt.wantPassed)
+			}
+			if res.FailReason != tt.wantReason {
+				t.Fatalf("reason = %q want %q", res.FailReason, tt.wantReason)
+			}
+		})
+	}
+}
diff --git a/internal/runner/runner.go b/internal/runner/runner.go
index 54cebb8..6352cef 100644
--- a/internal/runner/runner.go
+++ b/internal/runner/runner.go
@@ -955,20 +955,7 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 		}
 	}
 
-	// Optional received ceiling, independent of the verdict chain above:
-	// proves an upper bound (e.g. an end_date cutoff) was enforced. 0 =
-	// disabled.
-	if tc.Correctness.MaxReceived > 0 && recvMetrics.LinesReceived > tc.Correctness.MaxReceived {
-		msg := fmt.Sprintf("max_received exceeded: expected <= %s lines, got %s",
-			formatCount(tc.Correctness.MaxReceived), formatCount(recvMetrics.LinesReceived))
-		if result.Passed != nil && !*result.Passed {
-			result.FailReason = result.FailReason + "; " + msg
-		} else {
-			f := false
-			result.Passed = &f
-			result.FailReason = msg
-		}
-	}
+	applyMaxReceived(tc, recvMetrics.LinesReceived, &result)
 
 	// Optional load-balance fairness check (Feature E). Disabled cases
 	// return Passed=true and the result has no LoadBalance key.
@@ -1119,6 +1106,28 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 //  4. Start receiver
 //  5. Wait for subject to forward buffered logs to receiver
 //  6. Verify: all logs should arrive with 0% loss
+//
+// applyMaxReceived enforces the optional correctness.max_received ceiling on a
+// result, independent of the verdict chain that produced it: it proves an upper
+// bound (e.g. an end_date cutoff or a retry cap) was honoured. 0 = disabled. It
+// is shared by the generic correctness path and the specialized runners that
+// have receiver metrics, so a case cannot pass after exceeding its ceiling
+// just because its type dispatches elsewhere.
+func applyMaxReceived(tc *config.TestCase, linesReceived int64, result *results.RunResult) {
+	if tc.Correctness.MaxReceived <= 0 || linesReceived <= tc.Correctness.MaxReceived {
+		return
+	}
+	msg := fmt.Sprintf("max_received exceeded: expected <= %s lines, got %s",
+		formatCount(tc.Correctness.MaxReceived), formatCount(linesReceived))
+	if result.Passed != nil && !*result.Passed {
+		result.FailReason = result.FailReason + "; " + msg
+		return
+	}
+	f := false
+	result.Passed = &f
+	result.FailReason = msg
+}
+
 func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.Subject) (results.RunResult, error) {
 	configName := r.opts.ConfigName
 	subject = r.applySubjectOverrides(subject)
@@ -1370,6 +1379,8 @@ func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.S
 		result.FailReason = strings.Join(errors, "; ")
 	}
 
+	applyMaxReceived(tc, recvMetrics.LinesReceived, &result)
+
 	dir, err := r.saveResult(result, metricsCSVSrc)
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
@@ -1691,6 +1702,8 @@ func (r *Runner) runPersistenceShutdownCorrectness(tc *config.TestCase, subject
 		result.FailReason = strings.Join(errors, "; ")
 	}
 
+	applyMaxReceived(tc, recvMetrics.LinesReceived, &result)
+
 	dir, err := r.saveResult(result, metricsCSVSrc)
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
@@ -2064,6 +2077,8 @@ func (r *Runner) runMidDeliveryAction(tc *config.TestCase, subject config.Subjec
 
 	// Persist the result like every other run path — Run's contract is to
 	// return the *persisted* result.
+	applyMaxReceived(tc, recvMetrics.LinesReceived, &result)
+
 	dir, err := r.saveResult(result, metricsCSVSrc)
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)

From dbbc8a3ab9482c6367752377a1c04166a183fa6d Mon Sep 17 00:00:00 2001
From: Eren Aslan <16862833+erenaslandev@users.noreply.github.com>
Date: Sat, 29 Aug 2026 01:57:14 +0300
Subject: [PATCH 3/3] fix(runner): use persisted result for verdict in runner
 tests

---
 internal/runner/runner.go | 29 ++++++++++++++++++-----------
 1 file changed, 18 insertions(+), 11 deletions(-)

diff --git a/internal/runner/runner.go b/internal/runner/runner.go
index 6352cef..2b90171 100644
--- a/internal/runner/runner.go
+++ b/internal/runner/runner.go
@@ -1403,11 +1403,13 @@ func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.S
 	}
 	fmt.Printf("  total time: %.1fs\n", elapsed)
 
-	if passed {
+	// Print from the persisted result: applyMaxReceived may have failed it
+	// after the local verdict was computed.
+	if result.Passed != nil && *result.Passed {
 		fmt.Println("  persistence correctness: PASSED ✓")
 	} else {
 		fmt.Println("  persistence correctness: FAILED ✗")
-		for _, e := range errors {
+		for _, e := range strings.Split(result.FailReason, "; ") {
 			fmt.Printf("    - %s\n", e)
 		}
 	}
@@ -1726,11 +1728,13 @@ func (r *Runner) runPersistenceShutdownCorrectness(tc *config.TestCase, subject
 	}
 	fmt.Printf("  total time: %.1fs\n", elapsed)
 
-	if passed {
+	// Print from the persisted result: applyMaxReceived may have failed it
+	// after the local verdict was computed.
+	if result.Passed != nil && *result.Passed {
 		fmt.Println("  persistence restart correctness: PASSED ✓")
 	} else {
 		fmt.Println("  persistence restart correctness: FAILED ✗")
-		for _, e := range errors {
+		for _, e := range strings.Split(result.FailReason, "; ") {
 			fmt.Printf("    - %s\n", e)
 		}
 	}
@@ -2029,12 +2033,6 @@ func (r *Runner) runMidDeliveryAction(tc *config.TestCase, subject config.Subjec
 	if metrics.IOThroughputAvg > 0 {
 		fmt.Printf("  io throughput: avg %.1f MB/s\n", metrics.IOThroughputAvg/(1024*1024))
 	}
-	if passed {
-		fmt.Printf("  %s: PASSED ✓\n", f.verdictLabel)
-	} else {
-		fmt.Printf("  %s: FAILED ✗\n", f.verdictLabel)
-	}
-
 	result := results.RunResult{
 		TestName:        tc.Name,
 		Config:          configName,
@@ -2076,8 +2074,17 @@ func (r *Runner) runMidDeliveryAction(tc *config.TestCase, subject config.Subjec
 	}
 
 	// Persist the result like every other run path — Run's contract is to
-	// return the *persisted* result.
+	// return the *persisted* result. The verdict is printed from it so a
+	// max_received failure shows up on the CLI as well.
 	applyMaxReceived(tc, recvMetrics.LinesReceived, &result)
+	if result.Passed != nil && *result.Passed {
+		fmt.Printf("  %s: PASSED ✓\n", f.verdictLabel)
+	} else {
+		fmt.Printf("  %s: FAILED ✗\n", f.verdictLabel)
+		for _, e := range strings.Split(result.FailReason, "; ") {
+			fmt.Printf("    - %s\n", e)
+		}
+	}
 
 	dir, err := r.saveResult(result, metricsCSVSrc)
 	if err != nil {