diff --git a/06-chat-messaging-retention/kei-nan/.env.example b/06-chat-messaging-retention/kei-nan/.env.example
new file mode 100644
index 0000000..a25cc48
--- /dev/null
+++ b/06-chat-messaging-retention/kei-nan/.env.example
@@ -0,0 +1,8 @@
+# Copy this file to .env (which is gitignored) and set your cluster URL:
+#
+#   cp .env.example .env      # then edit .env
+#
+# chatstress reads CHATSTRESS_URL automatically once you `source .env`.
+#   rediss:// = TLS on (typical for cloud); redis:// = plaintext.
+#   format: scheme://<user>:<password>@<host>:<port>
+CHATSTRESS_URL=rediss://default:YOUR_PASSWORD@your-host.cloud.example.com:12345
diff --git a/06-chat-messaging-retention/kei-nan/.gitignore b/06-chat-messaging-retention/kei-nan/.gitignore
index b999454..2f75170 100644
--- a/06-chat-messaging-retention/kei-nan/.gitignore
+++ b/06-chat-messaging-retention/kei-nan/.gitignore
@@ -1,5 +1,16 @@
 # Build output and run artifacts (also covered by the repo-root .gitignore).
+# NOTE: anchor the binary name with a leading slash — a bare `chatstress`
+# pattern also matches the cmd/chatstress/ source directory and silently
+# excludes main.go from git.
 /bin/
 /out/
-chatstress
+/chatstress
 flexdata/
