diff --git a/cmd/harness/main.go b/cmd/harness/main.go
index bbe3847..dd778af 100644
--- a/cmd/harness/main.go
+++ b/cmd/harness/main.go
@@ -1,12 +1,16 @@
 package main
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
+	"os/signal"
 	"path/filepath"
 	"sort"
 	"strings"
+	"syscall"
 	"text/tabwriter"
 	"time"
 
@@ -52,6 +56,23 @@ var (
 )
 
 func main() {
+	// The root context is cancelled only by SIGINT/SIGTERM — never by a
+	// timeout — so context.Canceled unambiguously means "user interrupt"
+	// everywhere downstream. On the first signal the run loop winds down
+	// through its normal error path (deferred teardown still runs); the
+	// handler then unregisters itself so a second signal kills the process
+	// immediately via the default disposition.
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-sig
+		fmt.Fprintln(os.Stderr, "\ninterrupt — tearing down current run; press Ctrl+C again to exit immediately (then run `harness clean` to remove leftovers)")
+		cancel()
+		signal.Stop(sig)
+	}()
+
 	root := &cobra.Command{
 		Use:   "harness",
 		Short: "PipeBench — containerized data pipeline benchmarking",
@@ -73,7 +94,7 @@ No cloud account, Terraform, Ansible, or SSH required.`,
 	root.AddCommand(serveCmd())
 	root.AddCommand(versionCmd())
 
-	if err := root.Execute(); err != nil {
+	if err := root.ExecuteContext(ctx); err != nil {
 		os.Exit(1)
 	}
 }
@@ -240,7 +261,7 @@ func testCmd() *cobra.Command {
 				Drain:            drain,
 			}
 
-			r := runner.New(opts)
+			r := runner.New(cmd.Context(), opts)
 
 			// Stub out per-(hardware, subject) result files up front so a
 			// subject that fails every test still appears in the index.
@@ -284,7 +305,11 @@ func testCmd() *cobra.Command {
 				subjectOutcomes = subjectOutcomes[:0]
 			}
 
-			for _, p := range pairs {
+			for i, p := range pairs {
+				if cmd.Context().Err() != nil {
+					fmt.Fprintf(os.Stderr, "interrupted — skipping remaining %d run(s)\n", len(pairs)-i)
+					break
+				}
 				if perSubjectSummaries && p.subject.Name != currentSubject {
 					flushSubjectSummary()
 					currentSubject = p.subject.Name
@@ -307,7 +332,11 @@ func testCmd() *cobra.Command {
 
 				res, err := r.Run(p.tc, p.subject)
 				if err != nil {
-					fmt.Fprintf(os.Stderr, "ERROR running %s/%s: %v\n", p.tc.Name, p.subject.Name, err)
+					if errors.Is(err, context.Canceled) {
+						fmt.Fprintf(os.Stderr, "interrupted %s/%s\n", p.tc.Name, p.subject.Name)
+					} else {
+						fmt.Fprintf(os.Stderr, "ERROR running %s/%s: %v\n", p.tc.Name, p.subject.Name, err)
+					}
 					failed = append(failed, p.tc.Name+"/"+p.subject.Name)
 				}
 				o := runOutcome{
@@ -328,6 +357,9 @@ func testCmd() *cobra.Command {
 			}
 			printRunSummary(outcomes)
 
+			if cmd.Context().Err() != nil {
+				return fmt.Errorf("interrupted by signal (%d of %d runs attempted); leftover containers, if any: `harness clean`", len(outcomes), len(pairs))
+			}
 			if len(failed) > 0 {
 				return fmt.Errorf("%d run(s) failed: %v", len(failed), failed)
 			}
@@ -566,6 +598,16 @@ func cleanCmd() *cobra.Command {
 		Short: "Remove all bench containers and networks",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			fmt.Println("Removing bench containers…")
+			// Prefix sweep catches every bench-* container regardless of
+			// name (localstack, azurite, redpanda, vault, verifier, plural
+			// generators/receivers, …) — the fixed list below is only a
+			// fallback for when `docker ps` itself fails.
+			if out, err := exec.Command("docker", "ps", "-aq", "--filter", "name=^bench-").Output(); err == nil {
+				if ids := strings.Fields(string(out)); len(ids) > 0 {
+					rmArgs := append([]string{"rm", "-f"}, ids...)
+					_ = exec.Command("docker", rmArgs...).Run()
+				}
+			}
 			containers := []string{
 				"bench-generator",
 				"bench-receiver",
@@ -577,6 +619,16 @@ func cleanCmd() *cobra.Command {
 			for _, c := range containers {
 				_ = exec.Command("docker", "rm", "-f", c).Run()
 			}
+			// Compose networks are named <tmpdir-project>_bench; an
+			// interrupted run that never reached `compose down` leaves them
+			// behind. Removal fails harmlessly while a container still uses
+			// the network, so containers go first above.
+			fmt.Println("Removing bench networks…")
+			if out, err := exec.Command("docker", "network", "ls", "-q", "--filter", "name=bench").Output(); err == nil {
+				for _, id := range strings.Fields(string(out)) {
+					_ = exec.Command("docker", "network", "rm", id).Run()
+				}
+			}
 			fmt.Println("Done.")
 			return nil
 		},
diff --git a/containers/collector/main.go b/containers/collector/main.go
index 2095953..bbf8aea 100644
--- a/containers/collector/main.go
+++ b/containers/collector/main.go
@@ -215,15 +215,8 @@ func dockerStatsToRow(cur *dockerStats, prev *dockerStats) MetricsRow {
 		cpuPct = 100 * numCPU
 	}
 
-	cache := cur.MemoryStats.Stats["cache"]
-	memUsed := int64(cur.MemoryStats.Usage) - int64(cache)
-	if memUsed < 0 {
-		memUsed = int64(cur.MemoryStats.Usage)
-	}
-	memFree := int64(cur.MemoryStats.Limit) - int64(cur.MemoryStats.Usage)
-	if memFree < 0 {
-		memFree = 0
-	}
+	memUsed, cache := memUsage(cur.MemoryStats.Usage, cur.MemoryStats.Stats)
+	memFree := max(int64(cur.MemoryStats.Limit)-int64(cur.MemoryStats.Usage), 0)
 
 	var netRecv, netSend int64
 	for _, n := range cur.Networks {
@@ -280,7 +273,7 @@ func dockerStatsToRow(cur *dockerStats, prev *dockerStats) MetricsRow {
 		CpuUsr:  cpuPct,
 		CpuIdl:  100.0 - cpuPct,
 		MemUsed: memUsed,
-		MemCach: int64(cache),
+		MemCach: cache,
 		MemFree: memFree,
 		NetRecv: netRecv,
 		NetSend: netSend,
@@ -289,6 +282,34 @@ func dockerStatsToRow(cur *dockerStats, prev *dockerStats) MetricsRow {
 	}
 }
 
+// memUsage converts Docker stats memory counters into used/cache bytes.
+//
+// All page cache is excluded, except shmem: cgroups charge shared image-layer
+// and binary pages to whichever container faults them first, and pages
+// touched twice migrate to the active list — so both raw usage and the
+// `docker stats` formula (usage - inactive_file) swing tens of MB between
+// otherwise identical runs. A benchmark needs a reproducible figure, so
+// mem_used is anon + shmem + kernel. shmem stays counted because tmpfs/shm
+// segments are swap-backed real memory that cgroups fold into the page
+// cache number.
+func memUsage(usage uint64, stats map[string]uint64) (used, cache int64) {
+	// cgroup v1 keys the page cache as "total_cache"/"cache"; v2 as "file".
+	pageCache, ok := stats["total_cache"]
+	shmem := stats["total_shmem"]
+	if !ok {
+		if pageCache, ok = stats["cache"]; !ok {
+			pageCache = stats["file"]
+		}
+		shmem = stats["shmem"]
+	}
+	cache = int64(pageCache)
+
+	if pageCache >= shmem && usage >= pageCache-shmem {
+		return int64(usage - (pageCache - shmem)), cache
+	}
+	return int64(usage), cache
+}
+
 // --- Common ---
 
 func mustEnv(key string) string {
diff --git a/containers/collector/main_test.go b/containers/collector/main_test.go
new file mode 100644
index 0000000..5dc7c20
--- /dev/null
+++ b/containers/collector/main_test.go
@@ -0,0 +1,102 @@
+package main
+
+import "testing"
+
+func TestMemUsage(t *testing.T) {
+	tests := []struct {
+		name      string
+		usage     uint64
+		stats     map[string]uint64
+		wantUsed  int64
+		wantCache int64
+	}{
+		{
+			name:  "cgroup v2 excludes file cache",
+			usage: 200 << 20,
+			stats: map[string]uint64{
+				"file":          130 << 20,
+				"inactive_file": 120 << 20,
+				"active_file":   10 << 20,
+				"anon":          60 << 20,
+			},
+			wantUsed:  70 << 20,
+			wantCache: 130 << 20,
+		},
+		{
+			name:  "cgroup v2 keeps shmem counted",
+			usage: 200 << 20,
+			stats: map[string]uint64{
+				"file":  130 << 20,
+				"shmem": 30 << 20,
+			},
+			wantUsed:  100 << 20,
+			wantCache: 130 << 20,
+		},
+		{
+			name:  "cgroup v1 uses total_cache and total_shmem",
+			usage: 200 << 20,
+			stats: map[string]uint64{
+				"total_cache": 110 << 20,
+				"total_shmem": 20 << 20,
+				"cache":       90 << 20,
+			},
+			wantUsed:  110 << 20,
+			wantCache: 110 << 20,
+		},
+		{
+			name:  "cgroup v1 plain cache key fallback",
+			usage: 200 << 20,
+			stats: map[string]uint64{
+				"cache": 90 << 20,
+				"shmem": 10 << 20,
+			},
+			wantUsed:  120 << 20,
+			wantCache: 90 << 20,
+		},
+		{
+			name:  "cache >= usage falls back to raw usage",
+			usage: 50 << 20,
+			stats: map[string]uint64{
+				"file": 60 << 20,
+			},
+			wantUsed:  50 << 20,
+			wantCache: 60 << 20,
+		},
+		{
+			name:  "shmem > cache falls back to raw usage",
+			usage: 100 << 20,
+			stats: map[string]uint64{
+				"file":  10 << 20,
+				"shmem": 20 << 20,
+			},
+			wantUsed:  100 << 20,
+			wantCache: 10 << 20,
+		},
+		{
+			name:      "empty stats map reports raw usage",
+			usage:     70 << 20,
+			stats:     map[string]uint64{},
+			wantUsed:  70 << 20,
+			wantCache: 0,
+		},
+		{
+			name:      "nil stats map reports raw usage",
+			usage:     70 << 20,
+			stats:     nil,
+			wantUsed:  70 << 20,
+			wantCache: 0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			used, cache := memUsage(tt.usage, tt.stats)
+			if used != tt.wantUsed {
+				t.Errorf("used = %d, want %d", used, tt.wantUsed)
+			}
+			if cache != tt.wantCache {
+				t.Errorf("cache = %d, want %d", cache, tt.wantCache)
+			}
+		})
+	}
+}
diff --git a/internal/config/case.go b/internal/config/case.go
index 15cbd7b..ebc4632 100644
--- a/internal/config/case.go
+++ b/internal/config/case.go
@@ -7,6 +7,7 @@ import (
 	"path"
 	"path/filepath"
 	"regexp"
+	"sort"
 	"strings"
 	"time"
 
@@ -80,6 +81,15 @@ type TestCase struct {
 	// provider) without real infrastructure.
 	Vault *VaultConfig `yaml:"vault"`
 
+	// Database, when set, adds a real database backend to the test topology,
+	// keyed by `engine:` (see DatabaseEngines): the harness renders a
+	// `database` service plus a one-shot `database-init` that creates the
+	// database and runs the declared seed SQL, and the subject gates on the
+	// seeding completing. Lets a case exercise a subject's database
+	// poller/collector (e.g. vmetric's mssql customquery collector) against
+	// a real engine instead of a mock.
+	Database *DatabaseConfig `yaml:"database"`
+
 	// AWS, when set, adds a LocalStack emulator to the test topology and
 	// creates the declared S3/SQS/SNS/Kinesis/CloudWatch resources before
 	// the subject starts. See AWSConfig in cloud.go.
@@ -523,6 +533,58 @@ func (v *VaultConfig) MountOrDefault() string {
 	return "secret"
 }
 
+// DatabaseConfig configures the in-topology database backend (see
+// TestCase.Database). Engine selects a DatabaseEngines entry, which supplies
+// the image/credential defaults, env vars, healthcheck, and init command;
+// Image/Password/Database override the engine's defaults per-case.
+type DatabaseConfig struct {
+	// Engine selects the backend (e.g. "mssql") — a key into DatabaseEngines.
+	Engine string `yaml:"engine"`
+	// Image overrides the engine's default container image.
+	Image string `yaml:"image"`
+	// Password overrides the engine's default admin/root credential. Flows
+	// only through the compose `environment:` block and shell `$VAR`
+	// expansion — never string-built into a command line. Test-only.
+	Password string `yaml:"password"`
+	// Database is the database name created by database-init (default
+	// "bench").
+	Database string `yaml:"database"`
+	// SeedSQL is DDL/DML run once, after the server is healthy, via the
+	// engine's CLI against Database. Mounted as a file — never rendered
+	// inline into a command line.
+	SeedSQL string `yaml:"seed_sql"`
+	// TLS makes the database container present a per-run CA-signed server
+	// certificate (SAN "database") so a device can verify it instead of using
+	// insecure_skip_verify. The CA is mounted into the subject at
+	// /opt/vmetric/certs/ca.crt; whether a device trusts it is the device's
+	// own config choice (a negative case can withhold the CA to prove reject).
+	TLS bool `yaml:"tls"`
+}
+
+// ImageOrDefault, PasswordOrDefault, DatabaseOrDefault centralize the
+// database-backend defaults so the orchestrator and any caller render the
+// same values.
+func (d *DatabaseConfig) ImageOrDefault(engine DatabaseEngine) string {
+	if d != nil && d.Image != "" {
+		return d.Image
+	}
+	return engine.DefaultImage
+}
+
+func (d *DatabaseConfig) PasswordOrDefault(engine DatabaseEngine) string {
+	if d != nil && d.Password != "" {
+		return d.Password
+	}
+	return engine.DefaultPassword
+}
+
+func (d *DatabaseConfig) DatabaseOrDefault() string {
+	if d != nil && d.Database != "" {
+		return d.Database
+	}
+	return "bench"
+}
+
 // AgentConfig configures an external agent container in the test topology
 // (see TestCase.Agent). Unlike endpoints (which the subject connects out to),
 // the agent connects INTO the subject over the bench network — it starts after
@@ -806,6 +868,13 @@ func (tc *TestCase) UsesKafka() bool { return tc.Kafka != nil }
 // UsesVault reports whether the case adds a Vault dev server to the topology.
 func (tc *TestCase) UsesVault() bool { return tc.Vault != nil }
 
+// UsesDatabase reports whether the case adds a database backend to the topology.
+func (tc *TestCase) UsesDatabase() bool { return tc.Database != nil }
+
+// UsesDatabaseTLS reports whether the database container should terminate TLS
+// with a per-run CA-signed cert (so devices can verify instead of skip-verify).
+func (tc *TestCase) UsesDatabaseTLS() bool { return tc.Database != nil && tc.Database.TLS }
+
 // UsesVerifier reports whether the case drives correctness through the one-shot
 // DuckDB verifier container instead of a receiver.
 func (tc *TestCase) UsesVerifier() bool { return tc.Verifier != nil }
@@ -1009,6 +1078,9 @@ func (tc *TestCase) Validate() error {
 	if err := tc.validateVault(); err != nil {
 		return err
 	}
+	if err := tc.validateDatabase(); err != nil {
+		return err
+	}
 	if err := tc.validateKafkaAuth(); err != nil {
 		return err
 	}
@@ -2184,6 +2256,48 @@ func (tc *TestCase) validateVault() error {
 	return nil
 }
 
+// databaseNameRe constrains the database name, the only DatabaseConfig field
+// interpolated directly into a compose-embedded shell command (CREATE
+// DATABASE <name>). SeedSQL is mounted as a file and Password flows only
+// through a compose `environment:` value expanded by the shell at runtime,
+// so neither needs charset restriction (mirrors vaultPathRe/vaultTokenRe).
+var databaseNameRe = regexp.MustCompile(`^[A-Za-z0-9_]+$`)
+
+// validateDatabase checks the optional `database:` block: engine must be a
+// known DatabaseEngines entry, seeding at least some SQL is mandatory (a
+// database nothing seeds is a case-authoring mistake), and the database name
+// is charset-restricted.
+func (tc *TestCase) validateDatabase() error {
+	if tc.Database == nil {
+		return nil
+	}
+	engine, ok := DatabaseEngines[tc.Database.Engine]
+	if !ok {
+		known := make([]string, 0, len(DatabaseEngines))
+		for name := range DatabaseEngines {
+			known = append(known, name)
+		}
+		sort.Strings(known)
+		return fmt.Errorf("case %q: unknown database.engine %q, known engines: %s", tc.Name, tc.Database.Engine, strings.Join(known, ", "))
+	}
+	if tc.Database.SeedSQL == "" {
+		return fmt.Errorf("case %q: database block requires `seed_sql`", tc.Name)
+	}
+	if !databaseNameRe.MatchString(tc.Database.DatabaseOrDefault()) {
+		return fmt.Errorf("case %q: database.database %q must match %s", tc.Name, tc.Database.DatabaseOrDefault(), databaseNameRe)
+	}
+	// Reject database.tls for engines without TLS wiring at load time, rather
+	// than letting it fail later during compose rendering. An empty
+	// TLSServerCertPath is the registry's "no TLS support" marker (see the
+	// DatabaseEngine struct doc and the docker.go compose guard) — keying off it
+	// covers any current or future TLS-less engine (e.g. oracle) with no
+	// per-engine list.
+	if tc.Database.TLS && engine.TLSServerCertPath == "" {
+		return fmt.Errorf("case %q: database.tls is not supported by engine %q", tc.Name, tc.Database.Engine)
+	}
+	return nil
+}
+
 // validateRotation checks the optional `rotation:` block and the
 // director_agent_tls_cert_rotation_correctness type's structural requirements:
 // the block is required for (and only meaningful to) that type, the mode must be
diff --git a/internal/config/database_config_test.go b/internal/config/database_config_test.go
new file mode 100644
index 0000000..a316f36
--- /dev/null
+++ b/internal/config/database_config_test.go
@@ -0,0 +1,124 @@
+package config
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestValidateDatabase(t *testing.T) {
+	base := func() *TestCase {
+		return &TestCase{
+			Name: "database_case",
+			Database: &DatabaseConfig{
+				Engine:  "mssql",
+				SeedSQL: "CREATE TABLE t (id INT);",
+			},
+		}
+	}
+
+	tests := []struct {
+		name    string
+		mutate  func(tc *TestCase)
+		wantErr string // substring of the expected error, "" = valid
+	}{
+		{name: "valid mssql database block", mutate: func(*TestCase) {}},
+		{name: "valid oracle database block", mutate: func(tc *TestCase) { tc.Database.Engine = "oracle" }},
+		{name: "nil database block is a no-op", mutate: func(tc *TestCase) { tc.Database = nil }},
+		{
+			name:    "unknown engine",
+			mutate:  func(tc *TestCase) { tc.Database.Engine = "nosuchdb" },
+			wantErr: `unknown database.engine "nosuchdb"`,
+		},
+		{
+			name:    "missing seed_sql",
+			mutate:  func(tc *TestCase) { tc.Database.SeedSQL = "" },
+			wantErr: "requires `seed_sql`",
+		},
+		{
+			name:    "database name with shell metacharacter",
+			mutate:  func(tc *TestCase) { tc.Database.Database = "bench; rm -rf /" },
+			wantErr: "must match",
+		},
+		{name: "custom database name", mutate: func(tc *TestCase) { tc.Database.Database = "bench_2" }},
+		{
+			name:    "tls on a TLS-less engine is rejected",
+			mutate:  func(tc *TestCase) { tc.Database.Engine = "oracle"; tc.Database.TLS = true },
+			wantErr: `not supported by engine "oracle"`,
+		},
+		{name: "tls on a TLS-capable engine is allowed", mutate: func(tc *TestCase) { tc.Database.TLS = true }},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			tc := base()
+			tt.mutate(tc)
+			err := tc.validateDatabase()
+			if tt.wantErr == "" {
+				if err != nil {
+					t.Fatalf("validateDatabase() = %v, want nil", err)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("validateDatabase() = %v, want error containing %q", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestOracleEngineEntry(t *testing.T) {
+	engine, ok := DatabaseEngines["oracle"]
+	if !ok {
+		t.Fatal("oracle engine not registered in DatabaseEngines")
+	}
+	if engine.DefaultImage != "gvenzl/oracle-free:23-slim-faststart" {
+		t.Errorf("DefaultImage = %q", engine.DefaultImage)
+	}
+	if env := engine.BuildEnv("pw"); env["ORACLE_PASSWORD"] != "pw" {
+		t.Errorf("BuildEnv ORACLE_PASSWORD = %q, want pw", env["ORACLE_PASSWORD"])
+	}
+	if hc := engine.BuildHealthCmd("pw"); hc != "healthcheck.sh" {
+		t.Errorf("BuildHealthCmd = %q, want healthcheck.sh", hc)
+	}
+	init := engine.BuildInitCmd("pw", "bench")
+	// Password must never be string-built into the command — only the env ref.
+	if strings.Contains(init, "pw") {
+		t.Errorf("BuildInitCmd leaks the raw password: %q", init)
+	}
+	for _, want := range []string{"$$ORACLE_PASSWORD", "/db-seed/init.sql", "sqlplus", "FREEPDB1"} {
+		if !strings.Contains(init, want) {
+			t.Errorf("BuildInitCmd missing %q: %q", want, init)
+		}
+	}
+}
+
+func TestDatabaseConfigOrDefaults(t *testing.T) {
+	engine := DatabaseEngines["mssql"]
+
+	t.Run("nil config falls back to engine defaults", func(t *testing.T) {
+		var d *DatabaseConfig
+		if got := d.ImageOrDefault(engine); got != engine.DefaultImage {
+			t.Errorf("ImageOrDefault() = %q, want %q", got, engine.DefaultImage)
+		}
+		if got := d.PasswordOrDefault(engine); got != engine.DefaultPassword {
+			t.Errorf("PasswordOrDefault() = %q, want %q", got, engine.DefaultPassword)
+		}
+		if got := d.DatabaseOrDefault(); got != "bench" {
+			t.Errorf(`DatabaseOrDefault() = %q, want "bench"`, got)
+		}
+	})
+
+	t.Run("explicit values override the engine defaults", func(t *testing.T) {
+		d := &DatabaseConfig{Image: "custom:latest", Password: "custom-pass", Database: "custom_db"}
+		if got := d.ImageOrDefault(engine); got != "custom:latest" {
+			t.Errorf("ImageOrDefault() = %q, want custom:latest", got)
+		}
+		if got := d.PasswordOrDefault(engine); got != "custom-pass" {
+			t.Errorf("PasswordOrDefault() = %q, want custom-pass", got)
+		}
+		if got := d.DatabaseOrDefault(); got != "custom_db" {
+			t.Errorf("DatabaseOrDefault() = %q, want custom_db", got)
+		}
+	})
+}
diff --git a/internal/config/database_engines.go b/internal/config/database_engines.go
new file mode 100644
index 0000000..5ced5a3
--- /dev/null
+++ b/internal/config/database_engines.go
@@ -0,0 +1,253 @@
+package config
+
+import "fmt"
+
+// DatabaseEngine supplies everything the generic `database:` compose block
+// needs to run a specific engine: the default image/credential, the env vars
+// the image expects, a healthcheck command using a CLI already shipped in
+// the image, and the one-shot init command that creates the database and
+// runs the seed file mounted at /db-seed/init.sql. Adding a new engine
+// (mysql, postgresql, ...) is exactly one new entry in DatabaseEngines — no
+// orchestrator or compose-template changes required.
+type DatabaseEngine struct {
+	DefaultImage    string
+	DefaultPassword string
+	BuildEnv        func(password string) map[string]string
+	BuildHealthCmd  func(password string) string
+	BuildInitCmd    func(password, database string) string
+
+	// TLS wiring (used only when a case sets `database.tls: true`). The
+	// orchestrator generates a CA + RSA server cert and bind-mounts server.crt
+	// / server.key at TLSServerCertPath / TLSServerKeyPath in the container;
+	// BuildTLSConf returns a config file (mounted at its returned path) that
+	// points the engine at those cert files so it terminates TLS with them. An
+	// empty TLSServerCertPath means the engine does not support this mechanism.
+	TLSServerCertPath string
+	TLSServerKeyPath  string
+	BuildTLSConf      func() (mountPath, content string)
+
+	// BuildTLSCommand, when non-nil and the case enables TLS, overrides the
+	// database container's entrypoint with `/bin/sh -c <returned string>`.
+	// Engines whose image auto-reads the BuildTLSConf file (mssql via mssql.conf,
+	// mysql via /etc/mysql/conf.d) leave this nil and use the image defaults.
+	// Postgres needs it because it (a) refuses to start with a group/world-
+	// readable TLS key — the harness writes server.key as 0644 — and (b) has no
+	// auto-include conf dir, so TLS must be turned on via `-c` args. certPath /
+	// keyPath are the in-container mount paths (TLSServerCertPath/KeyPath).
+	BuildTLSCommand func(certPath, keyPath string) string
+}
+
+// DatabaseEngines is the registry of engines the `database:` case block can
+// select via `engine:`.
+var DatabaseEngines = map[string]DatabaseEngine{
+	"mssql": {
+		DefaultImage: "mcr.microsoft.com/mssql/server:2022-latest",
+		// 8+ chars, upper+lower+digit+symbol — satisfies MSSQL's SA
+		// password complexity policy. Test-only credential.
+		DefaultPassword: "PipeBench-Db1!",
+		BuildEnv: func(password string) map[string]string {
+			return map[string]string{
+				"ACCEPT_EULA":       "Y",
+				"MSSQL_PID":         "Developer",
+				"MSSQL_SA_PASSWORD": password,
+			}
+		},
+		BuildHealthCmd: func(password string) string {
+			// -C trusts the container's self-signed cert for this internal
+			// probe only — orthogonal to a case's own TLS verification
+			// setting under test on the director's connection. Two escaping
+			// layers apply here: (1) `$$` instead of `$` — docker compose
+			// itself interpolates `$VAR` in compose file values at parse
+			// time, so a bare `$MSSQL_SA_PASSWORD` would resolve against
+			// the HOST environment (usually unset -> blank) before the
+			// container's shell ever sees it; `$$` escapes to a literal `$`
+			// that compose passes through for the container's own `sh -c`
+			// to expand against its `environment:` block. (2) `\"` instead
+			// of `"` — this string is rendered inside a double-quoted YAML
+			// flow scalar (`["CMD-SHELL", "..."]`), so every literal `"`
+			// must be YAML-escaped or it terminates the scalar early and
+			// corrupts the generated compose file.
+			return `/opt/mssql-tools18/bin/sqlcmd -S localhost -U sa -P \"$$MSSQL_SA_PASSWORD\" -C -Q 'SELECT 1' -b || exit 1`
+		},
+		BuildInitCmd: func(password, database string) string {
+			// CREATE DATABASE must be the only statement in its batch,
+			// hence two separate sqlcmd invocations. Same `$$`/`\"`
+			// escaping requirement as BuildHealthCmd (rendered into
+			// "{{ ... }}" in the compose template).
+			return fmt.Sprintf(
+				`set -e; /opt/mssql-tools18/bin/sqlcmd -S database -U sa -P \"$$MSSQL_SA_PASSWORD\" -C -Q \"CREATE DATABASE %s\"; `+
+					`/opt/mssql-tools18/bin/sqlcmd -S database -U sa -P \"$$MSSQL_SA_PASSWORD\" -d %s -C -i /db-seed/init.sql; `+
+					`echo database seeding complete`,
+				database, database)
+		},
+		// SQL Server 2022 (Linux) reads its TLS cert/key from mssql.conf at
+		// startup. We mount the generated RSA server.crt/server.key at these
+		// paths and a mssql.conf that points at them. forceencryption=0 leaves
+		// TLS optional server-side (the healthcheck's plain sqlcmd -C still
+		// connects) while the director's encrypt=true still negotiates TLS and
+		// verifies the cert against the mounted CA.
+		TLSServerCertPath: "/etc/ssl/mssql/server.crt",
+		TLSServerKeyPath:  "/etc/ssl/mssql/server.key",
+		BuildTLSConf: func() (string, string) {
+			return "/var/opt/mssql/mssql.conf", "[network]\n" +
+				"tlscert = /etc/ssl/mssql/server.crt\n" +
+				"tlskey = /etc/ssl/mssql/server.key\n" +
+				"tlsprotocols = 1.2\n" +
+				"forceencryption = 0\n"
+		},
+	},
+	"mysql": {
+		DefaultImage: "mysql:8.4",
+		// MySQL has no password-complexity policy to satisfy; any value works.
+		DefaultPassword: "PipeBench-Db1!",
+		BuildEnv: func(password string) map[string]string {
+			return map[string]string{
+				"MYSQL_ROOT_PASSWORD": password,
+				// The official image only creates root@localhost by default.
+				// database-init and the director subject both connect over
+				// the compose network (hostname "database", not localhost),
+				// so root needs a host wildcard grant or every non-local
+				// connection is refused regardless of password.
+				"MYSQL_ROOT_HOST": "%",
+			}
+		},
+		BuildHealthCmd: func(password string) string {
+			// MUST target TCP, not the Unix socket. The official mysql image runs
+			// a temporary --skip-networking server (socket up, no TCP listener)
+			// while it inits the data dir; a socket probe (`-h localhost`) reports
+			// healthy during that phase — a false positive that lets database-init
+			// (which connects over TCP via `mysql -h database`) race ahead and hit
+			// "Can't connect ... 3306 (111)". `-h 127.0.0.1 --protocol=TCP` fails
+			// until the real networked server binds TCP, so healthy means the port
+			// is actually accepting connections. Mirrors the postgres entry's
+			// `pg_isready -h 127.0.0.1`. (mysqladmin ping reports the server alive
+			// even on auth error, so it passes as soon as TCP is up.) Same two
+			// escaping layers: `$$` survives compose's own `$VAR` interpolation to
+			// reach the container shell as a literal `$`, and `\"` survives the
+			// double-quoted YAML flow scalar this string is rendered into.
+			return `mysqladmin ping -h 127.0.0.1 --protocol=TCP -uroot -p\"$$MYSQL_ROOT_PASSWORD\" --silent || exit 1`
+		},
+		BuildInitCmd: func(password, database string) string {
+			// Connects over TCP to hostname "database" (not localhost), so
+			// this is the codepath MYSQL_ROOT_HOST=% exists for. `IF NOT
+			// EXISTS` keeps the command idempotent; same `$$`/`\"` escaping
+			// as BuildHealthCmd.
+			return fmt.Sprintf(
+				`set -e; mysql -h database -uroot -p\"$$MYSQL_ROOT_PASSWORD\" -e \"CREATE DATABASE IF NOT EXISTS %s\"; `+
+					`mysql -h database -uroot -p\"$$MYSQL_ROOT_PASSWORD\" %s < /db-seed/init.sql; `+
+					`echo database seeding complete`,
+				database, database)
+		},
+		// The official image includes /etc/mysql/conf.d/*.cnf from its base
+		// my.cnf. Dropping a conf file there overrides the image's own
+		// auto-generated (ephemeral, untrusted) SSL cert with the harness's
+		// CA-signed one.
+		TLSServerCertPath: "/etc/mysql/certs/server.crt",
+		TLSServerKeyPath:  "/etc/mysql/certs/server.key",
+		BuildTLSConf: func() (string, string) {
+			return "/etc/mysql/conf.d/tls.cnf", "[mysqld]\n" +
+				"ssl-cert = /etc/mysql/certs/server.crt\n" +
+				"ssl-key = /etc/mysql/certs/server.key\n"
+		},
+	},
+	"postgres": {
+		DefaultImage: "postgres:17",
+		// pg has no password-complexity policy; reuse the mssql credential for
+		// parity across engines. Test-only credential.
+		DefaultPassword: "PipeBench-Db1!",
+		BuildEnv: func(password string) map[string]string {
+			// The official image only auto-creates the `postgres` superuser;
+			// the bench db is created by BuildInitCmd. POSTGRES_PASSWORD is
+			// required for TCP auth from database-init and the subject.
+			return map[string]string{
+				"POSTGRES_PASSWORD": password,
+			}
+		},
+		BuildHealthCmd: func(password string) string {
+			// pg_isready needs no auth (no password/escaping here), but it MUST
+			// target TCP, not the Unix socket: the official postgres image runs a
+			// temporary socket-only server (listen_addresses='') while it inits the
+			// data dir, and a socket pg_isready reports "ready" during that phase — a
+			// false positive that lets database-init (which connects over TCP via
+			// `psql -h database`) race ahead of the real network listener. -h forces
+			// the TCP path; 127.0.0.1 (not "localhost") pins IPv4 to match the
+			// server's listener and avoid a localhost->::1 resolution stall.
+			return `pg_isready -h 127.0.0.1 -U postgres -d postgres || exit 1`
+		},
+		BuildInitCmd: func(password, database string) string {
+			// CREATE DATABASE cannot run inside a transaction, so it is its own
+			// psql -c invocation, then the seed file runs against it. Same
+			// `$$`/`\"` escaping requirement as the mssql init (rendered into
+			// "{{ ... }}" in the compose template): `$$` survives compose's own
+			// $VAR interpolation to reach the container shell as a literal `$`;
+			// `\"` survives the double-quoted YAML flow scalar. ON_ERROR_STOP
+			// makes psql fail the container on any SQL error.
+			return fmt.Sprintf(
+				`set -e; PGPASSWORD=\"$$POSTGRES_PASSWORD\" psql -h database -U postgres -v ON_ERROR_STOP=1 -c \"CREATE DATABASE %s\"; `+
+					`PGPASSWORD=\"$$POSTGRES_PASSWORD\" psql -h database -U postgres -d %s -v ON_ERROR_STOP=1 -f /db-seed/init.sql; `+
+					`echo database seeding complete`,
+				database, database)
+		},
+		TLSServerCertPath: "/etc/pg/certs/server.crt",
+		TLSServerKeyPath:  "/etc/pg/certs/server.key",
+		// Postgres drives TLS via -c args (BuildTLSCommand), not a config file,
+		// so this returns an inert mount that keeps the orchestrator's
+		// unconditional conf-mount valid without special-casing. Postgres never
+		// reads this path.
+		BuildTLSConf: func() (string, string) {
+			return "/etc/pg/pipebench-unused.conf",
+				"# PipeBench: postgres TLS is configured via -c args; this file is unused\n"
+		},
+		BuildTLSCommand: func(certPath, keyPath string) string {
+			// Copy the world-readable (0644) mounted key to a postgres-owned
+			// 0600 copy — postgres refuses to start with a group/world-readable
+			// key — then start postgres with TLS on (no auto-include conf dir,
+			// so ssl settings are passed via -c). Runs as the image's default
+			// root user; docker-entrypoint.sh then steps down to the postgres
+			// user, which reads the 0600 copy.
+			const keyCopy = "/var/lib/postgresql/pipebench-server.key"
+			return fmt.Sprintf(
+				"install -m 600 -o postgres -g postgres %s %s && "+
+					"exec docker-entrypoint.sh postgres -c ssl=on -c ssl_cert_file=%s -c ssl_key_file=%s",
+				keyPath, keyCopy, certPath, keyCopy)
+		},
+	},
+	"oracle": {
+		DefaultImage: "gvenzl/oracle-free:23-slim-faststart",
+		// Oracle has no password-complexity policy for these accounts; reuse the
+		// shared credential for parity across engines. Test-only credential.
+		DefaultPassword: "PipeBench-Db1!",
+		BuildEnv: func(password string) map[string]string {
+			// The gvenzl image sets the SYS/SYSTEM password from ORACLE_PASSWORD
+			// and opens the FREEPDB1 pluggable database. The bench device
+			// connects as SYSTEM to FREEPDB1 — Oracle has no CREATE DATABASE, so
+			// a "database" is a schema and the seeded table lives in SYSTEM's.
+			return map[string]string{
+				"ORACLE_PASSWORD": password,
+			}
+		},
+		BuildHealthCmd: func(password string) string {
+			// The gvenzl image ships healthcheck.sh on PATH; it returns success
+			// once the DB is open and accepting connections. No credentials or
+			// escaping needed (contrast the mssql/mysql CLI probes).
+			return "healthcheck.sh"
+		},
+		BuildInitCmd: func(password, database string) string {
+			// Oracle has no CREATE DATABASE (the database param is unused): the
+			// seed runs as SYSTEM against the pre-created FREEPDB1 PDB. A single-
+			// line brace group pipes the mounted seed — bracketed by WHENEVER
+			// SQLERROR + EXIT so any SQL error fails the one-shot container and
+			// sqlplus always terminates — into sqlplus. `set -e` makes a failed
+			// pipeline abort before the success echo (otherwise the trailing echo
+			// would mask a seeding failure). The password comes from
+			// $ORACLE_PASSWORD, never string-built: `$$` survives compose's own
+			// $VAR interpolation to reach the container shell as a literal `$`.
+			// Two escaping layers, same as the mssql/mysql/postgres entries: `$$`
+			// for compose interpolation, and `\"` because this string is rendered
+			// into a double-quoted YAML flow scalar (`- "{{ ... }}"`) — a bare `"`
+			// would terminate the scalar. Connects over TCP to hostname "database".
+			return `set -e; { echo 'WHENEVER SQLERROR EXIT SQL.SQLCODE'; cat /db-seed/init.sql; echo; echo EXIT; } | ` +
+				`sqlplus -S -L system/\"$$ORACLE_PASSWORD\"@//database:1521/FREEPDB1; echo database seeding complete`
+		},
+	},
+}
diff --git a/internal/orchestrator/database.go b/internal/orchestrator/database.go
new file mode 100644
index 0000000..cae015e
--- /dev/null
+++ b/internal/orchestrator/database.go
@@ -0,0 +1,51 @@
+package orchestrator
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/VirtualMetric/PipeBench/internal/config"
+)
+
+// PrepareDatabase writes the seed SQL for a `database:`-enabled case to the
+// run temp dir, so database-init can mount it read-only and run it via the
+// engine's CLI once the server is healthy.
+func PrepareDatabase(tmpDir string, d *config.DatabaseConfig) (string, error) {
+	seedPath := filepath.Join(tmpDir, "db-seed.sql")
+	if err := os.WriteFile(seedPath, []byte(d.SeedSQL), 0o644); err != nil {
+		return "", fmt.Errorf("writing database seed sql: %w", err)
+	}
+	return seedPath, nil
+}
+
+// DatabaseTLSPaths holds the host paths produced for a TLS-enabled database.
+type DatabaseTLSPaths struct {
+	// CertDir contains ca.crt, server.crt and server.key.
+	CertDir string
+	// ConfPath is the engine TLS config file to mount (empty if the engine
+	// has no BuildTLSConf).
+	ConfPath string
+}
+
+// PrepareDatabaseTLS generates a per-run CA + RSA server cert (SAN "database")
+// and writes the engine's TLS config file into tmpDir. server.crt/server.key
+// are mounted into the database container; ca.crt is mounted into the subject
+// so a device can verify the server certificate against it.
+func PrepareDatabaseTLS(tmpDir string, engine config.DatabaseEngine) (DatabaseTLSPaths, error) {
+	certDir := filepath.Join(tmpDir, "db-certs")
+	if _, err := GenerateDatabaseTLSCerts(certDir, []string{"database"}); err != nil {
+		return DatabaseTLSPaths{}, fmt.Errorf("generating database tls certs: %w", err)
+	}
+
+	paths := DatabaseTLSPaths{CertDir: certDir}
+	if engine.BuildTLSConf != nil {
+		_, content := engine.BuildTLSConf()
+		confPath := filepath.Join(tmpDir, "db-tls.conf")
+		if err := os.WriteFile(confPath, []byte(content), 0o644); err != nil {
+			return DatabaseTLSPaths{}, fmt.Errorf("writing database tls conf: %w", err)
+		}
+		paths.ConfPath = confPath
+	}
+	return paths, nil
+}
diff --git a/internal/orchestrator/docker.go b/internal/orchestrator/docker.go
index eed99de..d2fe7bb 100644
--- a/internal/orchestrator/docker.go
+++ b/internal/orchestrator/docker.go
@@ -1,6 +1,7 @@
 package orchestrator
 
 import (
+	"context"
 	"fmt"
 	"maps"
 	"os"
@@ -184,7 +185,7 @@ services:
     image: "{{ .SubjectImage }}"
     container_name: "{{ .SubjectContainer }}"
     networks: [bench]
-{{- if or .KafkaEnabled .AWSEnabled .AzureEnabled .VaultEnabled .MinioEnabled .PipelineBrokerEnabled .VerifierLocalDir }}
+{{- if or .KafkaEnabled .AWSEnabled .AzureEnabled .VaultEnabled .DatabaseEnabled .MinioEnabled .PipelineBrokerEnabled .VerifierLocalDir }}
     depends_on:
 {{- if .VerifierLocalDir }}
       data-init:
@@ -201,6 +202,10 @@ services:
       vault-init:
         condition: service_completed_successfully
 {{- end }}
+{{- if .DatabaseEnabled }}
+      database-init:
+        condition: service_completed_successfully
+{{- end }}
 {{- if .AWSEnabled }}
       localstack:
         condition: service_healthy
@@ -238,6 +243,9 @@ services:
 {{- if .CaseCertsHost }}
       - "{{ .CaseCertsHost }}:/opt/vmetric/certs:ro"
 {{- end }}
+{{- if .DatabaseCAHost }}
+      - "{{ .DatabaseCAHost }}:/opt/vmetric/certs/ca.pem:ro"
+{{- end }}
 {{- if .KrbHostDir }}
       - "{{ .KrbHostDir }}:/krb5:ro"
 {{- end }}
@@ -991,6 +999,57 @@ services:
     command:
       - "{{ .VaultInitCmd }}"
 {{- end }}
+{{- if .DatabaseEnabled }}
+
+  database:
+    image: "{{ .DatabaseImage }}"
+    container_name: "bench-database"
+    hostname: "database"
+    networks: [bench]
+    environment:
+{{- range $k, $v := .DatabaseEnv }}
+      {{ $k }}: "{{ $v }}"
+{{- end }}
+{{- if .DatabaseTLSEnabled }}
+    volumes:
+      - "{{ .DatabaseServerCertHost }}:{{ .DatabaseServerCertPath }}:ro"
+      - "{{ .DatabaseServerKeyHost }}:{{ .DatabaseServerKeyPath }}:ro"
+      - "{{ .DatabaseConfHost }}:{{ .DatabaseConfPath }}:ro"
+{{- end }}
+{{- if .DatabaseCommand }}
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+      - "{{ .DatabaseCommand }}"
+{{- end }}
+    healthcheck:
+      test: ["CMD-SHELL", "{{ .DatabaseHealthCmd }}"]
+      interval: 5s
+      timeout: 5s
+      retries: 40
+      start_period: 30s
+    restart: "no"
+
+  # One-shot: create the database and run the seed file, then exit 0. The
+  # subject gates on this completing so the first poll never races an
+  # empty/missing table.
+  database-init:
+    image: "{{ .DatabaseImage }}"
+    container_name: "bench-database-init"
+    networks: [bench]
+    depends_on:
+      database:
+        condition: service_healthy
+    environment:
+{{- range $k, $v := .DatabaseEnv }}
+      {{ $k }}: "{{ $v }}"
+{{- end }}
+    volumes:
+      - "{{ .DatabaseSeedHost }}:/db-seed/init.sql:ro"
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+      - "{{ .DatabaseInitCmd }}"
+    restart: "no"
+{{- end }}
 {{- if .AWSEnabled }}
 
   localstack:
@@ -1120,6 +1179,16 @@ type RunConfig struct {
 	VaultTLSHost     string
 	VaultSecretsHost string
 	VaultSeeds       []VaultSeed
+	// DatabaseSeedHost is the host path holding the seed SQL provisioned by
+	// PrepareDatabase for a `database:`-enabled case. NewComposeRunner
+	// populates it when empty; tests may set it directly.
+	DatabaseSeedHost string
+	// DatabaseCertDir / DatabaseConfHost are provisioned by PrepareDatabaseTLS
+	// for a `database.tls: true` case: DatabaseCertDir holds ca.crt +
+	// server.crt + server.key, DatabaseConfHost is the engine TLS config file.
+	// NewComposeRunner populates them when empty.
+	DatabaseCertDir  string
+	DatabaseConfHost string
 	// KrbHostDir / KerberosInitCmd are provisioned by PrepareKerberos for a
 	// `kafka.auth.mechanism: gssapi` case. NewComposeRunner populates them when
 	// empty; tests may set them directly to render without touching the KDC.
@@ -1129,6 +1198,12 @@ type RunConfig struct {
 
 // ComposeRunner manages a docker compose lifecycle for one test run.
 type ComposeRunner struct {
+	// runCtx is cancelled on SIGINT/SIGTERM. Forward-path operations (up,
+	// waits) run under it so an interrupt unwinds them; teardown operations
+	// (Down/Stop/Kill) deliberately use fresh bounded contexts instead, so
+	// cleanup still runs after cancellation. Run-scoped, so storing the ctx
+	// on the struct is the accepted exception to "don't put ctx in structs".
+	runCtx      context.Context
 	cfg         RunConfig
 	composeFile string
 
@@ -1146,7 +1221,12 @@ type ComposeRunner struct {
 var _ Orchestrator = (*ComposeRunner)(nil)
 
 // NewComposeRunner creates a ComposeRunner and writes the compose file.
-func NewComposeRunner(cfg RunConfig) (*ComposeRunner, error) {
+// ctx should be the run-scoped interrupt context; nil means "never
+// interrupted".
+func NewComposeRunner(ctx context.Context, cfg RunConfig) (*ComposeRunner, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	if err := os.MkdirAll(cfg.TmpDir, 0o755); err != nil {
 		return nil, err
 	}
@@ -1166,6 +1246,32 @@ func NewComposeRunner(cfg RunConfig) (*ComposeRunner, error) {
 		cfg.VaultTLSHost, cfg.VaultSecretsHost, cfg.VaultSeeds = vp.TLSDir, vp.SecretsDir, vp.Seeds
 	}
 
+	// Database backend topology prep, same pattern as Vault: a `database:`
+	// case needs the seed SQL file on disk before the compose file is
+	// rendered so database-init can bind-mount it.
+	if cfg.TestCase.UsesDatabase() && cfg.DatabaseSeedHost == "" {
+		seedPath, err := PrepareDatabase(cfg.TmpDir, cfg.TestCase.Database)
+		if err != nil {
+			return nil, fmt.Errorf("preparing database topology: %w", err)
+		}
+		cfg.DatabaseSeedHost = seedPath
+	}
+
+	// Database TLS prep: generate a CA + RSA server cert and the engine TLS
+	// config so the database container presents a verifiable cert and the
+	// subject can trust the CA. Same populate-when-empty pattern.
+	if cfg.TestCase.UsesDatabaseTLS() && cfg.DatabaseCertDir == "" {
+		engine, ok := config.DatabaseEngines[cfg.TestCase.Database.Engine]
+		if !ok {
+			return nil, fmt.Errorf("preparing database tls: unknown engine %q", cfg.TestCase.Database.Engine)
+		}
+		tp, err := PrepareDatabaseTLS(cfg.TmpDir, engine)
+		if err != nil {
+			return nil, fmt.Errorf("preparing database tls: %w", err)
+		}
+		cfg.DatabaseCertDir, cfg.DatabaseConfHost = tp.CertDir, tp.ConfPath
+	}
+
 	// Kerberos topology prep, same pattern as Vault: a gssapi case needs the
 	// KDC bootstrap command + the shared krb5 dir before the compose file is
 	// rendered.
@@ -1182,7 +1288,7 @@ func NewComposeRunner(cfg RunConfig) (*ComposeRunner, error) {
 		return nil, err
 	}
 
-	cr := &ComposeRunner{cfg: cfg, composeFile: composeFile}
+	cr := &ComposeRunner{runCtx: ctx, cfg: cfg, composeFile: composeFile}
 	cr.populateServiceNames()
 	return cr, nil
 }
@@ -1289,25 +1395,36 @@ func (r *ComposeRunner) resolveServiceAliases(names []string) []string {
 
 // StopServices sends SIGTERM to named services with the given grace timeout
 // before SIGKILL. Uses `docker compose stop -t <seconds> <svc>...`.
+// Runs on a fresh bounded context (grace + 30s margin) so it still works
+// after an interrupt and can't hang on a wedged daemon.
 func (r *ComposeRunner) StopServices(timeout time.Duration, services ...string) error {
 	secs := max(int(timeout.Seconds()), 0)
+	ctx, cancel := context.WithTimeout(context.Background(), timeout+30*time.Second)
+	defer cancel()
 	resolved := r.resolveServiceAliases(services)
 	args := append([]string{"stop", "-t", strconv.Itoa(secs)}, resolved...)
-	return r.compose(args...)
+	return r.composeCtx(ctx, args...)
 }
 
 // KillServices sends SIGKILL immediately — no grace period, no chance for
 // the process to clean up. Mirrors a crash scenario for persistence tests.
 // Uses `docker compose kill -s SIGKILL <svc>...`.
 func (r *ComposeRunner) KillServices(services ...string) error {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
 	resolved := r.resolveServiceAliases(services)
 	args := append([]string{"kill", "-s", "SIGKILL"}, resolved...)
-	return r.compose(args...)
+	return r.composeCtx(ctx, args...)
 }
 
-// Down stops and removes all containers and anonymous volumes.
+// Down stops and removes all containers and anonymous volumes. It is the
+// deferred teardown for every flow, so it runs on a fresh bounded context:
+// it must proceed after the run context is cancelled by an interrupt, and
+// it must not block forever on a wedged daemon.
 func (r *ComposeRunner) Down() error {
-	return r.compose("down", "-v", "--remove-orphans")
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+	return r.composeCtx(ctx, "down", "-v", "--remove-orphans")
 }
 
 // WaitForGeneratorExit blocks until ALL generator containers exit or the
@@ -1319,6 +1436,9 @@ func (r *ComposeRunner) WaitForGeneratorExit(timeout time.Duration) error {
 	pending := append([]string(nil), r.genContainers...)
 	var firstErr error
 	for time.Now().Before(deadline) {
+		if err := r.runCtx.Err(); err != nil {
+			return fmt.Errorf("interrupted while waiting for generator(s): %w", err)
+		}
 		var still []string
 		for _, name := range pending {
 			out, err := exec.Command("docker", "inspect", "--format={{.State.Status}}", name).Output()
@@ -1339,7 +1459,11 @@ func (r *ComposeRunner) WaitForGeneratorExit(timeout time.Duration) error {
 			return firstErr
 		}
 		pending = still
-		time.Sleep(2 * time.Second)
+		select {
+		case <-r.runCtx.Done():
+			return fmt.Errorf("interrupted while waiting for generator(s): %w", r.runCtx.Err())
+		case <-time.After(2 * time.Second):
+		}
 	}
 	return fmt.Errorf("generator(s) did not exit within %s (still running: %v)", timeout, pending)
 }
@@ -1357,7 +1481,11 @@ func (r *ComposeRunner) WaitForVerifierExit(timeout time.Duration) error {
 		if err == nil && strings.TrimSpace(string(out)) == "exited" {
 			return nil
 		}
-		time.Sleep(2 * time.Second)
+		select {
+		case <-r.runCtx.Done():
+			return fmt.Errorf("interrupted while waiting for verifier: %w", r.runCtx.Err())
+		case <-time.After(2 * time.Second):
+		}
 	}
 	return fmt.Errorf("verifier did not exit within %s", timeout)
 }
@@ -1385,19 +1513,25 @@ func (r *ComposeRunner) Logs(service string, lines int) string {
 	return string(out)
 }
 
-// StopCollector sends SIGTERM to the collector container and waits for it to exit.
-// The collector writes its CSV file on SIGTERM, so this must be called before CopyMetricsCSV.
+// StopCollector sends SIGTERM to the collector container and waits for it to
+// exit. The collector fsyncs each CSV row as it samples, so callers copy the
+// CSV first (while the collector still runs) and stop it after — see the
+// teardown order in runner.Run. Bounded so it works after an interrupt and
+// can't hang on a wedged daemon.
 func (r *ComposeRunner) StopCollector() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
 	// docker stop sends SIGTERM and waits up to 10s for graceful exit.
-	out, err := exec.Command("docker", "stop", "-t", "10", "bench-collector").CombinedOutput()
+	out, err := exec.CommandContext(ctx, "docker", "stop", "-t", "10", "bench-collector").CombinedOutput()
 	if err != nil {
 		return fmt.Errorf("docker stop bench-collector: %w\n%s", err, out)
 	}
 	return nil
 }
 
-// CopyMetricsCSV copies the metrics CSV from the (stopped) collector container to dst.
-// Call StopCollector first so the CSV has been flushed.
+// CopyMetricsCSV copies the metrics CSV from the (still running) collector
+// container to dst. Safe before StopCollector because the collector fsyncs
+// every row as it writes it.
 func (r *ComposeRunner) CopyMetricsCSV(dst string) error {
 	src := "bench-collector:/results/metrics.csv"
 	out, err := exec.Command("docker", "cp", src, dst).CombinedOutput()
@@ -1464,9 +1598,16 @@ func (r *ComposeRunner) GeneratorStdout() string {
 	return b.String()
 }
 
+// compose runs a docker compose subcommand under the run context: the first
+// interrupt cancels it. Teardown paths must use composeCtx with a fresh
+// bounded context instead, so cleanup still runs after cancellation.
 func (r *ComposeRunner) compose(args ...string) error {
+	return r.composeCtx(r.runCtx, args...)
+}
+
+func (r *ComposeRunner) composeCtx(ctx context.Context, args ...string) error {
 	cmdArgs := append([]string{"compose", "-f", r.composeFile}, args...)
-	cmd := exec.Command("docker", cmdArgs...)
+	cmd := exec.CommandContext(ctx, "docker", cmdArgs...)
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
 	return cmd.Run()
@@ -1685,6 +1826,35 @@ type composeVars struct {
 	VaultSecretsHost string
 	VaultInitCmd     string
 
+	// Database backend topology. DatabaseEnabled gates the database +
+	// database-init services and the subject's database-init depends_on.
+	// DatabaseEnv/DatabaseHealthCmd/DatabaseInitCmd are entirely built by
+	// the selected config.DatabaseEngine — this file has no engine-specific
+	// knowledge.
+	DatabaseEnabled   bool
+	DatabaseImage     string
+	DatabaseEnv       map[string]string
+	DatabaseHealthCmd string
+	DatabaseInitCmd   string
+	DatabaseSeedHost  string
+
+	// DatabaseCommand, when non-empty, overrides the database container's
+	// entrypoint with `/bin/sh -c <DatabaseCommand>` (engine BuildTLSCommand).
+	// Empty for engines that use their image defaults (mssql/mysql).
+	DatabaseCommand string
+
+	// Database TLS (gated by DatabaseTLSEnabled). *Host are host paths bind-
+	// mounted into the database container at the engine's *Path targets;
+	// DatabaseCAHost is the CA mounted into the subject so a device can verify.
+	DatabaseTLSEnabled     bool
+	DatabaseServerCertHost string
+	DatabaseServerKeyHost  string
+	DatabaseConfHost       string
+	DatabaseServerCertPath string
+	DatabaseServerKeyPath  string
+	DatabaseConfPath       string
+	DatabaseCAHost         string
+
 	RecvMode              string
 	RecvListen            string
 	RecvEnv               map[string]string
@@ -2230,6 +2400,54 @@ func writeCompose(path string, cfg RunConfig) error {
 		vars.VaultInitCmd = sb.String()
 	}
 
+	// Database backend: render the database + database-init services and
+	// gate the subject on the seeding completing. All engine-specific
+	// knowledge (image, env, healthcheck, init command) comes from the
+	// selected config.DatabaseEngine — this block just wires it through.
+	if tc.UsesDatabase() {
+		if cfg.DatabaseSeedHost == "" {
+			return fmt.Errorf("case %q uses database but the seed sql was not prepared", tc.Name)
+		}
+		engine, ok := config.DatabaseEngines[tc.Database.Engine]
+		if !ok {
+			// Unreachable in practice: TestCase.Validate rejects an unknown
+			// engine before a run ever reaches compose rendering.
+			return fmt.Errorf("case %q: unknown database.engine %q", tc.Name, tc.Database.Engine)
+		}
+		vars.DatabaseEnabled = true
+		vars.DatabaseImage = tc.Database.ImageOrDefault(engine)
+		password := tc.Database.PasswordOrDefault(engine)
+		dbName := tc.Database.DatabaseOrDefault()
+		vars.DatabaseEnv = engine.BuildEnv(password)
+		vars.DatabaseHealthCmd = engine.BuildHealthCmd(password)
+		vars.DatabaseInitCmd = engine.BuildInitCmd(password, dbName)
+		vars.DatabaseSeedHost = filepath.ToSlash(cfg.DatabaseSeedHost)
+
+		// Database TLS: mount the generated server cert/key + engine TLS conf
+		// into the database container and the CA into the subject so a device
+		// can verify the server certificate instead of skipping verification.
+		if tc.UsesDatabaseTLS() {
+			if cfg.DatabaseCertDir == "" || engine.TLSServerCertPath == "" {
+				return fmt.Errorf("case %q uses database.tls but the tls certs were not prepared (or engine %q lacks TLS support)", tc.Name, tc.Database.Engine)
+			}
+			confMount, _ := engine.BuildTLSConf()
+			vars.DatabaseTLSEnabled = true
+			vars.DatabaseServerCertHost = filepath.ToSlash(filepath.Join(cfg.DatabaseCertDir, "server.crt"))
+			vars.DatabaseServerKeyHost = filepath.ToSlash(filepath.Join(cfg.DatabaseCertDir, "server.key"))
+			vars.DatabaseConfHost = filepath.ToSlash(cfg.DatabaseConfHost)
+			vars.DatabaseServerCertPath = engine.TLSServerCertPath
+			vars.DatabaseServerKeyPath = engine.TLSServerKeyPath
+			vars.DatabaseConfPath = confMount
+			vars.DatabaseCAHost = filepath.ToSlash(filepath.Join(cfg.DatabaseCertDir, "ca.crt"))
+
+			// Engines that cannot pick up TLS from a mounted conf file (postgres)
+			// override the container entrypoint to enable TLS at startup.
+			if engine.BuildTLSCommand != nil {
+				vars.DatabaseCommand = engine.BuildTLSCommand(engine.TLSServerCertPath, engine.TLSServerKeyPath)
+			}
+		}
+	}
+
 	if tc.MultiReceiver() {
 		for i, rc := range tc.Receivers {
 			vars.Receivers = append(vars.Receivers, receiverTpl{
diff --git a/internal/orchestrator/tls.go b/internal/orchestrator/tls.go
index e05604f..b112865 100644
--- a/internal/orchestrator/tls.go
+++ b/internal/orchestrator/tls.go
@@ -4,6 +4,7 @@ import (
 	"crypto/ecdsa"
 	"crypto/elliptic"
 	"crypto/rand"
+	"crypto/rsa"
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/pem"
@@ -132,6 +133,98 @@ func GenerateTLSCerts(outDir string, serverHosts []string) (string, error) {
 	return abs, nil
 }
 
+// GenerateDatabaseTLSCerts writes a self-signed CA plus an RSA server leaf
+// (SAN = serverHosts) to outDir, for a database container that terminates TLS.
+// Unlike GenerateTLSCerts (ECDSA), the keys are RSA-2048: SQL Server on Linux
+// only accepts RSA certificates for its TLS endpoint. Files written:
+//
+//	ca.crt      — root CA certificate (PEM); the director trusts this to verify the server
+//	server.crt  — server leaf with SAN serverHosts (PEM), signed by ca.crt
+//	server.key  — server leaf private key (RSA, PEM)
+//
+// server.key is world-readable (0644) on purpose: it is bind-mounted read-only
+// into the database container, whose engine process runs as a non-root user
+// that differs from the host uid writing the file (mssql: uid 10001, mysql:
+// uid 999 — the exact uid is engine-specific, but the boundary it crosses is
+// not). Owner-only modes (0400/0600) would be unreadable across that uid
+// boundary and break the TLS endpoint; chown needs host root. Per-run
+// throwaway cert in a temp dir, removed with it — never real key material.
+func GenerateDatabaseTLSCerts(outDir string, serverHosts []string) (string, error) {
+	abs, err := filepath.Abs(outDir)
+	if err != nil {
+		return "", err
+	}
+	if err := os.MkdirAll(abs, 0o755); err != nil {
+		return "", fmt.Errorf("creating cert dir: %w", err)
+	}
+
+	// 1. Root CA
+	caKey, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		return "", fmt.Errorf("generate ca key: %w", err)
+	}
+	caTpl := &x509.Certificate{
+		SerialNumber:          bigSerial(),
+		Subject:               pkix.Name{CommonName: "PipeBench Database CA"},
+		NotBefore:             time.Now().Add(-1 * time.Hour),
+		NotAfter:              time.Now().Add(24 * time.Hour),
+		IsCA:                  true,
+		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageCRLSign,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+	}
+	caDER, err := x509.CreateCertificate(rand.Reader, caTpl, caTpl, &caKey.PublicKey, caKey)
+	if err != nil {
+		return "", fmt.Errorf("sign ca cert: %w", err)
+	}
+	if err := writePEMCert(filepath.Join(abs, "ca.crt"), caDER); err != nil {
+		return "", err
+	}
+
+	// 2. Server leaf (the database endpoint)
+	if len(serverHosts) == 0 {
+		serverHosts = []string{"database"}
+	}
+	serverDNS, serverIP := splitHosts(serverHosts)
+	srvKey, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		return "", fmt.Errorf("generate server key: %w", err)
+	}
+	srvTpl := &x509.Certificate{
+		SerialNumber: bigSerial(),
+		Subject:      pkix.Name{CommonName: serverHosts[0]},
+		NotBefore:    time.Now().Add(-1 * time.Hour),
+		NotAfter:     time.Now().Add(24 * time.Hour),
+		KeyUsage:     x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		DNSNames:     serverDNS,
+		IPAddresses:  serverIP,
+	}
+	srvDER, err := x509.CreateCertificate(rand.Reader, srvTpl, caTpl, &srvKey.PublicKey, caKey)
+	if err != nil {
+		return "", fmt.Errorf("sign server cert: %w", err)
+	}
+	if err := writePEMCert(filepath.Join(abs, "server.crt"), srvDER); err != nil {
+		return "", err
+	}
+	if err := writeRSAKey(filepath.Join(abs, "server.key"), srvKey, 0o644); err != nil {
+		return "", err
+	}
+
+	return abs, nil
+}
+
+// writeRSAKey writes an RSA private key as a PKCS#1 PEM block (SQL Server
+// accepts this form). See writePEMKey for the ECDSA equivalent.
+func writeRSAKey(path string, key *rsa.PrivateKey, perm os.FileMode) error {
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, perm)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	return pem.Encode(f, &pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(key)})
+}
+
 // RotateServerCert re-signs a fresh server leaf (new key, new serial, fresh
 // validity, same SAN set) using the CA key/cert already present in outDir
 // (ca.crt + ca.key), overwriting server.crt / server.key in place. The CA is
diff --git a/internal/runner/runner.go b/internal/runner/runner.go
index 64ecac6..7d51d32 100644
--- a/internal/runner/runner.go
+++ b/internal/runner/runner.go
@@ -111,6 +111,9 @@ type GeneratorResult struct {
 
 // Runner executes a single test case against a single subject.
 type Runner struct {
+	// ctx is cancelled only by SIGINT/SIGTERM — it never carries a
+	// deadline — so context.Canceled always means "user interrupt".
+	ctx   context.Context
 	opts  Options
 	store *results.Store
 }
@@ -125,10 +128,16 @@ func hardwareID() string {
 	return "custom"
 }
 
-// New creates a Runner.
-func New(opts Options) *Runner {
+// New creates a Runner. ctx should be cancelled on SIGINT/SIGTERM so
+// in-flight waits unwind through their error paths and deferred teardown
+// still runs; nil means "never interrupted".
+func New(ctx context.Context, opts Options) *Runner {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	opts.applyDefaults()
 	return &Runner{
+		ctx:   ctx,
 		opts:  opts,
 		store: results.NewStore(opts.ResultsDir),
 	}
@@ -350,7 +359,7 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 		TLSCertsHost:     tlsCertsHost,
 	}
 
-	cr, err := orchestrator.NewComposeRunner(runCfg)
+	cr, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -418,6 +427,11 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 
 		fmt.Printf("  waiting for generator (up to %s)…\n", genTimeout)
 		if err := orch.WaitForGeneratorExit(genTimeout); err != nil {
+			if errors.Is(err, context.Canceled) {
+				// A user interrupt must never be misread as an expected
+				// negative-test outcome — abort the run instead.
+				return results.RunResult{}, err
+			}
 			if tc.Correctness.ExpectFailure {
 				// A negative test EXPECTS the data path to fail — e.g. the
 				// generator gets 401s because auth correctly rejects a wrong
@@ -440,8 +454,8 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 			headStart = rem
 		}
 		fmt.Printf("  no generator — letting the subject run (head start %s)…\n", headStart)
-		if headStart > 0 {
-			time.Sleep(headStart)
+		if err := sleepCtx(r.ctx, headStart); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
 		}
 	}
 
@@ -479,7 +493,9 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 			var drainLast int64
 			drainStart := time.Now()
 			for time.Now().Before(drainDeadline) {
-				time.Sleep(drainPoll)
+				if err := sleepCtx(r.ctx, drainPoll); err != nil {
+					return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+				}
 				var totalLines int64
 				if tc.MultiReceiver() {
 					agg, _, qerr := r.aggregateReceivers(ports, 10*time.Second)
@@ -513,7 +529,9 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 			}
 			if drainGrace > 0 {
 				fmt.Printf("  waiting post-send receive grace (%s)…\n", drainGrace)
-				time.Sleep(drainGrace)
+				if err := sleepCtx(r.ctx, drainGrace); err != nil {
+					return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+				}
 			}
 		} else {
 			// Correctness tests need completeness rather than a fixed SLA window:
@@ -541,7 +559,9 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 			var drainStable int
 			var drainLast int64
 			for time.Now().Before(drainDeadline) {
-				time.Sleep(drainPoll)
+				if err := sleepCtx(r.ctx, drainPoll); err != nil {
+					return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+				}
 				var totalLines int64
 				if tc.MultiReceiver() {
 					agg, _, qerr := r.aggregateReceivers(ports, 10*time.Second)
@@ -915,7 +935,7 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 		// canonical results file the web UI consumes.
 		fmt.Println("  done. (drain mode — result not persisted)")
 	} else {
-		dir, err := r.store.Save(result, metricsCSVSrc)
+		dir, err := r.saveResult(result, metricsCSVSrc)
 		if err != nil {
 			return result, fmt.Errorf("saving results: %w", err)
 		}
@@ -1058,7 +1078,7 @@ func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.S
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	cr, err := orchestrator.NewComposeRunner(runCfg)
+	cr, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -1106,7 +1126,9 @@ func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.S
 
 	// PHASE 3: Wait for subject to persist all buffered data
 	fmt.Println("  phase 3: waiting for subject to persist data…")
-	time.Sleep(10 * time.Second)
+	if err := sleepCtx(r.ctx, 10*time.Second); err != nil {
+		return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+	}
 
 	// PHASE 4: Start receiver — subject should now forward persisted logs
 	fmt.Println("  phase 4: starting receiver (subject should forward buffered logs)…")
@@ -1129,7 +1151,9 @@ func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.S
 	stableRounds := 0
 	drainDeadline := time.Now().Add(drainTimeout)
 	for time.Now().Before(drainDeadline) {
-		time.Sleep(5 * time.Second)
+		if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 		rm, err := r.queryReceiverMetrics(metricsPort, 10*time.Second)
 		if err != nil {
 			continue
@@ -1224,7 +1248,7 @@ func (r *Runner) runPersistenceCorrectness(tc *config.TestCase, subject config.S
 		result.FailReason = strings.Join(errors, "; ")
 	}
 
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -1333,7 +1357,7 @@ func (r *Runner) runPersistenceShutdownCorrectness(tc *config.TestCase, subject
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	cr, err := orchestrator.NewComposeRunner(runCfg)
+	cr, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -1401,7 +1425,9 @@ func (r *Runner) runPersistenceShutdownCorrectness(tc *config.TestCase, subject
 	}
 
 	// Small pause so receiver is fully ready before subject comes back online
-	time.Sleep(3 * time.Second)
+	if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+		return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+	}
 
 	// PHASE 5: Restart subject — it should read persisted logs and forward them
 	fmt.Println("  phase 5: restarting subject (should replay persisted logs)…")
@@ -1423,7 +1449,9 @@ func (r *Runner) runPersistenceShutdownCorrectness(tc *config.TestCase, subject
 	stableRounds := 0
 	drainDeadline := time.Now().Add(drainTimeout)
 	for time.Now().Before(drainDeadline) {
-		time.Sleep(5 * time.Second)
+		if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 		rm, err := r.queryReceiverMetrics(metricsPort, 10*time.Second)
 		if err != nil {
 			continue
@@ -1516,7 +1544,7 @@ func (r *Runner) runPersistenceShutdownCorrectness(tc *config.TestCase, subject
 		result.FailReason = strings.Join(errors, "; ")
 	}
 
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -1649,7 +1677,7 @@ func (r *Runner) runKafkaMidDeliveryAction(tc *config.TestCase, subject config.S
 		}
 	}
 
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -1706,7 +1734,9 @@ func (r *Runner) runKafkaMidDeliveryAction(tc *config.TestCase, subject config.S
 				break
 			}
 		}
-		time.Sleep(500 * time.Millisecond)
+		if err := sleepCtx(r.ctx, 500*time.Millisecond); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 	}
 	if !fired {
 		return results.RunResult{}, fmt.Errorf("never reached mid-delivery (%s) before timeout", formatCount(mid))
@@ -1718,6 +1748,10 @@ func (r *Runner) runKafkaMidDeliveryAction(tc *config.TestCase, subject config.S
 	warmup := tc.WarmupOrDefault(30 * time.Second)
 	genTimeout := min(duration+warmup+2*time.Minute, r.opts.Timeout)
 	if err := orch.WaitForGeneratorExit(genTimeout); err != nil {
+		if errors.Is(err, context.Canceled) {
+			// A user interrupt is not a tolerable generator hiccup — abort.
+			return results.RunResult{}, err
+		}
 		fmt.Printf("  (generator wait: %v)\n", err)
 	}
 	genStats := r.parseGeneratorStats(orch.GeneratorStdout())
@@ -1730,7 +1764,9 @@ func (r *Runner) runKafkaMidDeliveryAction(tc *config.TestCase, subject config.S
 	stableRounds := 0
 	drainDeadline := time.Now().Add(drainTimeout)
 	for time.Now().Before(drainDeadline) {
-		time.Sleep(5 * time.Second)
+		if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 		rm, qerr := r.queryReceiverMetrics(metricsPort, 10*time.Second)
 		if qerr != nil {
 			continue
@@ -1809,7 +1845,7 @@ func (r *Runner) runKafkaMidDeliveryAction(tc *config.TestCase, subject config.S
 
 	// Persist the result like every other run path — Run's contract is to
 	// return the *persisted* result.
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -1852,7 +1888,9 @@ func (r *Runner) runKafkaInflightCrash(tc *config.TestCase, subject config.Subje
 				return fmt.Errorf("killing subject: %w", err)
 			}
 			// Settle before the consumer rejoins and replays uncommitted offsets.
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return fmt.Errorf("interrupted: %w", err)
+			}
 			if err := orch.UpServices("subject"); err != nil {
 				return fmt.Errorf("restarting subject: %w", err)
 			}
@@ -1911,7 +1949,9 @@ func (r *Runner) runKafkaCertRotation(tc *config.TestCase, subject config.Subjec
 			// Give the broker time to come back and the consumer time to
 			// bounce-detect and retry against the untrusted leaf (each attempt
 			// should fail certificate verification).
-			time.Sleep(25 * time.Second)
+			if err := sleepCtx(r.ctx, 25*time.Second); err != nil {
+				return fmt.Errorf("interrupted: %w", err)
+			}
 
 			lines, after := subjectLogStats(subj)
 			if lines == 0 {
@@ -1931,7 +1971,9 @@ func (r *Runner) runKafkaCertRotation(tc *config.TestCase, subject config.Subjec
 			}); err != nil {
 				return err
 			}
-			time.Sleep(5 * time.Second)
+			if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+				return fmt.Errorf("interrupted: %w", err)
+			}
 			fmt.Println("  broker cert restored under the trusted CA — expecting delivery to resume")
 			return nil
 		},
@@ -2100,7 +2142,7 @@ func (r *Runner) runDirectorAgentCertRotation(tc *config.TestCase, subject confi
 		TLSCertsHost:     certsDir,
 	}
 
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -2127,7 +2169,9 @@ func (r *Runner) runDirectorAgentCertRotation(tc *config.TestCase, subject confi
 		if d > rem {
 			return fmt.Errorf("run timeout (%s) exceeded during the rotation wait", r.opts.Timeout)
 		}
-		time.Sleep(d)
+		if err := sleepCtx(r.ctx, d); err != nil {
+			return fmt.Errorf("interrupted: %w", err)
+		}
 		return nil
 	}
 	defer func() {
@@ -2191,7 +2235,9 @@ func (r *Runner) runDirectorAgentCertRotation(tc *config.TestCase, subject confi
 				break
 			}
 		}
-		time.Sleep(2 * time.Second)
+		if err := sleepCtx(r.ctx, 2*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 	}
 	if !established {
 		return results.RunResult{}, fmt.Errorf(
@@ -2327,7 +2373,9 @@ func (r *Runner) runDirectorAgentCertRotation(tc *config.TestCase, subject confi
 		drainDeadline = runDeadline
 	}
 	for time.Now().Before(drainDeadline) {
-		time.Sleep(5 * time.Second)
+		if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 		rm, qerr := r.queryReceiverMetrics(metricsPort, 10*time.Second)
 		if qerr != nil {
 			continue
@@ -2410,7 +2458,7 @@ func (r *Runner) runDirectorAgentCertRotation(tc *config.TestCase, subject confi
 		result.FailReason = strings.Join(errs, "; ")
 	}
 
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -2510,7 +2558,7 @@ func (r *Runner) runDirectorAgentACLRotation(tc *config.TestCase, subject config
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -2531,7 +2579,9 @@ func (r *Runner) runDirectorAgentACLRotation(tc *config.TestCase, subject config
 		if d > rem {
 			return fmt.Errorf("run timeout (%s) exceeded during the rotation wait", r.opts.Timeout)
 		}
-		time.Sleep(d)
+		if err := sleepCtx(r.ctx, d); err != nil {
+			return fmt.Errorf("interrupted: %w", err)
+		}
 		return nil
 	}
 	defer func() {
@@ -2651,7 +2701,9 @@ func (r *Runner) runDirectorAgentACLRotation(tc *config.TestCase, subject config
 					break
 				}
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		started := finalCount >= minRecv
 		if !started {
@@ -2682,7 +2734,9 @@ func (r *Runner) runDirectorAgentACLRotation(tc *config.TestCase, subject config
 					break
 				}
 			}
-			time.Sleep(2 * time.Second)
+			if err := sleepCtx(r.ctx, 2*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		if !established {
 			return results.RunResult{}, fmt.Errorf(
@@ -2766,7 +2820,7 @@ func (r *Runner) runDirectorAgentACLRotation(tc *config.TestCase, subject config
 		result.FailReason = strings.Join(errs, "; ")
 	}
 
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -2845,15 +2899,19 @@ func clusterFormed(containers []string) bool {
 }
 
 // waitClusterReady polls until every node has formed the cluster AND a leader is
-// elected, or the deadline passes. Returns the leader index and whether ready.
-func waitClusterReady(containers []string, deadline time.Time) (int, bool) {
+// elected, or the deadline passes / ctx is cancelled. Returns the leader index
+// and whether ready — callers must check ctx.Err() after a not-ready result so
+// an interrupt isn't misread as "cluster never formed".
+func waitClusterReady(ctx context.Context, containers []string, deadline time.Time) (int, bool) {
 	for time.Now().Before(deadline) {
 		if clusterFormed(containers) {
 			if l, ok := leaderExistsNow(containers); ok {
 				return l, true
 			}
 		}
-		time.Sleep(3 * time.Second)
+		if sleepCtx(ctx, 3*time.Second) != nil {
+			break
+		}
 	}
 	l, _ := leaderExistsNow(containers)
 	return l, false
@@ -2898,7 +2956,7 @@ func agentlessCollectingNow(container string) bool {
 // waitAgentlessCollecting waits until the agentless device is assigned to a node
 // AND that owner is actively collecting (forwarding to the router). Returns the
 // owner node name ("1".."N"), its container, and whether it became ready.
-func waitAgentlessCollecting(subjectName string, containers []string, deadline time.Time) (string, string, bool) {
+func waitAgentlessCollecting(ctx context.Context, subjectName string, containers []string, deadline time.Time) (string, string, bool) {
 	for time.Now().Before(deadline) {
 		if owner := agentlessDeviceOwner(containers); owner != "" {
 			ownerContainer := fmt.Sprintf("bench-subject-%s-%s", subjectName, owner)
@@ -2906,7 +2964,9 @@ func waitAgentlessCollecting(subjectName string, containers []string, deadline t
 				return owner, ownerContainer, true
 			}
 		}
-		time.Sleep(5 * time.Second)
+		if sleepCtx(ctx, 5*time.Second) != nil {
+			break
+		}
 	}
 	return agentlessDeviceOwner(containers), "", false
 }
@@ -2915,7 +2975,7 @@ func waitAgentlessCollecting(subjectName string, containers []string, deadline t
 // (the stopped owner, identified by node name "1".."N") is actively collecting
 // the agentless device — i.e. the device failed over and the new owner re-deployed
 // and resumed collection. Returns the new owner's name, container, and readiness.
-func waitAgentlessCollectingExcluding(containers []string, exclude string, deadline time.Time) (string, string, bool) {
+func waitAgentlessCollectingExcluding(ctx context.Context, containers []string, exclude string, deadline time.Time) (string, string, bool) {
 	for time.Now().Before(deadline) {
 		for i, c := range containers {
 			name := strconv.Itoa(i + 1)
@@ -2926,7 +2986,9 @@ func waitAgentlessCollectingExcluding(containers []string, exclude string, deadl
 				return name, c, true
 			}
 		}
-		time.Sleep(5 * time.Second)
+		if sleepCtx(ctx, 5*time.Second) != nil {
+			break
+		}
 	}
 	return "", "", false
 }
@@ -3032,13 +3094,16 @@ func nodeHasClusterIP(container, ip string) bool {
 	return strings.Contains(string(out), "inet "+ip+"/")
 }
 
-// waitNodeHasClusterIP polls until the node holds ip, or the deadline passes.
-func waitNodeHasClusterIP(container, ip string, deadline time.Time) bool {
+// waitNodeHasClusterIP polls until the node holds ip, or the deadline passes /
+// ctx is cancelled.
+func waitNodeHasClusterIP(ctx context.Context, container, ip string, deadline time.Time) bool {
 	for time.Now().Before(deadline) {
 		if nodeHasClusterIP(container, ip) {
 			return true
 		}
-		time.Sleep(3 * time.Second)
+		if sleepCtx(ctx, 3*time.Second) != nil {
+			break
+		}
 	}
 	return false
 }
@@ -3111,7 +3176,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -3156,7 +3221,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 		formDeadline = runDeadline
 	}
 	fmt.Printf("  waiting for cluster to form + elect a leader (up to %s)…\n", time.Until(formDeadline).Round(time.Second))
-	leader, ready := waitClusterReady(nodes, formDeadline)
+	leader, ready := waitClusterReady(r.ctx, nodes, formDeadline)
 
 	var passed bool
 	var errs []string
@@ -3193,7 +3258,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 
 		if isAgentless {
 			fmt.Printf("  baseline: waiting for the agentless device to deploy + collect (up to %s)…\n", time.Until(drainDeadline).Round(time.Second))
-			owner, ownerContainer, baselineOK = waitAgentlessCollecting(subject.Name, nodes, drainDeadline)
+			owner, ownerContainer, baselineOK = waitAgentlessCollecting(r.ctx, subject.Name, nodes, drainDeadline)
 			if baselineOK {
 				fmt.Printf("  agentless device deployed; owner = node %s (%s), collecting ✓\n", owner, ownerContainer)
 			} else {
@@ -3220,7 +3285,9 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 						break
 					}
 				}
-				time.Sleep(3 * time.Second)
+				if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+					return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+				}
 			}
 			baselineOK = finalCount >= minRecv
 			if !baselineOK {
@@ -3243,7 +3310,9 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				actionOK = false
 				errs = append(errs, fmt.Sprintf("docker restart %s failed: %v (disruption did not happen)", nodes[follower-1], rerr))
 			}
-			time.Sleep(settle)
+			if err := sleepCtx(r.ctx, settle); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 			if _, ok := leaderExistsNow(nodes); !ok {
 				actionOK = false
 				errs = append(errs, "no leader after restarting a follower (cluster unexpectedly lost leadership)")
@@ -3258,7 +3327,9 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				errs = append(errs, fmt.Sprintf("docker restart %s failed: %v (disruption did not happen)", nodes[leader-1], rerr))
 			}
 			restartedAt := time.Now()
-			time.Sleep(settle)
+			if err := sleepCtx(r.ctx, settle); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 			// A leader restart must trigger a re-election. Anchor the search window to
 			// the instant the restart completed so a stale pre-restart "became leader"
 			// line can't satisfy the check; fall back to a leader simply being present.
@@ -3291,7 +3362,9 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 					errs = append(errs, fmt.Sprintf("docker stop %s failed: %v (disruption did not happen)", c, serr))
 				}
 			}
-			time.Sleep(settle)
+			if err := sleepCtx(r.ctx, settle); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 			if _, ok := leaderExistsNow(nodes); ok {
 				fmt.Printf("  note: a leader still appears present with %d/%d down (unexpected — quorum should be lost)\n", len(stopped), tc.Cluster.Nodes)
 			} else {
@@ -3309,7 +3382,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				recDeadline = runDeadline
 			}
 			fmt.Printf("  waiting for cluster to recover + re-elect a leader (up to %s)…\n", time.Until(recDeadline).Round(time.Second))
-			l2, ok2 := waitClusterReady(nodes, recDeadline)
+			l2, ok2 := waitClusterReady(r.ctx, nodes, recDeadline)
 			if !ok2 {
 				actionOK = false
 				errs = append(errs, fmt.Sprintf("cluster did not recover a leader after restarting the two nodes (leaderIdx=%d)", l2))
@@ -3337,7 +3410,9 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 					errs = append(errs, fmt.Sprintf("docker stop %s failed: %v (disruption did not happen)", ownerContainer, serr))
 				}
 				fmt.Printf("  waiting %s for the leader to detect the down node and reassign the device…\n", settle)
-				time.Sleep(settle)
+				if err := sleepCtx(r.ctx, settle); err != nil {
+					return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+				}
 
 				// HARD 1: the device is reassigned AWAY FROM the stopped owner — the
 				//   automatic ownership failover the case exists to prove ("the new owner is
@@ -3368,7 +3443,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				if collectDeadline.After(runDeadline) {
 					collectDeadline = runDeadline
 				}
-				if newOwner, _, ok := waitAgentlessCollectingExcluding(nodes, owner, collectDeadline); ok {
+				if newOwner, _, ok := waitAgentlessCollectingExcluding(r.ctx, nodes, owner, collectDeadline); ok {
 					fmt.Printf("  (soft) a surviving node (%s) is collecting the device after failover ✓\n", newOwner)
 				} else {
 					fmt.Println("  (soft) no survivor observed collecting within the window — known cluster data-plane gap")
@@ -3385,7 +3460,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				ipDeadline = runDeadline
 			}
 			fmt.Printf("  asserting leader node %d holds the cluster IP %s…\n", leader, vip)
-			if !waitNodeHasClusterIP(nodes[leader-1], vip, ipDeadline) {
+			if !waitNodeHasClusterIP(r.ctx, nodes[leader-1], vip, ipDeadline) {
 				actionOK = false
 				errs = append(errs, fmt.Sprintf("leader node %d did not bind the cluster IP %s", leader, vip))
 			} else {
@@ -3408,7 +3483,9 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				actionOK = false
 				errs = append(errs, fmt.Sprintf("docker stop %s failed: %v (disruption did not happen)", nodes[oldLeader-1], serr))
 			}
-			time.Sleep(settle)
+			if err := sleepCtx(r.ctx, settle); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 			newLeader, haveLeader := leaderExistsNow(nodes)
 			switch {
 			case !haveLeader:
@@ -3425,7 +3502,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				if migDeadline.After(runDeadline) {
 					migDeadline = runDeadline
 				}
-				if !waitNodeHasClusterIP(nodes[newLeader-1], vip, migDeadline) {
+				if !waitNodeHasClusterIP(r.ctx, nodes[newLeader-1], vip, migDeadline) {
 					actionOK = false
 					errs = append(errs, fmt.Sprintf("cluster IP %s did not migrate to the new leader (node %d)", vip, newLeader))
 				} else {
@@ -3442,7 +3519,9 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 				// and a follower never binds the VIP, so it must be absent. Check soon
 				// (before any later re-election could legitimately move leadership back).
 				if d := time.Until(runDeadline); d > 0 {
-					time.Sleep(min(10*time.Second, d))
+					if err := sleepCtx(r.ctx, min(10*time.Second, d)); err != nil {
+						return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+					}
 				}
 				if nodeHasClusterIP(nodes[oldLeader-1], vip) {
 					actionOK = false
@@ -3488,7 +3567,7 @@ func (r *Runner) runDirectorClusterCorrectness(tc *config.TestCase, subject conf
 		result.FailReason = strings.Join(errs, "; ")
 	}
 
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -3526,8 +3605,8 @@ type fleetStatBucket struct {
 	DroppedCount int64 `json:"dropped_count"`
 	BytesIn      int64 `json:"bytes_in"`
 	BytesOut     int64 `json:"bytes_out"`
-	ExecCount    int64 `json:"exec_count"`    // pipeline.* only: per-processor execution count
-	ExecTimeNs   int64 `json:"exec_time_ns"`  // pipeline.* only: per-processor execution time
+	ExecCount    int64 `json:"exec_count"`   // pipeline.* only: per-processor execution count
+	ExecTimeNs   int64 `json:"exec_time_ns"` // pipeline.* only: per-processor execution time
 }
 
 type fleetStatus struct {
@@ -3672,19 +3751,21 @@ func fleetSimDLProbe(simContainer string, params map[string]any) (map[string]int
 }
 
 // fleetWaitConnected polls until the director shows connected at the simulator.
-func fleetWaitConnected(simContainer, dirID string, deadline time.Time) bool {
+func fleetWaitConnected(ctx context.Context, simContainer, dirID string, deadline time.Time) bool {
 	for time.Now().Before(deadline) {
 		if st, err := fleetSimStatus(simContainer); err == nil && st.connected(dirID) {
 			return true
 		}
-		time.Sleep(3 * time.Second)
+		if sleepCtx(ctx, 3*time.Second) != nil {
+			break
+		}
 	}
 	return false
 }
 
 // fleetWaitCount polls until inbound[key].count >= min for the director, returning
 // the final count and whether the threshold was reached.
-func fleetWaitCount(simContainer, dirID, key string, min int, deadline time.Time) (int, bool) {
+func fleetWaitCount(ctx context.Context, simContainer, dirID, key string, min int, deadline time.Time) (int, bool) {
 	last := 0
 	for time.Now().Before(deadline) {
 		if st, err := fleetSimStatus(simContainer); err == nil {
@@ -3693,7 +3774,9 @@ func fleetWaitCount(simContainer, dirID, key string, min int, deadline time.Time
 				return last, true
 			}
 		}
-		time.Sleep(3 * time.Second)
+		if sleepCtx(ctx, 3*time.Second) != nil {
+			break
+		}
 	}
 	return last, false
 }
@@ -3703,7 +3786,7 @@ func fleetWaitCount(simContainer, dirID, key string, min int, deadline time.Time
 // reached. A capture's lifecycle markers ("Capture Started/Completed", tiny)
 // arrive immediately on session start, while real captured data only flushes on
 // the session's periodic (≈5s) ticker — so this must POLL, not sample once.
-func fleetWaitMaxLen(simContainer, dirID, key string, min int, deadline time.Time) (int, bool) {
+func fleetWaitMaxLen(ctx context.Context, simContainer, dirID, key string, min int, deadline time.Time) (int, bool) {
 	last := 0
 	for time.Now().Before(deadline) {
 		if st, err := fleetSimStatus(simContainer); err == nil {
@@ -3712,7 +3795,9 @@ func fleetWaitMaxLen(simContainer, dirID, key string, min int, deadline time.Tim
 				return last, true
 			}
 		}
-		time.Sleep(3 * time.Second)
+		if sleepCtx(ctx, 3*time.Second) != nil {
+			break
+		}
 	}
 	return last, false
 }
@@ -3722,7 +3807,7 @@ func fleetWaitMaxLen(simContainer, dirID, key string, min int, deadline time.Tim
 // deterministic for a fixed, replayed input, so an over-count is as much a
 // failure as an under-count. Returns nil on success, or an error describing the
 // outstanding mismatch when the deadline passes.
-func fleetWaitStats(simContainer, dirID string, expect map[string]map[string]int64, deadline time.Time) error {
+func fleetWaitStats(ctx context.Context, simContainer, dirID string, expect map[string]map[string]int64, deadline time.Time) error {
 	var lastErr error
 	for {
 		st, err := fleetSimStatus(simContainer)
@@ -3754,7 +3839,9 @@ func fleetWaitStats(simContainer, dirID string, expect map[string]map[string]int
 			}
 			return lastErr
 		}
-		time.Sleep(3 * time.Second)
+		if err := sleepCtx(ctx, 3*time.Second); err != nil {
+			return fmt.Errorf("interrupted: %w", err)
+		}
 	}
 }
 
@@ -3856,7 +3943,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -3910,7 +3997,9 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 				connected = true
 				break
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		if connected {
 			errs = append(errs, "director connected despite a bad fleet token (auth not enforced)")
@@ -3932,7 +4021,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			connDeadline = runDeadline
 		}
 		fmt.Printf("  waiting for the director to connect to the fleet simulator (up to %s)…\n", time.Until(connDeadline).Round(time.Second))
-		if !fleetWaitConnected(simContainer, dirID, connDeadline) {
+		if !fleetWaitConnected(r.ctx, simContainer, dirID, connDeadline) {
 			errs = append(errs, "director never connected to the fleet simulator")
 			passed = false
 			return r.saveFleetResult(tc, subject, configName, startTime, finalCount, passed, errs, simContainer, subjectContainer)
@@ -3980,7 +4069,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		if applyDeadline.After(runDeadline) {
 			applyDeadline = runDeadline
 		}
-		if _, ok := fleetWaitCount(simContainer, dirID, "rep.config", 1, applyDeadline); !ok {
+		if _, ok := fleetWaitCount(r.ctx, simContainer, dirID, "rep.config", 1, applyDeadline); !ok {
 			errs = append(errs, "director did not acknowledge the delivered config over the fleet link")
 			passed = false
 			return r.saveFleetResult(tc, subject, configName, startTime, finalCount, passed, errs, simContainer, subjectContainer)
@@ -4018,16 +4107,21 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			return r.saveFleetResult(tc, subject, configName, startTime, finalCount, passed, errs, simContainer, subjectContainer)
 		}
 		fmt.Println("  letting the agent stream before the restart…")
-		time.Sleep(25 * time.Second)
+		if err := sleepCtx(r.ctx, 25*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 		fmt.Println("  restarting the director (stop + start) mid-stream…")
 		if err := orch.StopServices(30*time.Second, "subject"); err != nil {
 			errs = append(errs, "failed to stop director for restart: "+err.Error())
 		}
-		time.Sleep(8 * time.Second) // director-down window — the agent should retry, not storm
+		// Director-down window — the agent should retry, not storm.
+		if err := sleepCtx(r.ctx, 8*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 		if err := orch.UpServices("subject"); err != nil {
 			errs = append(errs, "failed to restart director: "+err.Error())
 		}
-		if !fleetWaitConnected(simContainer, dirID, time.Now().Add(90*time.Second)) {
+		if !fleetWaitConnected(r.ctx, simContainer, dirID, time.Now().Add(90*time.Second)) {
 			errs = append(errs, "director did not reconnect to the fleet simulator after restart")
 		} else {
 			fmt.Println("  director reconnected after restart ✓")
@@ -4035,7 +4129,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 				if vmfBytes, e := os.ReadFile(filepath.Join(caseDir, "configs", fc.DeliverConfig)); e == nil {
 					b64 := base64.StdEncoding.EncodeToString(vmfBytes)
 					_, _ = fleetSimSend(simContainer, dirID, "config", map[string]any{"data_b64": b64})
-					_, _ = fleetWaitCount(simContainer, dirID, "rep.config", 2, time.Now().Add(60*time.Second))
+					_, _ = fleetWaitCount(r.ctx, simContainer, dirID, "rep.config", 2, time.Now().Add(60*time.Second))
 					fmt.Println("  re-delivered operational config after restart ✓")
 				}
 			}
@@ -4058,14 +4152,14 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			minHealth = 2
 		}
 		fmt.Printf("  waiting for >= %d health frames + connection_state…\n", minHealth)
-		hc, okH := fleetWaitCount(simContainer, dirID, "req.health", minHealth, scenarioDeadline())
+		hc, okH := fleetWaitCount(r.ctx, simContainer, dirID, "req.health", minHealth, scenarioDeadline())
 		finalCount = int64(hc)
 		if !okH {
 			errs = append(errs, fmt.Sprintf("director did not publish >= %d health frames (got %d)", minHealth, hc))
 		} else {
 			fmt.Printf("  health frames: %d ✓\n", hc)
 		}
-		if cs, okC := fleetWaitCount(simContainer, dirID, "req.connection_state", 1, scenarioDeadline()); !okC {
+		if cs, okC := fleetWaitCount(r.ctx, simContainer, dirID, "req.connection_state", 1, scenarioDeadline()); !okC {
 			errs = append(errs, "director did not publish connection_state")
 		} else {
 			fmt.Printf("  connection_state frames: %d ✓\n", cs)
@@ -4107,7 +4201,9 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 				ready = true
 				break
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		if !ready {
 			errs = append(errs, "director /dl endpoint never returned an HTTP response")
@@ -4178,7 +4274,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		if _, e := fleetSimSend(simContainer, dirID, "remote_check_ssh", params); e != nil {
 			errs = append(errs, "failed to send remote_check: "+e.Error())
 		}
-		if _, ok := fleetWaitCount(simContainer, dirID, "rep.remote_check", 1, scenarioDeadline()); !ok {
+		if _, ok := fleetWaitCount(r.ctx, simContainer, dirID, "rep.remote_check", 1, scenarioDeadline()); !ok {
 			errs = append(errs, "no remote_check reply from director")
 		} else {
 			last := ""
@@ -4216,7 +4312,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			if _, se := fleetSimSend(simContainer, dirID, "config", map[string]any{"data_b64": b64}); se != nil {
 				return fmt.Errorf("pushing %s: %w", filename, se)
 			}
-			if _, ok := fleetWaitCount(simContainer, dirID, "rep.config", wantReplies, scenarioDeadline()); !ok {
+			if _, ok := fleetWaitCount(r.ctx, simContainer, dirID, "rep.config", wantReplies, scenarioDeadline()); !ok {
 				return fmt.Errorf("no config reply from director after pushing %s", filename)
 			}
 			last := ""
@@ -4284,7 +4380,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			if _, se := fleetSimSend(simContainer, dirID, "config", map[string]any{"data_b64": b64}); se != nil {
 				return fmt.Errorf("pushing %s: %w", filename, se)
 			}
-			if _, ok := fleetWaitCount(simContainer, dirID, "rep.config", before+1, scenarioDeadline()); !ok {
+			if _, ok := fleetWaitCount(r.ctx, simContainer, dirID, "rep.config", before+1, scenarioDeadline()); !ok {
 				return fmt.Errorf("no config reply from director after pushing %s", filename)
 			}
 			last := ""
@@ -4317,7 +4413,9 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		}
 		fmt.Printf("  confirming delivery is SUPPRESSED for %s (receiver must stay ~0)…\n", baseline)
 		if baseline > 0 {
-			time.Sleep(baseline)
+			if err := sleepCtx(r.ctx, baseline); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		rmBase, qErr := r.queryReceiverMetrics(metricsPort, 10*time.Second)
 		if qErr != nil {
@@ -4362,7 +4460,9 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 					break
 				}
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		delivered := finalCount - baseCount
 		if delivered < minRecv {
@@ -4416,7 +4516,9 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		// Phase 0: confirm delivery is BLOCKED under config A (dead target), so
 		// post-change delivery is attributable to the pushed change.
 		if d := min(15*time.Second, time.Until(runDeadline)); d > 0 {
-			time.Sleep(d)
+			if err := sleepCtx(r.ctx, d); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		rmB, _ := r.queryReceiverMetrics(metricsPort, 10*time.Second)
 		fmt.Printf("  before config change: received=%s (expected ~0; target A is a dead endpoint)\n", formatCount(rmB.LinesReceived))
@@ -4442,7 +4544,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			errs = append(errs, "failed to push changed config: "+se.Error())
 			break
 		}
-		if _, ok := fleetWaitCount(simContainer, dirID, "rep.config", before+1, scenarioDeadline()); !ok {
+		if _, ok := fleetWaitCount(r.ctx, simContainer, dirID, "rep.config", before+1, scenarioDeadline()); !ok {
 			errs = append(errs, "no config reply from director after the change push")
 		}
 
@@ -4471,7 +4573,9 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 					break
 				}
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		if finalCount < target {
 			errs = append(errs, fmt.Sprintf(
@@ -4503,7 +4607,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		// Generator traffic (if the case ships one) flows through the director so
 		// records get captured.
 		key := "req." + cmd + "_reply"
-		if c, ok := fleetWaitCount(simContainer, dirID, key, 1, scenarioDeadline()); !ok {
+		if c, ok := fleetWaitCount(r.ctx, simContainer, dirID, key, 1, scenarioDeadline()); !ok {
 			errs = append(errs, fmt.Sprintf("no %s capture data streamed back (got %d frames)", cmd, c))
 		} else {
 			finalCount = int64(c)
@@ -4516,7 +4620,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			// asserts at least one frame carried real captured data. Used by the
 			// where:raw mis-frame case.
 			if fc.LiveMinFrameBytes > 0 {
-				if maxLen, ok := fleetWaitMaxLen(simContainer, dirID, key, fc.LiveMinFrameBytes, scenarioDeadline()); !ok {
+				if maxLen, ok := fleetWaitMaxLen(r.ctx, simContainer, dirID, key, fc.LiveMinFrameBytes, scenarioDeadline()); !ok {
 					errs = append(errs, fmt.Sprintf("%s returned only small frames (max %d bytes < live_min_frame_bytes %d) — capture saw lifecycle markers, no real data", cmd, maxLen, fc.LiveMinFrameBytes))
 				} else {
 					fmt.Printf("  %s largest frame %d bytes (>= %d) — real data captured ✓\n", cmd, maxLen, fc.LiveMinFrameBytes)
@@ -4528,10 +4632,10 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		// The director forwards stats/metrics partitions as traffic flows. The case
 		// ships a generator → director pipeline; assert metrics frames arrive.
 		fmt.Println("  waiting for stats/metrics frames…")
-		c1, ok1 := fleetWaitCount(simContainer, dirID, "req.metricsvmf", 1, scenarioDeadline())
+		c1, ok1 := fleetWaitCount(r.ctx, simContainer, dirID, "req.metricsvmf", 1, scenarioDeadline())
 		c2 := 0
 		if !ok1 {
-			c2, _ = fleetWaitCount(simContainer, dirID, "req.metrics", 1, scenarioDeadline())
+			c2, _ = fleetWaitCount(r.ctx, simContainer, dirID, "req.metrics", 1, scenarioDeadline())
 		}
 		finalCount = int64(c1 + c2)
 		if c1 < 1 && c2 < 1 {
@@ -4545,7 +4649,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		// arrived. This is what proves the director's pipeline-error / dropped
 		// counters increment correctly and surface on the stats path.
 		if len(fc.ExpectStats) > 0 {
-			if serr := fleetWaitStats(simContainer, dirID, fc.ExpectStats, scenarioDeadline()); serr != nil {
+			if serr := fleetWaitStats(r.ctx, simContainer, dirID, fc.ExpectStats, scenarioDeadline()); serr != nil {
 				errs = append(errs, "decoded stats mismatch: "+serr.Error())
 			} else {
 				fmt.Println("  decoded stats match expectations ✓")
@@ -4574,13 +4678,15 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 				reconnected = true
 				break
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		if !reconnected {
 			errs = append(errs, "director did not reconnect after the simulator restarted")
 		} else {
 			// And resume publishing health on the fresh connection.
-			if hc, ok := fleetWaitCount(simContainer, dirID, "req.health", 1, scenarioDeadline()); ok {
+			if hc, ok := fleetWaitCount(r.ctx, simContainer, dirID, "req.health", 1, scenarioDeadline()); ok {
 				finalCount = int64(hc)
 				fmt.Printf("  director reconnected and resumed health (%d) ✓\n", hc)
 			} else {
@@ -4592,7 +4698,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		// A self-managed director still publishes health but must IGNORE platform
 		// commands (ListenRequests is not started). Send a remote_check and assert
 		// NO reply, while health keeps flowing.
-		if hc, ok := fleetWaitCount(simContainer, dirID, "req.health", 1, scenarioDeadline()); ok {
+		if hc, ok := fleetWaitCount(r.ctx, simContainer, dirID, "req.health", 1, scenarioDeadline()); ok {
 			fmt.Printf("  self-managed director still publishes health (%d) ✓\n", hc)
 			finalCount = int64(hc)
 		} else {
@@ -4609,7 +4715,9 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			if d > settle {
 				d = settle
 			}
-			time.Sleep(d)
+			if err := sleepCtx(r.ctx, d); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		if st, e := fleetSimStatus(simContainer); e != nil {
 			errs = append(errs, "could not read simulator status to confirm the command was ignored: "+e.Error())
@@ -4628,7 +4736,7 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 		// the approval to the agent. Verdict: the simulator observes the forwarded
 		// enrollment (rep.check_enrollment), proving the agent→director→platform path.
 		fmt.Println("  waiting for the agent's enrollment to be forwarded to the platform…")
-		if c, ok := fleetWaitCount(simContainer, dirID, "rep.check_enrollment", 1, scenarioDeadline()); !ok {
+		if c, ok := fleetWaitCount(r.ctx, simContainer, dirID, "rep.check_enrollment", 1, scenarioDeadline()); !ok {
 			errs = append(errs, fmt.Sprintf("director did not forward agent enrollment to the platform (got %d frames)", c))
 		} else {
 			finalCount = int64(c)
@@ -4677,7 +4785,9 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 					break
 				}
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		// One final read so the content verdict reflects the settled receiver state.
 		if cur, qe := r.queryReceiverMetrics(metricsPort, 10*time.Second); qe == nil {
@@ -4769,12 +4879,14 @@ func (r *Runner) runFleetAutomationCorrectness(tc *config.TestCase, subject conf
 			errs = append(errs, "failed to push changed config: "+se.Error())
 			break
 		}
-		if _, ok := fleetWaitCount(simContainer, dirID, "rep.config", before+1, scenarioDeadline()); !ok {
+		if _, ok := fleetWaitCount(r.ctx, simContainer, dirID, "rep.config", before+1, scenarioDeadline()); !ok {
 			errs = append(errs, "no config reply from director after the change push")
 		}
 		// Let the director reload + the target write under config B before verifying.
 		if d := min(settle, time.Until(runDeadline)); d > 0 {
-			time.Sleep(d)
+			if err := sleepCtx(r.ctx, d); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		fmt.Println("  running the DuckDB verifier against the post-update target objects…")
 		m, verr := r.runVerifier(orch, tc, tmpDir, runDeadline)
@@ -4834,7 +4946,7 @@ func (r *Runner) saveFleetResult(tc *config.TestCase, subject config.Subject, co
 	if !passed {
 		result.FailReason = strings.Join(errs, "; ")
 	}
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -4889,7 +5001,9 @@ func (r *Runner) ccfWaitLines(metricsPort int, target int64, deadline time.Time)
 				return last, true
 			}
 		}
-		time.Sleep(3 * time.Second)
+		if sleepCtx(r.ctx, 3*time.Second) != nil {
+			break
+		}
 	}
 	return last, false
 }
@@ -4913,7 +5027,9 @@ func (r *Runner) ccfWaitStable(metricsPort int, deadline time.Time) int64 {
 				last = rm.LinesReceived
 			}
 		}
-		time.Sleep(3 * time.Second)
+		if sleepCtx(r.ctx, 3*time.Second) != nil {
+			break
+		}
 	}
 	if last < 0 {
 		return 0
@@ -4991,7 +5107,7 @@ func (r *Runner) runCCFCorrectness(tc *config.TestCase, subject config.Subject)
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -5046,7 +5162,9 @@ func (r *Runner) runCCFCorrectness(tc *config.TestCase, subject config.Subject)
 			apiReady = true
 			break
 		}
-		time.Sleep(2 * time.Second)
+		if err := sleepCtx(r.ctx, 2*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 	}
 	if !apiReady {
 		errs = append(errs, "mock CCF API never became reachable")
@@ -5095,7 +5213,9 @@ func (r *Runner) runCCFCorrectness(tc *config.TestCase, subject config.Subject)
 					break
 				}
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		if !rejected {
 			errs = append(errs, "mock recorded no auth failures within the window — the director never polled")
@@ -5202,7 +5322,7 @@ func (r *Runner) saveCCFResult(tc *config.TestCase, subject config.Subject, conf
 	if !passed {
 		result.FailReason = strings.Join(errs, "; ")
 	}
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -5299,7 +5419,7 @@ func (r *Runner) runHTTPSourceCorrectness(tc *config.TestCase, subject config.Su
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -5353,7 +5473,9 @@ func (r *Runner) runHTTPSourceCorrectness(tc *config.TestCase, subject config.Su
 			if s, _, _, e := httpSenderStats(senderContainer, ctrlPort); e == nil && s >= expected {
 				break
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		got := r.ccfWaitStable(metricsPort, time.Now().Add(12*time.Second))
 		finalCount = got
@@ -5385,7 +5507,9 @@ func (r *Runner) runHTTPSourceCorrectness(tc *config.TestCase, subject config.Su
 			if s, _, _, e := httpSenderStats(senderContainer, ctrlPort); e == nil && s >= expected {
 				break
 			}
-			time.Sleep(3 * time.Second)
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		got := r.ccfWaitStable(metricsPort, time.Now().Add(12*time.Second))
 		finalCount = got
@@ -5452,7 +5576,7 @@ func (r *Runner) saveHTTPSourceResult(tc *config.TestCase, subject config.Subjec
 	if !passed {
 		result.FailReason = strings.Join(errs, "; ")
 	}
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -5524,7 +5648,7 @@ func (r *Runner) runClickHouseTargetCorrectness(tc *config.TestCase, subject con
 		GeneratorImage: r.opts.GeneratorImage, ReceiverImage: r.opts.ReceiverImage, CollectorImage: r.opts.CollectorImage,
 		ReceiverHostPort: r.opts.ReceiverHostPort, ExtraSubjectEnv: extraEnv, CPULimit: r.opts.CPULimit, MemLimit: r.opts.MemLimit,
 	}
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -5569,7 +5693,9 @@ func (r *Runner) runClickHouseTargetCorrectness(tc *config.TestCase, subject con
 			ready = true
 			break
 		}
-		time.Sleep(2 * time.Second)
+		if err := sleepCtx(r.ctx, 2*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 	}
 	if !ready {
 		errs = append(errs, "clickhouse-server never became reachable")
@@ -5649,10 +5775,14 @@ func (r *Runner) runClickHouseTargetCorrectness(tc *config.TestCase, subject con
 				}
 			}
 		}
-		time.Sleep(3 * time.Second)
+		if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 	}
 	// Let any late/duplicate inserts settle, then re-read for an exact check.
-	time.Sleep(5 * time.Second)
+	if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+		return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+	}
 	if out, e := chExec(chContainer, "SELECT count() FROM "+fqTable); e == nil {
 		if n, perr := strconv.ParseInt(strings.TrimSpace(out), 10, 64); perr == nil {
 			finalCount = n
@@ -5691,7 +5821,7 @@ func (r *Runner) saveClickHouseTargetResult(tc *config.TestCase, subject config.
 	if !passed {
 		result.FailReason = strings.Join(errs, "; ")
 	}
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -5786,9 +5916,13 @@ func (r *Runner) runMQTTTargetCorrectness(tc *config.TestCase, subject config.Su
 		if finalCount >= expected {
 			break
 		}
-		time.Sleep(3 * time.Second)
+		if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
+	}
+	if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+		return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
 	}
-	time.Sleep(5 * time.Second)
 	finalCount = mqttSubLineCount(subContainer, recvFile)
 	if finalCount < expected {
 		errs = append(errs, fmt.Sprintf("under-delivery: %d of %d messages received by the subscriber", finalCount, expected))
@@ -5870,7 +6004,9 @@ func (r *Runner) runRedisSourceCorrectness(tc *config.TestCase, subject config.S
 			settleDeadline = runDeadline
 		}
 		for time.Now().Before(settleDeadline) {
-			time.Sleep(5 * time.Second)
+			if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		got := r.ccfWaitStable(metricsPort, time.Now().Add(10*time.Second))
 		if got > int64(rc.ExpectMax) {
@@ -5974,7 +6110,9 @@ func (r *Runner) runEndpointSourceCorrectness(tc *config.TestCase, subject confi
 		}
 		fmt.Printf("  reject case: waiting %s then asserting ≤ %d records reached the receiver…\n", time.Until(settleDeadline).Round(time.Second), es.ExpectMax)
 		for time.Now().Before(settleDeadline) {
-			time.Sleep(5 * time.Second)
+			if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+				return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+			}
 		}
 		got := r.ccfWaitStable(metricsPort, time.Now().Add(10*time.Second))
 		if got > int64(es.ExpectMax) {
@@ -6038,7 +6176,7 @@ func (r *Runner) setupAuxRun(tc *config.TestCase, subject config.Subject, config
 		GeneratorImage: r.opts.GeneratorImage, ReceiverImage: r.opts.ReceiverImage, CollectorImage: r.opts.CollectorImage,
 		ReceiverHostPort: r.opts.ReceiverHostPort, ExtraSubjectEnv: extraEnv, CPULimit: r.opts.CPULimit, MemLimit: r.opts.MemLimit,
 	}
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return nil, caseDir, tmpDir, fmt.Errorf("compose setup: %w", err)
 	}
@@ -6066,7 +6204,7 @@ func (r *Runner) saveAuxResult(tc *config.TestCase, subject config.Subject, conf
 	if !passed {
 		result.FailReason = strings.Join(errs, "; ")
 	}
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -6171,7 +6309,7 @@ func (r *Runner) runHTTPVaultCertRotation(tc *config.TestCase, subject config.Su
 		GeneratorImage: r.opts.GeneratorImage, ReceiverImage: r.opts.ReceiverImage, CollectorImage: r.opts.CollectorImage,
 		ReceiverHostPort: r.opts.ReceiverHostPort, ExtraSubjectEnv: extraEnv, CPULimit: r.opts.CPULimit, MemLimit: r.opts.MemLimit,
 	}
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -6230,7 +6368,9 @@ func (r *Runner) runHTTPVaultCertRotation(tc *config.TestCase, subject config.Su
 				break
 			}
 		}
-		time.Sleep(3 * time.Second)
+		if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 	}
 	if fp1 == "" {
 		errs = append(errs, "never observed initial HTTPS delivery + server cert (vault-sourced cert may have failed to load)")
@@ -6270,7 +6410,9 @@ func (r *Runner) runHTTPVaultCertRotation(tc *config.TestCase, subject config.Su
 			rotated = true
 			break
 		}
-		time.Sleep(3 * time.Second)
+		if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 	}
 	if rotated {
 		fmt.Printf("  director hot-reloaded the rotated cert (fp %s… → %s…) ✓\n", fpShort(fp1), fpShort(fp2))
@@ -6368,7 +6510,7 @@ func (r *Runner) runKafkaOffsetCommitRestart(tc *config.TestCase, subject config
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	orch, err := orchestrator.NewComposeRunner(runCfg)
+	orch, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -6409,6 +6551,10 @@ func (r *Runner) runKafkaOffsetCommitRestart(tc *config.TestCase, subject config
 	warmup := tc.WarmupOrDefault(30 * time.Second)
 	genTimeout := min(duration+warmup+2*time.Minute, r.opts.Timeout)
 	if err := orch.WaitForGeneratorExit(genTimeout); err != nil {
+		if errors.Is(err, context.Canceled) {
+			// A user interrupt is not a tolerable generator hiccup — abort.
+			return results.RunResult{}, err
+		}
 		fmt.Printf("  (generator wait: %v)\n", err)
 	}
 	genStats := r.parseGeneratorStats(orch.GeneratorStdout())
@@ -6432,7 +6578,9 @@ func (r *Runner) runKafkaOffsetCommitRestart(tc *config.TestCase, subject config
 				break
 			}
 		}
-		time.Sleep(2 * time.Second)
+		if err := sleepCtx(r.ctx, 2*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 	}
 	if !delivered {
 		return results.RunResult{}, fmt.Errorf("receiver never reached full delivery (%s) before timeout", formatCount(sent))
@@ -6441,12 +6589,16 @@ func (r *Runner) runKafkaOffsetCommitRestart(tc *config.TestCase, subject config
 	// Settle so the delivery-bound offset commits land at the broker; the
 	// graceful stop below additionally drains pending commits on shutdown.
 	fmt.Println("  full delivery reached — settling 5s, then graceful restart…")
-	time.Sleep(5 * time.Second)
+	if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+		return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+	}
 
 	if err := orch.StopServices(30*time.Second, "subject"); err != nil {
 		return results.RunResult{}, fmt.Errorf("stopping subject: %w", err)
 	}
-	time.Sleep(3 * time.Second)
+	if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+		return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+	}
 	fmt.Println("  restarting subject (must resume from committed offsets)…")
 	if err := orch.UpServices("subject"); err != nil {
 		return results.RunResult{}, fmt.Errorf("restarting subject: %w", err)
@@ -6460,7 +6612,9 @@ func (r *Runner) runKafkaOffsetCommitRestart(tc *config.TestCase, subject config
 	stableRounds := 0
 	observeDeadline := time.Now().Add(90 * time.Second)
 	for time.Now().Before(observeDeadline) {
-		time.Sleep(5 * time.Second)
+		if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 		rm, qerr := r.queryReceiverMetrics(metricsPort, 10*time.Second)
 		if qerr != nil {
 			continue
@@ -6546,7 +6700,7 @@ func (r *Runner) runKafkaOffsetCommitRestart(tc *config.TestCase, subject config
 		result.FailReason = strings.Join(errors, "; ")
 	}
 
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -6635,7 +6789,7 @@ func (r *Runner) runPersistenceFileRestartCorrectness(tc *config.TestCase, subje
 		MemLimit:         r.opts.MemLimit,
 	}
 
-	cr, err := orchestrator.NewComposeRunner(runCfg)
+	cr, err := orchestrator.NewComposeRunner(r.ctx, runCfg)
 	if err != nil {
 		return results.RunResult{}, fmt.Errorf("compose setup: %w", err)
 	}
@@ -6672,7 +6826,9 @@ func (r *Runner) runPersistenceFileRestartCorrectness(tc *config.TestCase, subje
 	warmup := tc.WarmupOrDefault(10 * time.Second)
 	stopAfter := warmup + rotateAt + 5*time.Second
 	fmt.Printf("  phase 2: waiting %s (rotation fires at warmup+%s)…\n", stopAfter, rotateAt)
-	time.Sleep(stopAfter)
+	if err := sleepCtx(r.ctx, stopAfter); err != nil {
+		return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+	}
 
 	// PHASE 3: SIGTERM subject. Its file-tail state must be flushed to disk
 	// (persist file, position file, sincedb, …) so the restart can resume.
@@ -6713,7 +6869,9 @@ func (r *Runner) runPersistenceFileRestartCorrectness(tc *config.TestCase, subje
 	drainDeadline := time.Now().Add(drainTimeout)
 	var recvMetrics ReceiverMetrics
 	for time.Now().Before(drainDeadline) {
-		time.Sleep(5 * time.Second)
+		if err := sleepCtx(r.ctx, 5*time.Second); err != nil {
+			return results.RunResult{}, fmt.Errorf("interrupted: %w", err)
+		}
 		rm, err := r.queryReceiverMetrics(metricsPort, 10*time.Second)
 		if err != nil {
 			continue
@@ -6782,7 +6940,7 @@ func (r *Runner) runPersistenceFileRestartCorrectness(tc *config.TestCase, subje
 		result.FailReason = strings.Join(perrs, "; ")
 	}
 
-	dir, err := r.store.Save(result, "")
+	dir, err := r.saveResult(result, "")
 	if err != nil {
 		return result, fmt.Errorf("saving results: %w", err)
 	}
@@ -6815,24 +6973,55 @@ func (r *Runner) runPersistenceFileRestartCorrectness(tc *config.TestCase, subje
 	return result, nil
 }
 
+// saveResult persists a run result unless the run was interrupted — a verdict
+// computed after cancellation reflects a half-finished run (waits bail out
+// early on cancel) and must never land in the store.
+func (r *Runner) saveResult(result results.RunResult, metricsCSVSrc string) (string, error) {
+	if err := r.ctx.Err(); err != nil {
+		return "", fmt.Errorf("interrupted: %w", err)
+	}
+	return r.store.Save(result, metricsCSVSrc)
+}
+
+// sleepCtx sleeps for d or until ctx is cancelled; returns ctx.Err() on cancel.
+func sleepCtx(ctx context.Context, d time.Duration) error {
+	if d <= 0 {
+		return ctx.Err()
+	}
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-t.C:
+		return nil
+	}
+}
+
 func (r *Runner) queryReceiverMetrics(port int, timeout time.Duration) (ReceiverMetrics, error) {
 	url := fmt.Sprintf("http://localhost:%d/metrics", port)
 	deadline := time.Now().Add(timeout)
 	for time.Now().Before(deadline) {
 		resp, err := http.Get(url) //nolint:noctx
 		if err != nil {
-			time.Sleep(time.Second)
+			if err := sleepCtx(r.ctx, time.Second); err != nil {
+				return ReceiverMetrics{}, fmt.Errorf("interrupted: %w", err)
+			}
 			continue
 		}
 		body, err := io.ReadAll(resp.Body)
 		resp.Body.Close()
 		if err != nil {
-			time.Sleep(time.Second)
+			if err := sleepCtx(r.ctx, time.Second); err != nil {
+				return ReceiverMetrics{}, fmt.Errorf("interrupted: %w", err)
+			}
 			continue
 		}
 		var m ReceiverMetrics
 		if err := json.Unmarshal(body, &m); err != nil {
-			time.Sleep(time.Second)
+			if err := sleepCtx(r.ctx, time.Second); err != nil {
+				return ReceiverMetrics{}, fmt.Errorf("interrupted: %w", err)
+			}
 			continue
 		}
 		return m, nil
@@ -7062,7 +7251,9 @@ func (r *Runner) runSyslogVaultCertRotation(tc *config.TestCase, subject config.
 			stallRounds := 0
 			rejected := false
 			for time.Now().Before(phase1Deadline) {
-				time.Sleep(2 * time.Second)
+				if err := sleepCtx(r.ctx, 2*time.Second); err != nil {
+					return fmt.Errorf("interrupted: %w", err)
+				}
 				rm, qerr := r.queryReceiverMetrics(metricsPort, 5*time.Second)
 				if qerr != nil {
 					continue
@@ -7104,7 +7295,9 @@ func (r *Runner) runSyslogVaultCertRotation(tc *config.TestCase, subject config.
 			// Recovery is confirmed by the final loss verdict: if the director
 			// does not recover, lines_out < expected and loss_percent > ceiling.
 			fmt.Println("  phase 2: trusted cert restored in Vault — waiting 25s for director to detect and recover…")
-			time.Sleep(25 * time.Second)
+			if err := sleepCtx(r.ctx, 25*time.Second); err != nil {
+				return fmt.Errorf("interrupted: %w", err)
+			}
 			fmt.Println("  phase 2: director should have restarted with the restored cert — delivery resuming")
 			return nil
 		},
diff --git a/internal/runner/sleep_test.go b/internal/runner/sleep_test.go
new file mode 100644
index 0000000..a0a93dc
--- /dev/null
+++ b/internal/runner/sleep_test.go
@@ -0,0 +1,54 @@
+package runner
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestSleepCtxCancelled(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	start := time.Now()
+	err := sleepCtx(ctx, 10*time.Second)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("want context.Canceled, got %v", err)
+	}
+	if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
+		t.Fatalf("pre-cancelled sleep took %s, want immediate return", elapsed)
+	}
+}
+
+func TestSleepCtxCompletes(t *testing.T) {
+	if err := sleepCtx(context.Background(), 10*time.Millisecond); err != nil {
+		t.Fatalf("want nil, got %v", err)
+	}
+}
+
+func TestSleepCtxCancelMidSleep(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		time.Sleep(20 * time.Millisecond)
+		cancel()
+	}()
+	start := time.Now()
+	err := sleepCtx(ctx, 10*time.Second)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("want context.Canceled, got %v", err)
+	}
+	if elapsed := time.Since(start); elapsed > time.Second {
+		t.Fatalf("mid-sleep cancel took %s, want prompt return", elapsed)
+	}
+}
+
+func TestSleepCtxNonPositive(t *testing.T) {
+	if err := sleepCtx(context.Background(), 0); err != nil {
+		t.Fatalf("want nil for d=0 on live ctx, got %v", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if err := sleepCtx(ctx, -time.Second); !errors.Is(err, context.Canceled) {
+		t.Fatalf("want context.Canceled for negative d on cancelled ctx, got %v", err)
+	}
+}