+
+# Secrets — cloud connection URLs live here and must NEVER be committed.
+# (.env.example is the committed placeholder.)
+.env
+.env.*
+!.env.example
+config/local.yaml
+config/*.local.yaml
diff --git a/06-chat-messaging-retention/kei-nan/README.md b/06-chat-messaging-retention/kei-nan/README.md
index 8706b1c..a4c9169 100644
--- a/06-chat-messaging-retention/kei-nan/README.md
+++ b/06-chat-messaging-retention/kei-nan/README.md
@@ -91,25 +91,43 @@ Open the dashboard at **http://localhost:8080**. It stays up after the run
 completes so you can inspect the final state (Ctrl-C to exit). Confirm the disk
 path is live with `redis-cli INFO search | grep search_disk_usage`.
 
-Flags: `-addr`, `-config`, `-duration`, `-web-addr`, `-flush`, `-no-web`, `-seed`,
-`-password`, `-username`, `-tls`.
+Flags: `-url`, `-addr`, `-config`, `-duration`, `-web-addr`, `-flush`, `-no-web`,
+`-seed`, `-password`, `-username`, `-tls`.
 
 ### Connecting to a cloud endpoint
 
 The connection is a **launch-time** parameter (the harness connects once, creates
 the index, then runs) — there's no connection box in the dashboard, which only
-controls the workload. Point it at a cloud Flex/BigRedis endpoint with flags or
-config:
+controls the workload.
+
+**Recommended (keeps the password out of shell history and git): a full URL via
+`.env`.** Copy `.env.example` to `.env` (gitignored) and set your cluster's
+connection string:
+
+```bash
+cp .env.example .env      # then edit .env:
+#   CHATSTRESS_URL=rediss://default:<password>@my-endpoint.cloud.redislabs.com:12345
+set -a; source .env; set +a
+./bin/chatstress run -config config/bugbash.yaml
+```
+
+`rediss://` enables TLS; `redis://` is plaintext. The harness reads
+`CHATSTRESS_URL` automatically; you can also pass the URL explicitly with `-url`.
+Only `host:port` is ever printed/logged, never the credentials.
+
+Alternatively, use discrete flags or config keys:
 
 ```bash
 ./bin/chatstress run -config config/bugbash.yaml \
   -addr my-endpoint.cloud.redislabs.com:6379 -tls -username default -password "$REDIS_PASSWORD"
 ```
 
-Config equivalents: `addr`, `tls: true`, `tls_skip_verify` (self-signed certs),
-`username`, `password`. Redis Enterprise/Cloud presents a single proxy endpoint,
-so the standard client works; a raw OSS-cluster endpoint (with `MOVED`) is not
-supported. TLS uses the system root CAs unless `tls_skip_verify` is set.
+Config equivalents: `url`, or `addr` + `tls: true` / `tls_skip_verify`
+(self-signed certs) / `username` / `password`. Precedence: `-url` flag >
+`$CHATSTRESS_URL` > config `url` > `addr`. Redis Enterprise/Cloud presents a
+single proxy endpoint, so the standard client works; a raw OSS-cluster endpoint
+(with `MOVED`) is not supported. TLS uses the system root CAs unless
+`tls_skip_verify` is set.
 
 ## Workload / query profiles
 
diff --git a/06-chat-messaging-retention/kei-nan/cmd/chatstress/main.go b/06-chat-messaging-retention/kei-nan/cmd/chatstress/main.go
new file mode 100644
index 0000000..4530acd
--- /dev/null
+++ b/06-chat-messaging-retention/kei-nan/cmd/chatstress/main.go
@@ -0,0 +1,201 @@
+// Command chatstress drives the use-case-6 (chat / messaging with retention)
+// stress workload against a Redis Search on Disk (Flex) index, exposes a live
+// web dashboard, and verifies the "no stale hits" correctness property.
+//
+// Usage:
+//
+//	chatstress run          -config config/smoke.yaml [-addr host:port] [-flush]
+//	chatstress reload-check  -config config/smoke.yaml [-addr host:port]
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"chatstress/internal/config"
+	"chatstress/internal/control"
+	"chatstress/internal/metrics"
+	"chatstress/internal/oracle"
+	"chatstress/internal/redisx"
+	"chatstress/internal/report"
+	"chatstress/internal/web"
+	"chatstress/internal/workload"
+)
+
+func main() {
+	mode, args := parseMode(os.Args[1:])
+
+	fs := flag.NewFlagSet("chatstress "+mode, flag.ExitOnError)
+	configPath := fs.String("config", "", "path to YAML config")
+	url := fs.String("url", "", "full Redis URL (redis://|rediss://); overrides -addr/-password/-tls. Prefer $CHATSTRESS_URL / a .env file to keep the password out of shell history")
+	addr := fs.String("addr", "", "override Redis address (host:port)")
+	password := fs.String("password", "", "override Redis password (cloud/auth)")
+	username := fs.String("username", "", "override Redis ACL username (cloud)")
+	useTLS := fs.Bool("tls", false, "connect over TLS (cloud endpoints)")
+	webAddr := fs.String("web-addr", "", "override web dashboard address")
+	durStr := fs.String("duration", "", "override run duration (e.g. 90s, 30m)")
+	seed := fs.Int64("seed", 0, "override RNG seed (0 = use config)")
+	flush := fs.Bool("flush", false, "FLUSHALL before starting (clean slate)")
+	noWeb := fs.Bool("no-web", false, "disable the web dashboard")
+	_ = fs.Parse(args)
+
+	cfg, err := config.Load(*configPath)
+	if err != nil {
+		die("config: %v", err)
+	}
+	// Connection URL precedence: -url flag > $CHATSTRESS_URL > config `url`. A URL
+	// (when present) supersedes addr/username/password/tls in redisx.New.
+	if *url != "" {
+		cfg.URL = *url
+	} else if env := os.Getenv("CHATSTRESS_URL"); env != "" {
+		cfg.URL = env
+	}
+	if cfg.URL != "" {
+		// Redacted host:port for display/logs only — never print the URL itself.
+		if a := redisx.AddrFromURL(cfg.URL); a != "" {
+			cfg.Addr = a
+		}
+	}
+	if *addr != "" {
+		cfg.Addr = *addr
+	}
+	if *password != "" {
+		cfg.Password = *password
+	}
+	if *username != "" {
+		cfg.Username = *username
+	}
+	if *useTLS {
+		cfg.TLS = true
+	}
+	if *webAddr != "" {
+		cfg.Web.Addr = *webAddr
+	}
+	if *seed != 0 {
+		cfg.Seed = *seed
+	}
+	if *noWeb {
+		cfg.Web.Enabled = false
+	}
+	if *durStr != "" {
+		d, perr := time.ParseDuration(*durStr)
+		if perr != nil {
+			die("bad -duration: %v", perr)
+		}
+		cfg.Duration = config.Duration(d)
+	}
+
+	// Size the pool for the max live query concurrency so scaling threads up from
+	// the UI yields real concurrent connections rather than pool contention.
+	pool := cfg.Ingest.Workers + cfg.Query.MaxWorkers + 32
+	cli, err := redisx.New(redisx.Options{
+		URL: cfg.URL, Addr: cfg.Addr, Username: cfg.Username, Password: cfg.Password,
+		TLS: cfg.TLS, TLSSkipVerify: cfg.TLSSkipVerify, Index: cfg.Index, PoolSize: pool,
+	})
+	if err != nil {
+		die("connect: %v", err)
+	}
+	defer cli.Close()
+
+	// Signal-cancellable root context.
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	switch mode {
+	case "run":
+		runMode(ctx, cfg, cli, *flush)
+	case "reload-check":
+		if err := reloadCheck(ctx, cfg, cli); err != nil {
+			die("reload-check FAILED: %v", err)
+		}
+	default:
+		die("unknown mode %q (want: run | reload-check)", mode)
+	}
+}
+
+func runMode(ctx context.Context, cfg *config.Config, cli *redisx.Client, flush bool) {
+	if flush {
+		fmt.Println("FLUSHALL (clean slate)…")
+		if err := cli.FlushAll(context.Background()); err != nil {
+			die("flushall: %v", err)
+		}
+	}
+	// Create the index BEFORE loading data (SKIPINITIALSCAN => no back-indexing).
+	if err := cli.CreateIndex(context.Background(), cfg.Prefix); err != nil {
+		die("create index: %v", err)
+	}
+
+	met := metrics.New()
+	orc := oracle.New(256, cfg.Oracle.MaxTracked, cfg.Oracle.SampleRate, cfg.Oracle.GraceMs, cfg.Oracle.SkewMs)
+	ctrl := control.New(cfg)
+	rep := report.New(met, orc)
+
+	if cfg.Web.Enabled {
+		srv := web.New(cfg, cli, met, orc, ctrl)
+		go func() {
+			if err := srv.Start(ctx); err != nil {
+				fmt.Fprintf(os.Stderr, "web server: %v\n", err)
+			}
+		}()
+		fmt.Printf("dashboard: http://localhost%s\n", cfg.Web.Addr)
+	}
+
+	fmt.Printf("running for %s against %s (index %q)…\n", cfg.Duration.D(), cfg.Addr, cfg.Index)
+	runner := workload.New(cfg, cli, orc, met, ctrl)
+
+	rctx, cancel := context.WithTimeout(ctx, cfg.Duration.D())
+	defer cancel()
+	go printLoop(rctx, rep, cfg.Sample.Interval.D())
+	runner.Run(rctx)
+
+	summary, err := rep.Finalize(cfg)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "finalize: %v\n", err)
+	}
+	fmt.Print(summary.Text())
+	fmt.Printf("artifacts: %s/summary.json, %s/metrics.csv\n", cfg.OutDir, cfg.OutDir)
+
+	// If the run finished on its own (not via Ctrl-C) and the dashboard is up,
+	// keep serving so the final state can be inspected.
+	if cfg.Web.Enabled && ctx.Err() == nil {
+		fmt.Println("workload complete — dashboard still live; press Ctrl-C to exit.")
+		<-ctx.Done()
+	}
+}
+
+func printLoop(ctx context.Context, rep *report.Reporter, every time.Duration) {
+	if every <= 0 {
+		every = 2 * time.Second
+	}
+	t := time.NewTicker(every)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			fmt.Println(rep.Line())
+		}
+	}
+}
+
+// parseMode extracts the subcommand (default "run") from the leading args.
+func parseMode(argv []string) (string, []string) {
+	if len(argv) > 0 {
+		switch argv[0] {
+		case "run", "reload-check":
+			return argv[0], argv[1:]
+		}
+	}
+	return "run", argv
+}
+
+func die(format string, a ...any) {
+	fmt.Fprintf(os.Stderr, "chatstress: "+format+"\n", a...)
+	os.Exit(1)
+}
diff --git a/06-chat-messaging-retention/kei-nan/cmd/chatstress/reload.go b/06-chat-messaging-retention/kei-nan/cmd/chatstress/reload.go
new file mode 100644
index 0000000..d100f3c
--- /dev/null
+++ b/06-chat-messaging-retention/kei-nan/cmd/chatstress/reload.go
@@ -0,0 +1,158 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"chatstress/internal/config"
+	"chatstress/internal/gendata"
+	"chatstress/internal/model"
+	"chatstress/internal/redisx"
+)
+
+// reloadCheck verifies the restart/RDB-reload correctness axis: load a mix of
+// long-lived and already-expired documents, take an in-process RDB save+load
+// round-trip via DEBUG RELOAD, and assert that (a) expired docs are never
+// visible (no stale hits) before or after the reload, and (b) live docs survive.
+//
+// Each document carries a UNIQUE text token (e.g. "expuid1234") so a scoped
+// search resolves to exactly one document, making the presence/absence checks
+// deterministic (unlike the vocabulary-based tokens used by the main workload).
+func reloadCheck(ctx context.Context, cfg *config.Config, cli *redisx.Client) error {
+	const (
+		nLive       = 2000
+		nExpire     = 2000
+		sampleCheck = 300
+		batchSz     = 500
+	)
+
+	fmt.Println("reload-check: RDB save/load with expired docs present")
+	if err := cli.FlushAll(ctx); err != nil {
+		return fmt.Errorf("flushall: %w", err)
+	}
+	if err := cli.CreateIndex(ctx, cfg.Prefix); err != nil {
+		return fmt.Errorf("create index: %w", err)
+	}
+
+	space := model.Space{
+		Tenants:           cfg.Tenants,
+		ChannelsPerTenant: cfg.ChannelsPerTenant,
+		UsersPerTenant:    cfg.UsersPerTenant,
+		ThreadsPerChannel: cfg.ThreadsPerChannel,
+	}
+	pick := model.NewPicker(space, cfg.Seed, 1)
+	gen := gendata.New(cfg.Seed, 2, cfg.Body.MinWords, cfg.Body.MaxWords, 0)
+
+	type rec struct{ key, channel, token string }
+	var seq int64
+
+	build := func(n int, ttl time.Duration, uniqPrefix string) ([]rec, error) {
+		recs := make([]rec, 0, n)
+		batch := make([]*model.Message, 0, batchSz)
+		flush := func() error {
+			if len(batch) == 0 {
+				return nil
+			}
+			err := cli.IngestBatch(ctx, batch)
+			batch = batch[:0]
+			return err
+		}
+		now := time.Now().UnixMilli()
+		for i := 0; i < n; i++ {
+			seq++
+			tid, tenant := pick.Tenant()
+			cid, channel := pick.ChannelID(tid)
+			base, _ := gen.Body()
+			token := uniqPrefix + strconv.FormatInt(seq, 10)
+			m := &model.Message{
+				Key: model.Key(cfg.Prefix, seq), Seq: seq,
+				Tenant: tenant, Channel: channel,
+				User: pick.User(tid), Thread: pick.Thread(cid),
+				Plan: "check", Body: base + " " + token, Token: token,
+				TTL: ttl, CreateMs: now,
+			}
+			batch = append(batch, m)
+			recs = append(recs, rec{key: m.Key, channel: channel, token: token})
+			if len(batch) == batchSz {
+				if err := flush(); err != nil {
+					return nil, err
+				}
+			}
+		}
+		return recs, flush()
+	}
+
+	live, err := build(nLive, time.Hour, "liveuid")
+	if err != nil {
+		return fmt.Errorf("load live docs: %w", err)
+	}
+	expd, err := build(nExpire, time.Second, "expuid")
+	if err != nil {
+		return fmt.Errorf("load expiring docs: %w", err)
+	}
+
+	// Let the short TTLs expire and indexing settle.
+	time.Sleep(2500 * time.Millisecond)
+
+	has := func(r rec) bool {
+		q := "@" + model.FieldChannel + ":{" + r.channel + "} " + r.token
+		_, keys, err := cli.Search(ctx, q, 10, 0)
+		if err != nil {
+			return false
+		}
+		for _, k := range keys {
+			if k == r.key {
+				return true
+			}
+		}
+		return false
+	}
+	sample := func(rs []rec, n int) []rec {
+		if len(rs) <= n {
+			return rs
+		}
+		step := len(rs) / n
+		out := make([]rec, 0, n)
+		for i := 0; i < len(rs); i += step {
+			out = append(out, rs[i])
+		}
+		return out
+	}
+	countFound := func(rs []rec) int {
+		c := 0
+		for _, r := range rs {
+			if has(r) {
+				c++
+			}
+		}
+		return c
+	}
+
+	liveS := sample(live, sampleCheck)
+	expS := sample(expd, sampleCheck)
+
+	preLive := countFound(liveS)
+	preStale := countFound(expS)
+	fmt.Printf("pre-reload:  live visible %d/%d, expired-visible %d/%d\n", preLive, len(liveS), preStale, len(expS))
+
+	fmt.Println("issuing DEBUG RELOAD (RDB save + load)…")
+	if err := cli.DebugReload(ctx); err != nil {
+		return fmt.Errorf("DEBUG RELOAD (needs enable-debug-command): %w", err)
+	}
+	time.Sleep(500 * time.Millisecond)
+
+	postLive := countFound(liveS)
+	postStale := countFound(expS)
+	fmt.Printf("post-reload: live visible %d/%d, expired-visible %d/%d\n", postLive, len(liveS), postStale, len(expS))
+
+	if preStale > 0 || postStale > 0 {
+		return fmt.Errorf("expired docs were visible (pre=%d, post=%d) — STALE HITS", preStale, postStale)
+	}
+	if postLive < len(liveS)*8/10 {
+		fmt.Printf("WARNING: post-reload live recall low (%d/%d) — indexing lag or a real regression\n", postLive, len(liveS))
+	}
+	fmt.Println("reload-check PASSED: no stale hits before/after RDB reload; live docs preserved.")
+	return nil
+}
diff --git a/06-chat-messaging-retention/kei-nan/internal/config/config.go b/06-chat-messaging-retention/kei-nan/internal/config/config.go
index 0406064..57f4a8c 100644
--- a/06-chat-messaging-retention/kei-nan/internal/config/config.go
+++ b/06-chat-messaging-retention/kei-nan/internal/config/config.go
@@ -60,6 +60,7 @@ var KnownProfiles = map[string]string{
 
 // Config is the full harness configuration.
 type Config struct {
+	URL           string   `yaml:"url"` // full redis://|rediss:// URL; overrides addr/username/password/tls
 	Addr          string   `yaml:"addr"`
 	Username      string   `yaml:"username"` // ACL user (cloud); empty = default
 	Password      string   `yaml:"password"`
diff --git a/06-chat-messaging-retention/kei-nan/internal/fuzz/fuzz_test.go b/06-chat-messaging-retention/kei-nan/internal/fuzz/fuzz_test.go
new file mode 100644
index 0000000..b6e6abe
--- /dev/null
+++ b/06-chat-messaging-retention/kei-nan/internal/fuzz/fuzz_test.go
@@ -0,0 +1,107 @@
+package fuzz
+
+import (
+	"math/rand"
+	"regexp"
+	"strings"
+	"testing"
+
+	"chatstress/internal/gendata"
+	"chatstress/internal/model"
+	"chatstress/internal/oracle"
+)
+
+// hasArg reports whether the arg list contains tok as a standalone string arg
+// (case-insensitive) — i.e. a command keyword, not text inside the query body.
+func hasArg(args []any, tok string) bool {
+	for _, a := range args {
+		if s, ok := a.(string); ok && strings.EqualFold(s, tok) {
+			return true
+		}
+	}
+	return false
+}
+
+// TestFuzzDiskLegal is the fuzzer's core contract: over many generated queries it
+// must NEVER emit an argument that Flex/BigRedis rejects.
+func TestFuzzDiskLegal(t *testing.T) {
+	sp := model.Space{Tenants: 100, ChannelsPerTenant: 10, UsersPerTenant: 50, ThreadsPerChannel: 8}
+	pick := model.NewPicker(sp, 1, 7)
+	gen := gendata.New(1, 9, 4, 12, 100000)
+	rnd := rand.New(rand.NewSource(0xC0FFEE))
+	params := Params{Tiers: []string{"free", "pro", "enterprise"}, PageDepth: 2000, Limit: 25, TimeoutMs: 300}
+	sample := oracle.Sample{Key: "msg:000000000001", Channel: "c5", Token: "deploy"}
+
+	// TAG clause: @<field>:{ <content> } — content must be alphanumeric / _ / | (OR)
+	// only. Any *, ?, %, [, ] etc. would be a prefix/wildcard/lexrange TAG query,
+	// which disk rejects.
+	tagRe := regexp.MustCompile(`@[a-z_]+:\{([^}]*)\}`)
+	tagContentOK := regexp.MustCompile(`^[A-Za-z0-9_|]+$`)
+
+	// Forbidden as a standalone arg on ANY Flex query.
+	forbiddenAny := []string{"SUMMARIZE", "HIGHLIGHT", "SLOP", "INORDER", "GEOFILTER", "WITHCURSOR", "WITHSUFFIXTRIE", "PAYLOAD"}
+
+	for i := 0; i < 20000; i++ {
+		hasLive := i%2 == 0
+		q := Build(rnd, pick, gen, params, sample, hasLive)
+
+		if q.Cmd != "FT.SEARCH" && q.Cmd != "FT.AGGREGATE" {
+			t.Fatalf("unexpected command %q: %s", q.Cmd, q.Display)
+		}
+		if len(q.Args) == 0 {
+			t.Fatalf("empty args: %s", q.Display)
+		}
+		// Args[0] is the free-text query EXPRESSION (arbitrary vocabulary words, e.g.
+		// "payload" or "filter" — legitimate search terms). Command keywords can only
+		// appear as OPTION args after it, so scan Args[1:] for forbidden keywords.
+		opts := q.Args[1:]
+		for _, bad := range forbiddenAny {
+			if hasArg(opts, bad) {
+				t.Fatalf("emitted forbidden arg %q: %s", bad, q.Display)
+			}
+		}
+		if q.Cmd == "FT.SEARCH" {
+			if hasArg(opts, "SORTBY") {
+				t.Fatalf("FT.SEARCH must not use SORTBY on disk (vector-only): %s", q.Display)
+			}
+			if hasArg(opts, "FILTER") {
+				t.Fatalf("FT.SEARCH must not use a numeric FILTER on disk: %s", q.Display)
+			}
+			if !hasArg(opts, "DIALECT") {
+				t.Fatalf("FT.SEARCH must carry DIALECT (for wildcard/fuzzy): %s", q.Display)
+			}
+		}
+		// The query expression is always the first arg.
+		expr, ok := q.Args[0].(string)
+		if !ok {
+			t.Fatalf("first arg not a string: %s", q.Display)
+		}
+		for _, m := range tagRe.FindAllStringSubmatch(expr, -1) {
+			if !tagContentOK.MatchString(m[1]) {
+				t.Fatalf("illegal TAG content %q (prefix/wildcard/lexrange not allowed on disk): %s", m[1], q.Display)
+			}
+		}
+	}
+}
+
+// TestFuzzBothCommandsAppear guards that the fuzzer actually exercises both
+// FT.SEARCH and FT.AGGREGATE (a regression that only ever emitted one would
+// silently shrink coverage).
+func TestFuzzBothCommandsAppear(t *testing.T) {
+	pick := model.NewPicker(model.Space{Tenants: 10, ChannelsPerTenant: 5, UsersPerTenant: 10, ThreadsPerChannel: 4}, 2, 3)
+	gen := gendata.New(2, 4, 4, 10, 0)
+	rnd := rand.New(rand.NewSource(1))
+	p := Params{Tiers: []string{"free"}, PageDepth: 100, Limit: 10}
+	var search, agg int
+	for i := 0; i < 2000; i++ {
+		switch Build(rnd, pick, gen, p, oracle.Sample{}, false).Cmd {
+		case "FT.SEARCH":
+			search++
+		case "FT.AGGREGATE":
+			agg++
+		}
+	}
+	if search == 0 || agg == 0 {
+		t.Fatalf("expected both command types; got search=%d agg=%d", search, agg)
+	}
+}
diff --git a/06-chat-messaging-retention/kei-nan/internal/oracle/oracle_test.go b/06-chat-messaging-retention/kei-nan/internal/oracle/oracle_test.go
new file mode 100644
index 0000000..028a8d6
--- /dev/null
+++ b/06-chat-messaging-retention/kei-nan/internal/oracle/oracle_test.go
@@ -0,0 +1,104 @@
+package oracle
+
+import (
+	"math"
+	"testing"
+	"time"
+)
+
+const (
+	testGrace = 1500 // ms: deletion de-index grace
+	testSkew  = 1000 // ms: expiry clock-skew tolerance
+	t0        = int64(1_700_000_000_000)
+)
+
+func newTestOracle() *Oracle {
+	// sampleRate 1.0 => every key is tracked, so Check is deterministic.
+	return New(8, 1_000_000, 1.0, testGrace, testSkew)
+}
+
+// TestOracleCheckExpiry covers the t0/skew rule for the expiry classification.
+func TestOracleCheckExpiry(t *testing.T) {
+	cases := []struct {
+		name     string
+		expireAt int64
+		want     Violation
+	}{
+		{"expired well before t0 (beyond skew)", t0 - 5000, Expired},
+		{"expired just past the skew boundary", t0 - testSkew - 1, Expired},
+		{"expired within skew tolerance", t0 - testSkew + 1, None},
+		{"expires exactly at t0 (not yet, within skew)", t0, None},
+		{"not expired", t0 + 5000, None},
+		{"persistent (never expires)", math.MaxInt64, None},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			o := newTestOracle()
+			o.Add("k", "c1", "tok", c.expireAt, 1)
+			if got := o.Check("k", t0); got != c.want {
+				t.Fatalf("Check=%v, want %v (expireAt=%d, t0=%d, skew=%d)", got, c.want, c.expireAt, t0, testSkew)
+			}
+		})
+	}
+}
+
+// TestOracleCheckDeletion covers the grace window for the deletion classification.
+func TestOracleCheckDeletion(t *testing.T) {
+	cases := []struct {
+		name      string
+		deletedAt int64
+		want      Violation
+	}{
+		{"deleted well before t0 (beyond grace)", t0 - 5000, Deleted},
+		{"deleted just past the grace boundary", t0 - testGrace - 1, Deleted},
+		{"deleted within grace (async de-index tolerated)", t0 - testGrace + 1, None},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			o := newTestOracle()
+			o.Add("k", "c1", "tok", math.MaxInt64, 1) // persistent so expiry can't fire
+			o.MarkDeleted("k", c.deletedAt)
+			if got := o.Check("k", t0); got != c.want {
+				t.Fatalf("Check=%v, want %v (deletedAt=%d, t0=%d, grace=%d)", got, c.want, c.deletedAt, t0, testGrace)
+			}
+		})
+	}
+}
+
+// TestOracleCheckUntracked: a key the oracle never saw is not classifiable.
+func TestOracleCheckUntracked(t *testing.T) {
+	o := newTestOracle()
+	if got := o.Check("never-added", t0); got != None {
+		t.Fatalf("Check on untracked key = %v, want None", got)
+	}
+}
+
+// TestOracleCheckCounters: a genuine expired stale hit is recorded exactly once.
+func TestOracleCheckCounters(t *testing.T) {
+	o := newTestOracle()
+	o.Add("k", "c1", "tok", t0-5000, 1)
+	o.Check("k", t0)
+	o.Check("k", t0)
+	s := o.Stats()
+	if s.StaleExpired != 2 || s.StaleDeleted != 0 {
+		t.Fatalf("staleExpired=%d staleDeleted=%d, want 2/0", s.StaleExpired, s.StaleDeleted)
+	}
+	if o.StaleHits() != 2 {
+		t.Fatalf("StaleHits=%d, want 2", o.StaleHits())
+	}
+}
+
+// TestOraclePrune removes only entries that expired/were deleted well in the past.
+func TestOraclePrune(t *testing.T) {
+	o := newTestOracle()
+	o.Add("live", "c1", "tok", math.MaxInt64, 1)   // persistent -> keep
+	o.Add("recent", "c1", "tok", math.MaxInt64, 2) // will delete recently -> keep
+	o.MarkDeleted("recent", time.Now().UnixMilli())
+	before := o.Stats().Tracked
+	if before != 2 {
+		t.Fatalf("tracked=%d, want 2", before)
+	}
+	if removed := o.Prune(); removed != 0 {
+		t.Fatalf("Prune removed %d, want 0 (nothing is old enough)", removed)
+	}
+}
diff --git a/06-chat-messaging-retention/kei-nan/internal/redisx/client.go b/06-chat-messaging-retention/kei-nan/internal/redisx/client.go
index 606f3f6..5344588 100644
--- a/06-chat-messaging-retention/kei-nan/internal/redisx/client.go
+++ b/06-chat-messaging-retention/kei-nan/internal/redisx/client.go
@@ -26,9 +26,12 @@ type Client struct {
 	index string
 }
 
-// Options configures the connection. Username/Password/TLS support cloud
-// endpoints (e.g. Redis Cloud / Enterprise Flex over TLS with ACL auth).
+// Options configures the connection. A full URL (redis://|rediss://, typically
+// from a cloud provider) takes precedence; otherwise Addr + Username/Password/TLS
+// are used. All three support cloud endpoints (e.g. Redis Cloud / Enterprise Flex
+// over TLS with ACL auth).
 type Options struct {
+	URL           string // full connection URL; overrides Addr/Username/Password/TLS
 	Addr          string
 	Username      string
 	Password      string
@@ -41,26 +44,49 @@ type Options struct {
 // New dials the server. RESP2 is forced so FT.INFO / FT.SEARCH replies are plain
 // arrays that are simple to parse.
 func New(o Options) (*Client, error) {
-	ropts := &redis.Options{
-		Addr:     o.Addr,
-		Username: o.Username,
-		Password: o.Password,
-		Protocol: 2,
-		PoolSize: o.PoolSize,
-	}
-	if o.TLS {
-		ropts.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: o.TLSSkipVerify}
+	var ropts *redis.Options
+	if o.URL != "" {
+		p, err := redis.ParseURL(o.URL)
+		if err != nil {
+			// Deliberately do NOT include the URL in the error — it carries the
+			// password.
+			return nil, fmt.Errorf("parse redis url: %w", err)
+		}
+		ropts = p
+		if o.TLSSkipVerify && ropts.TLSConfig != nil {
+			ropts.TLSConfig.InsecureSkipVerify = true
+		}
+	} else {
+		ropts = &redis.Options{Addr: o.Addr, Username: o.Username, Password: o.Password}
+		if o.TLS {
+			ropts.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: o.TLSSkipVerify}
+		}
 	}
+	ropts.Protocol = 2 // force RESP2 regardless of what the URL requested
+	ropts.PoolSize = o.PoolSize
 	rdb := redis.NewClient(ropts)
 	c := &Client{rdb: rdb, index: o.Index}
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	if err := rdb.Ping(ctx).Err(); err != nil {
-		return nil, fmt.Errorf("ping %s: %w", o.Addr, err)
+		return nil, fmt.Errorf("ping %s: %w", ropts.Addr, err) // ropts.Addr is host:port only
 	}
 	return c, nil
 }
 
+// AddrFromURL returns just the host:port of a connection URL, with no
+// credentials, for display/logging. Returns "" if url is empty or unparseable.
+func AddrFromURL(url string) string {
+	if url == "" {
+		return ""
+	}
+	p, err := redis.ParseURL(url)
+	if err != nil {
+		return ""
+	}
+	return p.Addr
+}
+
 // Close releases the connection pool.
 func (c *Client) Close() error { return c.rdb.Close() }
 
diff --git a/06-chat-messaging-retention/kei-nan/internal/workload/limiter.go b/06-chat-messaging-retention/kei-nan/internal/workload/limiter.go
index 991b619..d853482 100644
--- a/06-chat-messaging-retention/kei-nan/internal/workload/limiter.go
+++ b/06-chat-messaging-retention/kei-nan/internal/workload/limiter.go
@@ -68,3 +68,95 @@ func (l *Limiter) Close() {
 		close(l.stop)
 	}
 }
+
+// RateLimiter is a shared token-bucket whose target rate is read LIVE on every
+// tick (0 = unlimited). Unlike a per-worker pre-sleep, tokens are produced
+// independently of query latency, so N workers blocking on Wait() deliver the
+// configured aggregate rate (capped only by what the workers can actually
+// sustain). Used by the query workers so the "rate" control is honored.
+type RateLimiter struct {
+	ch   chan struct{}
+	rate func() int
+	stop chan struct{}
+}
+
+// NewRateLimiter starts a token producer reading rate() live.
+func NewRateLimiter(rate func() int) *RateLimiter {
+	l := &RateLimiter{ch: make(chan struct{}, 4096), rate: rate, stop: make(chan struct{})}
+	go l.run()
+	return l
+}
+
+func (l *RateLimiter) run() {
+	const tick = 25 * time.Millisecond
+	t := time.NewTicker(tick)
+	defer t.Stop()
+	var acc float64 // fractional token accumulator (handles low rates)
+	prev := -1      // last observed rate, to detect changes
+	for {
+		select {
+		case <-l.stop:
+			return
+		case <-t.C:
+			r := l.rate()
+			// On any rate DROP (including ->0), flush buffered permits. Otherwise a
+			// backlog of stale tokens (e.g. accumulated while unlimited, or when
+			// workers are latency-bound and can't keep up) lets workers burst at
+			// full capacity for seconds before the new, lower rate takes effect.
+			if r < prev {
+				drain(l.ch)
+			}
+			prev = r
+			if r <= 0 {
+				acc = 0 // unlimited: Wait() returns without needing a token
+				continue
+			}
+			acc += float64(r) * tick.Seconds()
+			n := int(acc)
+			acc -= float64(n)
+			for i := 0; i < n; i++ {
+				select {
+				case l.ch <- struct{}{}:
+				default: // bucket full: cap the burst, drop the surplus
+					acc = 0
+					i = n
+				}
+			}
+		}
+	}
+}
+
+// drain empties a token channel without blocking.
+func drain(ch chan struct{}) {
+	for {
+		select {
+		case <-ch:
+		default:
+			return
+		}
+	}
+}
+
+// Wait blocks for a token. An unlimited rate (<=0) returns immediately, and a
+// rate change is picked up within the poll interval even while blocked — so
+// switching TO unlimited never strands a worker waiting on the token channel,
+// and no bucket "top-up" hack is needed.
+func (l *RateLimiter) Wait(ctx context.Context) {
+	const poll = 50 * time.Millisecond
+	for {
+		if l.rate() <= 0 {
+			return
+		}
+		select {
+		case <-l.ch:
+			return
+		case <-ctx.Done():
+			return
+		case <-time.After(poll):
+			// re-check the rate (handles the ->0 transition without a token)
+		}
+	}
+}
+
+// Close stops the token producer.
+func (l *RateLimiter) Close() { close(l.stop) }
diff --git a/06-chat-messaging-retention/kei-nan/internal/workload/workload.go b/06-chat-messaging-retention/kei-nan/internal/workload/workload.go
index 2e88cdf..ec1b14d 100644
--- a/06-chat-messaging-retention/kei-nan/internal/workload/workload.go
+++ b/06-chat-messaging-retention/kei-nan/internal/workload/workload.go
@@ -72,11 +72,13 @@ func (r *Runner) Run(ctx context.Context) {
 	editLim := NewLimiter(r.cfg.Edit.Rate)
 	delLim := NewLimiter(r.cfg.Delete.Rate)
 	slideLim := NewLimiter(r.cfg.Sliding.Rate)
+	queryLim := NewRateLimiter(r.ctrl.QueryRate) // shared, live-rate token bucket
 	defer func() {
 		ingestLim.Close()
 		editLim.Close()
 		delLim.Close()
 		slideLim.Close()
+		queryLim.Close()
 	}()
 
 	for i := 0; i < r.cfg.Ingest.Workers; i++ {
@@ -84,9 +86,9 @@ func (r *Runner) Run(ctx context.Context) {
 		go func(id int) { defer wg.Done(); r.ingester(ctx, id, ingestLim) }(i)
 	}
 	// A supervisor spawns/stops query workers to match the live concurrency
-	// (threads) set from the UI; each self-paces to the live rate.
+	// (threads); all share queryLim so the aggregate rate is honored.
 	wg.Add(1)
-	go func() { defer wg.Done(); r.querySupervisor(ctx) }()
+	go func() { defer wg.Done(); r.querySupervisor(ctx, queryLim) }()
 
 	wg.Add(3)
 	go func() { defer wg.Done(); r.editor(ctx, editLim) }()
@@ -179,7 +181,7 @@ func (r *Runner) ingester(ctx context.Context, id int, lim *Limiter) {
 
 // querySupervisor keeps the number of live query workers equal to the control's
 // concurrency setting, spawning new workers or cancelling extras as it changes.
-func (r *Runner) querySupervisor(ctx context.Context) {
+func (r *Runner) querySupervisor(ctx context.Context, lim *RateLimiter) {
 	var workers []context.CancelFunc
 	var wwg sync.WaitGroup
 	nextID := 0
@@ -190,7 +192,7 @@ func (r *Runner) querySupervisor(ctx context.Context) {
 		id := nextID
 		nextID++
 		wwg.Add(1)
-		go func() { defer wwg.Done(); r.queryer(wctx, id) }()
+		go func() { defer wwg.Done(); r.queryer(wctx, id, lim) }()
 	}
 	shrink := func() {
 		last := len(workers) - 1
@@ -221,7 +223,7 @@ func (r *Runner) querySupervisor(ctx context.Context) {
 	}
 }
 
-func (r *Runner) queryer(ctx context.Context, id int) {
+func (r *Runner) queryer(ctx context.Context, id int, lim *RateLimiter) {
 	rnd := rand.New(rand.NewSource(r.cfg.Seed*53 + int64(id)*17 + 3))
 	slowMs := r.cfg.Query.SlowMs
 	iter := 0
@@ -233,7 +235,7 @@ func (r *Runner) queryer(ctx context.Context, id int) {
 			sleepCtx(ctx, 200*time.Millisecond)
 			continue
 		}
-		r.paceQuery(ctx)
+		lim.Wait(ctx) // shared token bucket: delivers the configured aggregate rate
 
 		item, ok := r.ctrl.PickPooled(rnd.Int())
 		if !ok {
@@ -319,27 +321,6 @@ func sleepCtx(ctx context.Context, d time.Duration) {
 	}
 }
 
-// paceQuery self-throttles a query worker to (liveRate / liveConcurrency) qps,
-// reading both live so the UI can change rate and thread count at runtime.
-// rate <= 0 means unlimited (load is then bounded only by the thread count).
-func (r *Runner) paceQuery(ctx context.Context) {
-	rate := r.ctrl.QueryRate()
-	workers := r.ctrl.Concurrency()
-	if rate <= 0 || workers <= 0 {
-		return
-	}
-	interval := time.Duration(int64(time.Second) * int64(workers) / int64(rate))
-	if interval <= 0 {
-		return
-	}
-	t := time.NewTimer(interval)
-	defer t.Stop()
-	select {
-	case <-t.C:
-	case <-ctx.Done():
-	}
-}
-
 func (r *Runner) editor(ctx context.Context, lim *Limiter) {
 	gen := gendata.New(r.cfg.Seed, 999, r.cfg.Body.MinWords, r.cfg.Body.MaxWords, r.cfg.Body.VocabHigh)
 	for {