From 4e46df2e678c9304cca0a2913ff424f65bf9f3f8 Mon Sep 17 00:00:00 2001
From: Lucas <lucasabreuoliveira@gmail.com>
Date: Thu, 30 Jul 2026 12:17:45 +0200
Subject: [PATCH 1/2] refactor(core): collapse speculative flow-store
 scaffolding
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The conversation flow engine carried a durable/multi-instance Store seam
that nothing built against. Cut it down to what the single-instance,
in-memory v1 actually uses:

- Remove the ConversationStore and expirer interfaces (one impl each,
  never injected) — fold memConversationStore's map straight into
  conversationManager.
- Drop ConversationState.Version — write-only, only ever self-incremented,
  read nowhere; it existed for a future compare-and-swap path.
- Replace the 256-way striped locks + inlined FNV-1a hasher with a single
  mutex. Sweep now holds it across the scan, so the expiredKeys snapshot +
  per-key re-check goes away.
- Remove sweepRecovered — with no custom store, sweep has no panic path.

Behavior is unchanged; the flow/advance/sweeper tests still pass under
-race. Net ~190 fewer lines, no dependency change.
---
 internal/core/conversation.go      | 216 +++++++----------------------
 internal/core/conversation_test.go | 212 +++++++++-------------------
 internal/core/core.go              |   4 +-
 internal/core/flow_test.go         |  60 ++++----
 4 files changed, 150 insertions(+), 342 deletions(-)

diff --git a/internal/core/conversation.go b/internal/core/conversation.go
index 85b2d9a..28b13e7 100644
--- a/internal/core/conversation.go
+++ b/internal/core/conversation.go
@@ -2,19 +2,12 @@ package core
 
 import (
 	"context"
-	"log/slog"
 	"maps"
 	"strings"
 	"sync"
 	"time"
 )
 
-// conversationShards is the fixed number of striped locks the conversationManager
-// holds. It bounds lock memory by construction: there is no per-key lock map that
-// grows with distinct conversations. Two keys that hash to the same shard
-// serialize unnecessarily, but with this many shards that is rare and brief.
-const conversationShards = 256
-
 // defaultSweepInterval is how often the background sweeper reaps expired flow
 // state when started without an explicit interval.
 const defaultSweepInterval = time.Minute
@@ -27,20 +20,13 @@ const defaultSweepInterval = time.Minute
 const defaultFlowTimeout = 10 * time.Minute
 
 // ConversationState is the per-conversation flow state the engine carries between
-// messages. It is flat and serializable by design so a future durable Store can
-// persist it without migration; secret answers are excluded from any serialized
-// form (see flow.go). Version bumps on every Set within a key's lifetime; it is
-// unused by the single-instance v1 (the in-process striped locks suffice) and is
-// exercised from day one so the eventual Store-backed compare-and-swap path needs
-// no state change. It is not preserved across a delete-and-recreate of the same
-// key (a swept conversation that restarts begins again at 1) — the multi-instance
-// CAS path owns cross-lifetime versioning when it lands.
+// messages, keyed per user+channel and held in memory for the life of the
+// process (a restart loses every in-flight flow).
 type ConversationState struct {
 	FlowID    string
 	Step      int
 	Answers   map[string]string
 	ExpiresAt time.Time
-	Version   uint64
 }
 
 // isExpired reports whether s has a set expiry that is at or before now. A zero
@@ -49,158 +35,62 @@ func (s ConversationState) isExpired(now time.Time) bool {
 	return !s.ExpiresAt.IsZero() && !s.ExpiresAt.After(now)
 }
 
-// ConversationStore persists per-conversation flow state by key. The default
-// implementation is in-memory (memConversationStore); the interface is the seam a
-// durable backend (e.g. Redis) drops into later. The conversationManager
-// serializes logical access per key with its striped locks, but a store is still
-// touched from the background sweeper, so implementations must be safe for
-// concurrent use.
-type ConversationStore interface {
-	Get(key string) (ConversationState, bool)
-	Set(key string, state ConversationState)
-	Delete(key string)
-}
-
-// expirer is the optional sweeper capability a ConversationStore may implement: it
-// reports the keys eligible for reaping at a given instant. memConversationStore
-// implements it; a backend with native TTLs (e.g. Redis) need not, and the
-// manager simply skips background sweeping for such a store. This mirrors the
-// AttachmentResolver optional-capability idiom in core.go.
-type expirer interface {
-	expiredKeys(now time.Time) []string
-}
-
-// memConversationStore is the default in-memory ConversationStore: a map guarded
-// by its own RWMutex, held only for the map access itself and never across a
-// validator. It is the volatile home for in-flight flows; a process restart loses
-// everything in it.
-type memConversationStore struct {
-	mu sync.RWMutex
-	m  map[string]ConversationState
-}
-
-func newMemConversationStore() *memConversationStore {
-	return &memConversationStore{m: make(map[string]ConversationState)}
-}
-
-func (s *memConversationStore) Get(key string) (ConversationState, bool) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	st, ok := s.m[key]
-	return st, ok
-}
-
-// Set stores state under key, bumping Version past the previous value for the key
-// (1 for a fresh key) so the field is always monotonic and correct.
-func (s *memConversationStore) Set(key string, state ConversationState) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	state.Version = s.m[key].Version + 1
-	s.m[key] = state
-}
-
-func (s *memConversationStore) Delete(key string) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	delete(s.m, key)
-}
-
-// expiredKeys returns the keys whose ExpiresAt is set and at or before now. A zero
-// ExpiresAt never expires.
-func (s *memConversationStore) expiredKeys(now time.Time) []string {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	var expired []string
-	for k, st := range s.m {
-		if st.isExpired(now) {
-			expired = append(expired, k)
-		}
-	}
-	return expired
-}
-
-// conversationManager owns the in-process coordination for flows: a fixed-size
-// striped lock set plus the background TTL sweeper. The ConversationStore behind
-// it owns only persistence. The user validator runs UNDER a shard lock, never
-// inside a store call, so a durable store can later sit behind compare-and-swap
-// without relocating the validator (which is arbitrary user Go code).
+// conversationManager owns the in-process coordination for flows: the per-
+// conversation state map plus the background TTL sweeper, both guarded by one
+// mutex.
+//
+// ponytail: single global mutex, not per-key striped locks. It serializes every
+// conversation's state transition — and the user validator that runs under it,
+// per Validate's "keep it fast" contract — against every other conversation. For
+// a single-instance in-memory chat bot that is ample; shard the lock by key only
+// if flow throughput is ever a measured bottleneck.
 type conversationManager struct {
-	store ConversationStore
-	locks [conversationShards]sync.Mutex
+	mu sync.Mutex
+	m  map[string]ConversationState
 }
 
 func newConversationManager() *conversationManager {
-	return &conversationManager{store: newMemConversationStore()}
-}
-
-// shardFor returns the lock guarding key. The same key always maps to the same
-// shard. FNV-1a is inlined over the key's bytes to avoid the per-call hasher heap
-// allocation fnv.New32a() incurs (it returns an interface).
-func (m *conversationManager) shardFor(key string) *sync.Mutex {
-	const (
-		offset32 = 2166136261
-		prime32  = 16777619
-	)
-	h := uint32(offset32)
-	for i := 0; i < len(key); i++ {
-		h ^= uint32(key[i])
-		h *= prime32
-	}
-	return &m.locks[h%conversationShards]
+	return &conversationManager{m: make(map[string]ConversationState)}
 }
 
-// withLock runs fn while holding key's shard lock, releasing it even if fn
-// panics. The panic propagates to dispatch's recover; the shard stays usable for
-// the next operation on any key it guards.
-func (m *conversationManager) withLock(key string, fn func()) {
-	mu := m.shardFor(key)
-	mu.Lock()
-	defer mu.Unlock()
+// withLock runs fn while holding the manager mutex, releasing it even if fn
+// panics (the panic propagates to dispatch's recover). The get/set/del helpers
+// assume this lock is held.
+func (m *conversationManager) withLock(fn func()) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	fn()
 }
 
-// sweepRecovered runs sweep with a recover so a panic in a custom
-// ConversationStore cannot take down the sweeper goroutine (and with it the
-// process), mirroring the recover that guards dispatch in core.go. logger (may be
-// nil → slog.Default) is passed in rather than held on the manager so a sweeper
-// goroutine never shares mutable diagnostics state with a concurrent reconnect.
-func (m *conversationManager) sweepRecovered(now time.Time, logger *slog.Logger) {
-	defer func() {
-		if r := recover(); r != nil {
-			if logger == nil {
-				logger = slog.Default()
-			}
-			logger.Error("botbooter: recovered from panic in conversation sweeper", "panic", r)
-		}
-	}()
-	m.sweep(now)
+// get, set and del are plain map operations; the caller MUST hold m.mu (via
+// withLock).
+func (m *conversationManager) get(key string) (ConversationState, bool) {
+	st, ok := m.m[key]
+	return st, ok
 }
 
-// sweep deletes every expired entry as of now. It takes each key's shard lock and
-// re-checks expiry under it, so it can never race a concurrent advance that just
-// refreshed the entry's TTL. A store without the expirer capability is a no-op.
+func (m *conversationManager) set(key string, st ConversationState) { m.m[key] = st }
+
+func (m *conversationManager) del(key string) { delete(m.m, key) }
+
+// sweep deletes every entry expired as of now. It holds the manager mutex across
+// the whole scan, so it can never race a concurrent advance that just refreshed
+// an entry's TTL (deleting during a map range is safe in Go).
 func (m *conversationManager) sweep(now time.Time) {
-	ex, ok := m.store.(expirer)
-	if !ok {
-		return
-	}
-	for _, key := range ex.expiredKeys(now) {
-		m.withLock(key, func() {
-			if st, ok := m.store.Get(key); ok && st.isExpired(now) {
-				m.store.Delete(key)
+	m.withLock(func() {
+		for key, st := range m.m {
+			if st.isExpired(now) {
+				delete(m.m, key)
 			}
-		})
-	}
+		}
+	})
 }
 
 // startSweeper runs sweep on an interval until ctx is done, then closes the
 // returned channel so a caller can observe that the goroutine has exited — making
 // the sweeper leak-free and testable. A non-positive interval falls back to
-// defaultSweepInterval. logger (may be nil) routes sweeper panic recovery through
-// the Bot's configured diagnostics sink; it is captured by the goroutine, not
-// stored on the manager, so a reconnect that starts a fresh sweeper never races a
-// lingering old one over shared state.
-func (m *conversationManager) startSweeper(ctx context.Context, interval time.Duration, logger *slog.Logger) <-chan struct{} {
+// defaultSweepInterval.
+func (m *conversationManager) startSweeper(ctx context.Context, interval time.Duration) <-chan struct{} {
 	if interval <= 0 {
 		interval = defaultSweepInterval
 	}
@@ -214,7 +104,7 @@ func (m *conversationManager) startSweeper(ctx context.Context, interval time.Du
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
-				m.sweepRecovered(time.Now(), logger)
+				m.sweep(time.Now())
 			}
 		}
 	}()
@@ -294,7 +184,7 @@ func (b *Bot) sendFlowMessage(ctx context.Context, channelID, text string) {
 	}
 }
 
-// start begins flow for msg's conversation. Under the shard lock it records the
+// start begins flow for msg's conversation. Under the lock it records the
 // initial state if and only if none exists (set-if-absent); after releasing the
 // lock it sends the first prompt. A losing racer — state already present — is a
 // no-op, and its trigger message is dropped, never consumed as the first answer.
@@ -302,11 +192,11 @@ func (m *conversationManager) start(ctx context.Context, b *Bot, msg *Message, f
 	key := conversationKey(msg)
 
 	var started bool
-	m.withLock(key, func() {
-		if _, ok := m.store.Get(key); ok {
+	m.withLock(func() {
+		if _, ok := m.get(key); ok {
 			return // already active for this conversation; drop the trigger
 		}
-		m.store.Set(key, ConversationState{
+		m.set(key, ConversationState{
 			FlowID:    flow.id,
 			Step:      0,
 			Answers:   map[string]string{},
@@ -333,7 +223,7 @@ func (m *conversationManager) advance(ctx context.Context, b *Bot, msg *Message)
 		handled bool
 		action  func()
 	)
-	m.withLock(key, func() {
+	m.withLock(func() {
 		handled, action = m.transitionLocked(ctx, b, key, msg, now)
 	})
 
@@ -353,7 +243,7 @@ func (m *conversationManager) advance(ctx context.Context, b *Bot, msg *Message)
 // reaped but reported handled (consuming the trigger) and runs the optional
 // OnTimeout. The asymmetry is deliberate — see the design spec.
 func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key string, msg *Message, now time.Time) (handled bool, action func()) {
-	state, ok := m.store.Get(key)
+	state, ok := m.get(key)
 	if !ok {
 		return false, nil // no active flow; dispatch continues to the command table
 	}
@@ -362,7 +252,7 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
 	if !ok {
 		// State outlived its flow registration (e.g. a renamed flow); reap and
 		// fall through rather than panic.
-		m.store.Delete(key)
+		m.del(key)
 		return false, nil
 	}
 
@@ -371,12 +261,12 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
 	// eaten by dispatch's recover but leave the state in place to wedge every
 	// later message until its TTL.)
 	if state.Step < 0 || state.Step >= len(flow.steps) {
-		m.store.Delete(key)
+		m.del(key)
 		return false, nil
 	}
 
 	if state.isExpired(now) {
-		m.store.Delete(key)
+		m.del(key)
 		if flow.onTimeout != nil {
 			return true, func() { flow.onTimeout(ctx, b, msg) }
 		}
@@ -391,7 +281,7 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
 
 	// The cancel word shadows every step, so it precedes validation.
 	if flow.cancelWord != "" && content == flow.cancelWord {
-		m.store.Delete(key)
+		m.del(key)
 		if flow.onCancel != nil {
 			return true, func() { flow.onCancel(ctx, b, msg) }
 		}
@@ -406,7 +296,7 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
 	// quiet lets it expire.
 	slideTTL := func() {
 		state.ExpiresAt = now.Add(flow.timeoutOrDefault())
-		m.store.Set(key, state)
+		m.set(key, state)
 	}
 
 	// Empty/whitespace answers are non-answers: re-prompt without storing an answer.
@@ -451,7 +341,7 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
 	// engine reusing the underlying map.
 	if state.Step == len(flow.steps)-1 {
 		completed := Answers(maps.Clone(answers))
-		m.store.Delete(key)
+		m.del(key)
 		return true, func() { flow.onComplete(ctx, b, msg, completed) }
 	}
 
diff --git a/internal/core/conversation_test.go b/internal/core/conversation_test.go
index 4de67ac..03be883 100644
--- a/internal/core/conversation_test.go
+++ b/internal/core/conversation_test.go
@@ -10,69 +10,23 @@ import (
 	"github.com/lao/botbooter/internal/asserts"
 )
 
-func TestMemConversationStore_GetSetDelete(t *testing.T) {
-	s := newMemConversationStore()
-
-	_, ok := s.Get("k")
-	asserts.False(t, ok, "missing key reports absent")
-
-	s.Set("k", ConversationState{FlowID: "f", Step: 2})
-	st, ok := s.Get("k")
-	asserts.True(t, ok, "present after Set")
-	asserts.Equal(t, st.FlowID, "f", "FlowID round-trips")
-	asserts.Equal(t, st.Step, 2, "Step round-trips")
-
-	s.Delete("k")
-	_, ok = s.Get("k")
-	asserts.False(t, ok, "absent after Delete")
-}
-
-func TestMemConversationStore_VersionBump(t *testing.T) {
-	s := newMemConversationStore()
-
-	s.Set("k", ConversationState{FlowID: "f"})
-	st, _ := s.Get("k")
-	asserts.Equal(t, st.Version, uint64(1), "first Set bumps Version to 1")
-
-	s.Set("k", ConversationState{FlowID: "f", Step: 1})
-	st, _ = s.Get("k")
-	asserts.Equal(t, st.Version, uint64(2), "second Set bumps Version to 2")
-
-	// A caller-supplied Version is ignored: the store owns the counter.
-	s.Set("k", ConversationState{FlowID: "f", Version: 99})
-	st, _ = s.Get("k")
-	asserts.Equal(t, st.Version, uint64(3), "store overrides a caller-supplied Version")
-}
-
-func TestMemConversationStore_ExpiredKeys(t *testing.T) {
-	s := newMemConversationStore()
-	now := time.Now()
-	s.Set("at", ConversationState{ExpiresAt: now}) // == now → expired
-	s.Set("past", ConversationState{ExpiresAt: now.Add(-time.Second)})
-	s.Set("future", ConversationState{ExpiresAt: now.Add(time.Hour)})
-	s.Set("zero", ConversationState{}) // no expiry
-
-	set := map[string]bool{}
-	for _, k := range s.expiredKeys(now) {
-		set[k] = true
-	}
-
-	asserts.True(t, set["at"], "ExpiresAt == now is expired")
-	asserts.True(t, set["past"], "past ExpiresAt is expired")
-	asserts.False(t, set["future"], "future ExpiresAt is not expired")
-	asserts.False(t, set["zero"], "zero ExpiresAt never expires")
-	asserts.Equal(t, len(set), 2, "exactly two keys expired")
-}
-
-func TestConversationManager_StripedLockBounded(t *testing.T) {
+func TestConversationManager_GetSetDel(t *testing.T) {
 	m := newConversationManager()
 
-	// Lock memory is bounded by construction: a fixed array, not a per-key map.
-	asserts.Equal(t, len(m.locks), conversationShards, "fixed shard count")
+	m.withLock(func() {
+		_, ok := m.get("k")
+		asserts.False(t, ok, "missing key reports absent")
+
+		m.set("k", ConversationState{FlowID: "f", Step: 2})
+		st, ok := m.get("k")
+		asserts.True(t, ok, "present after set")
+		asserts.Equal(t, st.FlowID, "f", "FlowID round-trips")
+		asserts.Equal(t, st.Step, 2, "Step round-trips")
 
-	a := m.shardFor("alpha")
-	b := m.shardFor("alpha")
-	asserts.True(t, a == b, "the same key maps to the same shard deterministically")
+		m.del("k")
+		_, ok = m.get("k")
+		asserts.False(t, ok, "absent after del")
+	})
 }
 
 func TestConversationManager_WithLockReleasesOnPanic(t *testing.T) {
@@ -80,70 +34,49 @@ func TestConversationManager_WithLockReleasesOnPanic(t *testing.T) {
 
 	func() {
 		defer func() { _ = recover() }()
-		m.withLock("k", func() { panic("boom") })
+		m.withLock(func() { panic("boom") })
 	}()
 
-	// The shard must have been released; a second op on the same key (same shard)
-	// must not deadlock.
+	// The mutex must have been released; a second op must not deadlock.
 	done := make(chan struct{})
 	go func() {
-		m.withLock("k", func() {})
+		m.withLock(func() {})
 		close(done)
 	}()
 
 	select {
 	case <-done:
 	case <-time.After(time.Second):
-		t.Fatal("withLock wedged the shard after a panic")
+		t.Fatal("withLock wedged the mutex after a panic")
 	}
 }
 
 func TestConversationManager_Sweep(t *testing.T) {
 	m := newConversationManager()
 	now := time.Now()
-	m.store.Set("expired", ConversationState{FlowID: "f", ExpiresAt: now.Add(-time.Minute)})
-	m.store.Set("fresh", ConversationState{FlowID: "f", ExpiresAt: now.Add(time.Minute)})
-	m.store.Set("noexpiry", ConversationState{FlowID: "f"})
-
-	m.sweep(now)
-
-	_, ok := m.store.Get("expired")
-	asserts.False(t, ok, "expired entry is swept")
-	_, ok = m.store.Get("fresh")
-	asserts.True(t, ok, "fresh entry survives")
-	_, ok = m.store.Get("noexpiry")
-	asserts.True(t, ok, "zero-ExpiresAt entry survives")
-}
-
-// fakeExpirer reports a fixed key set as expired regardless of the store's actual
-// contents, letting a test drive sweep's under-lock re-check independently of the
-// snapshot.
-type fakeExpirer struct {
-	*memConversationStore
-	report []string
-}
-
-func (f *fakeExpirer) expiredKeys(time.Time) []string { return f.report }
-
-func TestConversationManager_SweepRechecksUnderLock(t *testing.T) {
-	store := newMemConversationStore()
-	now := time.Now()
-	// The snapshot claims "k" is expired, but the store holds a fresh "k" — as if
-	// advance refreshed the TTL between the snapshot and the delete.
-	store.Set("k", ConversationState{FlowID: "f", ExpiresAt: now.Add(time.Hour)})
-	m := &conversationManager{store: &fakeExpirer{memConversationStore: store, report: []string{"k"}}}
+	m.withLock(func() {
+		m.set("expired", ConversationState{FlowID: "f", ExpiresAt: now.Add(-time.Minute)})
+		m.set("fresh", ConversationState{FlowID: "f", ExpiresAt: now.Add(time.Minute)})
+		m.set("noexpiry", ConversationState{FlowID: "f"})
+	})
 
 	m.sweep(now)
 
-	_, ok := m.store.Get("k")
-	asserts.True(t, ok, "the under-lock re-check spares a still-fresh entry")
+	m.withLock(func() {
+		_, ok := m.get("expired")
+		asserts.False(t, ok, "expired entry is swept")
+		_, ok = m.get("fresh")
+		asserts.True(t, ok, "fresh entry survives")
+		_, ok = m.get("noexpiry")
+		asserts.True(t, ok, "zero-ExpiresAt entry survives")
+	})
 }
 
 func TestConversationManager_SweeperLifecycleExits(t *testing.T) {
 	m := newConversationManager()
 	ctx, cancel := context.WithCancel(context.Background())
 
-	done := m.startSweeper(ctx, 10*time.Millisecond, nil)
+	done := m.startSweeper(ctx, 10*time.Millisecond)
 	cancel()
 
 	select {
@@ -155,15 +88,19 @@ func TestConversationManager_SweeperLifecycleExits(t *testing.T) {
 
 func TestConversationManager_SweeperReapsExpired(t *testing.T) {
 	m := newConversationManager()
-	m.store.Set("k", ConversationState{FlowID: "f", ExpiresAt: time.Now().Add(-time.Minute)})
+	m.withLock(func() {
+		m.set("k", ConversationState{FlowID: "f", ExpiresAt: time.Now().Add(-time.Minute)})
+	})
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	m.startSweeper(ctx, 5*time.Millisecond, nil)
+	m.startSweeper(ctx, 5*time.Millisecond)
 
 	deadline := time.After(2 * time.Second)
 	for {
-		if _, ok := m.store.Get("k"); !ok {
+		var present bool
+		m.withLock(func() { _, present = m.get("k") })
+		if !present {
 			return
 		}
 		select {
@@ -230,8 +167,11 @@ func TestConversationManager_AdvanceHappyPath(t *testing.T) {
 	asserts.Equal(t, got[0], "name?", "first prompt")
 	asserts.Equal(t, got[1], "color?", "second prompt")
 
-	_, ok := bot.conversations.store.Get(conversationKey(msgFrom("u", "c", "")))
-	asserts.False(t, ok, "state cleared after completion")
+	var present bool
+	bot.conversations.withLock(func() {
+		_, present = bot.conversations.get(conversationKey(msgFrom("u", "c", "")))
+	})
+	asserts.False(t, present, "state cleared after completion")
 }
 
 func TestConversationManager_SecretAnswerKeepsExactBytes(t *testing.T) {
@@ -363,39 +303,16 @@ func TestBot_SweeperLifecycle_ConnectDisconnect(t *testing.T) {
 	}
 }
 
-// panicStore is a ConversationStore whose Delete panics, used to prove the
-// background sweeper recovers rather than crashing the process. expiredKeys/Get
-// are promoted from the embedded memConversationStore.
-type panicStore struct {
-	*memConversationStore
-}
-
-func (p *panicStore) Delete(string) { panic("store delete exploded") }
-
-func TestConversationManager_SweeperRecoversFromStorePanic(t *testing.T) {
-	store := newMemConversationStore()
-	store.Set("k", ConversationState{FlowID: "f", ExpiresAt: time.Now().Add(-time.Minute)})
-	m := &conversationManager{store: &panicStore{memConversationStore: store}}
-
-	// sweepRecovered must swallow the store's panic; if it propagated, the test
-	// binary would crash here.
-	m.sweepRecovered(time.Now(), nil)
-
-	// Delete panicked before removing the entry, so it remains — and the manager
-	// is still alive.
-	_, ok := store.Get("k")
-	asserts.True(t, ok, "a panicking Delete is recovered; the sweeper did not crash")
-}
-
 // TestConversationManager_SweepConcurrentWithAdvance hammers sweep against a
-// goroutine that keeps refreshing a key's TTL under the shard lock. Both touch
-// the same shard, so under -race this asserts no data race and that an actively
-// refreshed entry is never reaped (the under-lock re-check always sees a future
-// ExpiresAt).
+// goroutine that keeps refreshing a key's TTL under the manager lock. Under -race
+// this asserts no data race and that an actively refreshed entry is never reaped
+// (sweep holds the lock across its scan, so it always sees a future ExpiresAt).
 func TestConversationManager_SweepConcurrentWithAdvance(t *testing.T) {
 	m := newConversationManager()
 	key := "u\x00c"
-	m.store.Set(key, ConversationState{FlowID: "f", ExpiresAt: time.Now().Add(time.Hour)})
+	m.withLock(func() {
+		m.set(key, ConversationState{FlowID: "f", ExpiresAt: time.Now().Add(time.Hour)})
+	})
 
 	const rounds = 2000
 	var wg sync.WaitGroup
@@ -403,13 +320,13 @@ func TestConversationManager_SweepConcurrentWithAdvance(t *testing.T) {
 	go func() {
 		defer wg.Done()
 		for i := 0; i < rounds; i++ {
-			m.withLock(key, func() {
-				st, ok := m.store.Get(key)
+			m.withLock(func() {
+				st, ok := m.get(key)
 				if !ok {
 					st = ConversationState{FlowID: "f"}
 				}
 				st.ExpiresAt = time.Now().Add(time.Hour)
-				m.store.Set(key, st)
+				m.set(key, st)
 			})
 		}
 	}()
@@ -421,11 +338,12 @@ func TestConversationManager_SweepConcurrentWithAdvance(t *testing.T) {
 	}()
 	wg.Wait()
 
-	_, ok := m.store.Get(key)
+	var ok bool
+	m.withLock(func() { _, ok = m.get(key) })
 	asserts.True(t, ok, "an actively-refreshed entry is never swept")
 }
 
-// TestConversationManager_ConcurrentWithLock proves the shard locks serialize the
+// TestConversationManager_ConcurrentWithLock proves the manager lock serializes the
 // read-modify-write for a key: 50 increments across 5 keys land exactly once each
 // with no lost updates. Run under -race to also assert no data race on the map.
 func TestConversationManager_ConcurrentWithLock(t *testing.T) {
@@ -439,21 +357,23 @@ func TestConversationManager_ConcurrentWithLock(t *testing.T) {
 		go func(i int) {
 			defer wg.Done()
 			key := "k" + strconv.Itoa(i%keys)
-			m.withLock(key, func() {
-				st, _ := m.store.Get(key)
+			m.withLock(func() {
+				st, _ := m.get(key)
 				st.FlowID = "f"
 				st.Step++
-				m.store.Set(key, st)
+				m.set(key, st)
 			})
 		}(i)
 	}
 	wg.Wait()
 
 	total := 0
-	for i := 0; i < keys; i++ {
-		st, ok := m.store.Get("k" + strconv.Itoa(i))
-		asserts.True(t, ok, "key should exist after concurrent writes")
-		total += st.Step
-	}
+	m.withLock(func() {
+		for i := 0; i < keys; i++ {
+			st, ok := m.get("k" + strconv.Itoa(i))
+			asserts.True(t, ok, "key should exist after concurrent writes")
+			total += st.Step
+		}
+	})
 	asserts.Equal(t, total, goroutines, "every increment applied exactly once (no lost updates)")
 }
diff --git a/internal/core/core.go b/internal/core/core.go
index ab4ff6a..9f82abc 100644
--- a/internal/core/core.go
+++ b/internal/core/core.go
@@ -472,10 +472,10 @@ func (b *Bot) Connect(ctx context.Context) error {
 	// sweeper to observe cancellation and exit (production teardown never blocks
 	// on sweeperDone), so a fast reconnect can briefly run two sweepers against
 	// the shared per-Bot conversationManager. That overlap is benign: sweep
-	// re-checks expiry under each shard lock before deleting, so a concurrent
+	// re-checks expiry under the manager lock before deleting, so a concurrent
 	// double-sweep is idempotent.
 	if b.conversations != nil && len(b.flows) > 0 {
-		c.sweeperDone = b.conversations.startSweeper(runCtx, defaultSweepInterval, b.log())
+		c.sweeperDone = b.conversations.startSweeper(runCtx, defaultSweepInterval)
 	}
 	return nil
 }
diff --git a/internal/core/flow_test.go b/internal/core/flow_test.go
index a11a7fc..e111564 100644
--- a/internal/core/flow_test.go
+++ b/internal/core/flow_test.go
@@ -102,13 +102,13 @@ func TestFlow_ValidatorRePromptsSameStep(t *testing.T) {
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "bad")) // invalid
 
-	s, _ := bot.conversations.store.Get(key)
+	s, _ := bot.conversations.get(key)
 	asserts.Equal(t, s.Step, 0, "stayed on step 0 after an invalid answer")
 	_, stored := s.Answers["email"]
 	asserts.False(t, stored, "invalid answer is not stored")
 
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "a@b.com")) // valid
-	s2, _ := bot.conversations.store.Get(key)
+	s2, _ := bot.conversations.get(key)
 	asserts.Equal(t, s2.Step, 1, "advanced after a valid answer")
 
 	got := adapter.messages()
@@ -129,7 +129,7 @@ func TestFlow_EmptyAnswerRePrompts(t *testing.T) {
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "   ")) // whitespace = non-answer
 
-	s, _ := bot.conversations.store.Get(key)
+	s, _ := bot.conversations.get(key)
 	asserts.Equal(t, s.Step, 0, "stayed on step 0 for an empty answer")
 	asserts.Equal(t, len(s.Answers), 0, "empty answer is not stored")
 
@@ -154,7 +154,7 @@ func TestFlow_CancelBailsAndRunsOnCancel(t *testing.T) {
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "cancel"))
 
 	asserts.True(t, canceled, "OnCancel ran")
-	_, ok := bot.conversations.store.Get(key)
+	_, ok := bot.conversations.get(key)
 	asserts.False(t, ok, "state cleared on cancel")
 }
 
@@ -182,11 +182,11 @@ func TestFlow_TTLSlidesOnEachStep(t *testing.T) {
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
-	s1, _ := bot.conversations.store.Get(key)
+	s1, _ := bot.conversations.get(key)
 
 	time.Sleep(2 * time.Millisecond) // ensure the clock advances between steps
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "alpha"))
-	s2, _ := bot.conversations.store.Get(key)
+	s2, _ := bot.conversations.get(key)
 
 	asserts.True(t, s2.ExpiresAt.After(s1.ExpiresAt), "TTL slides forward on each successful step")
 }
@@ -202,12 +202,12 @@ func TestFlow_TTLSlidesOnRejectedAnswer(t *testing.T) {
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
-	s1, _ := bot.conversations.store.Get(key)
+	s1, _ := bot.conversations.get(key)
 
 	time.Sleep(2 * time.Millisecond) // ensure the clock advances
 	handled := bot.conversations.advance(ctx, bot, msgFrom("u", "c", "bad"))
 	asserts.True(t, handled, "a rejected answer is consumed")
-	s2, ok := bot.conversations.store.Get(key)
+	s2, ok := bot.conversations.get(key)
 	asserts.True(t, ok, "state survives a rejected answer")
 	asserts.Equal(t, s2.Step, 0, "step does not advance on a rejected answer")
 	asserts.True(t, s2.ExpiresAt.After(s1.ExpiresAt), "TTL slides forward even when the answer is rejected")
@@ -223,7 +223,7 @@ func TestFlow_ExpiredStateTimesOut(t *testing.T) {
 		OnTimeout(func(context.Context, *Bot, *Message) { timedOut = true })
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
-	bot.conversations.store.Set(key, ConversationState{
+	bot.conversations.set(key, ConversationState{
 		FlowID:    "f",
 		ExpiresAt: time.Now().Add(-time.Minute),
 		Answers:   map[string]string{},
@@ -232,7 +232,7 @@ func TestFlow_ExpiredStateTimesOut(t *testing.T) {
 	handled := bot.conversations.advance(ctx, bot, msgFrom("u", "c", "late"))
 	asserts.True(t, handled, "an expired-state message is consumed")
 	asserts.True(t, timedOut, "OnTimeout ran")
-	_, ok := bot.conversations.store.Get(key)
+	_, ok := bot.conversations.get(key)
 	asserts.False(t, ok, "expired state is reaped")
 }
 
@@ -240,7 +240,7 @@ func TestFlow_UnregisteredFlowIDFallsThrough(t *testing.T) {
 	bot := New(SlackBotType, &recordingAdapter{})
 	key := conversationKey(msgFrom("u", "c", ""))
 
-	bot.conversations.store.Set(key, ConversationState{
+	bot.conversations.set(key, ConversationState{
 		FlowID:    "ghost",
 		ExpiresAt: time.Now().Add(time.Hour),
 		Answers:   map[string]string{},
@@ -248,14 +248,13 @@ func TestFlow_UnregisteredFlowIDFallsThrough(t *testing.T) {
 
 	handled := bot.conversations.advance(context.Background(), bot, msgFrom("u", "c", "hello"))
 	asserts.False(t, handled, "state for an unregistered flow falls through to the command table")
-	_, ok := bot.conversations.store.Get(key)
+	_, ok := bot.conversations.get(key)
 	asserts.False(t, ok, "stale state is reaped, not left to panic later")
 }
 
-func TestFlow_PanicInValidatorDoesNotWedgeShard(t *testing.T) {
+func TestFlow_PanicInValidatorDoesNotWedgeLock(t *testing.T) {
 	bot := New(SlackBotType, &recordingAdapter{})
 	ctx := context.Background()
-	key := conversationKey(msgFrom("u", "c", ""))
 
 	f := NewFlow("f").Ask("a", "a?", Validate(func(string) error { panic("validator boom") })).OnComplete(noopComplete)
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
@@ -268,23 +267,22 @@ func TestFlow_PanicInValidatorDoesNotWedgeShard(t *testing.T) {
 		bot.conversations.advance(ctx, bot, msgFrom("u", "c", "x"))
 	}()
 
-	// The shard must have been released by defer despite the panic.
+	// The manager mutex must have been released by defer despite the panic.
 	done := make(chan struct{})
 	go func() {
-		bot.conversations.withLock(key, func() {})
+		bot.conversations.withLock(func() {})
 		close(done)
 	}()
 	select {
 	case <-done:
 	case <-time.After(time.Second):
-		t.Fatal("a panic in the validator wedged the shard")
+		t.Fatal("a panic in the validator wedged the manager lock")
 	}
 }
 
-func TestFlow_PanicInOnCompleteDoesNotWedgeShard(t *testing.T) {
+func TestFlow_PanicInOnCompleteDoesNotWedgeLock(t *testing.T) {
 	bot := New(SlackBotType, &recordingAdapter{})
 	ctx := context.Background()
-	key := conversationKey(msgFrom("u", "c", ""))
 
 	f := NewFlow("f").Ask("a", "a?").OnComplete(func(context.Context, *Bot, *Message, Answers) { panic("complete boom") })
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
@@ -295,17 +293,17 @@ func TestFlow_PanicInOnCompleteDoesNotWedgeShard(t *testing.T) {
 		bot.conversations.advance(ctx, bot, msgFrom("u", "c", "alpha")) // last step -> OnComplete panics
 	}()
 
-	// OnComplete runs after the shard is released and state is deleted, so the
-	// shard must be free.
+	// OnComplete runs after the lock is released and state is deleted, so the
+	// mutex must be free.
 	done := make(chan struct{})
 	go func() {
-		bot.conversations.withLock(key, func() {})
+		bot.conversations.withLock(func() {})
 		close(done)
 	}()
 	select {
 	case <-done:
 	case <-time.After(time.Second):
-		t.Fatal("a panic in OnComplete wedged the shard")
+		t.Fatal("a panic in OnComplete wedged the manager lock")
 	}
 }
 
@@ -385,7 +383,7 @@ func TestFlow_OutOfRangeStepFallsThrough(t *testing.T) {
 	key := conversationKey(msgFrom("u", "c", ""))
 
 	// A durable store could load a state whose flow has since lost steps.
-	bot.conversations.store.Set(key, ConversationState{
+	bot.conversations.set(key, ConversationState{
 		FlowID:    "f",
 		Step:      99,
 		ExpiresAt: time.Now().Add(time.Hour),
@@ -394,7 +392,7 @@ func TestFlow_OutOfRangeStepFallsThrough(t *testing.T) {
 
 	handled := bot.conversations.advance(context.Background(), bot, msgFrom("u", "c", "hi"))
 	asserts.False(t, handled, "out-of-range step falls through to the command table")
-	_, ok := bot.conversations.store.Get(key)
+	_, ok := bot.conversations.get(key)
 	asserts.False(t, ok, "corrupt state is reaped, not left to wedge")
 
 	handled2 := bot.conversations.advance(context.Background(), bot, msgFrom("u", "c", "again"))
@@ -450,7 +448,7 @@ func TestFlow_ConcurrentAdvanceNoDeadlock(t *testing.T) {
 	}
 	wg.Wait()
 
-	if s, ok := bot.conversations.store.Get(key); ok {
+	if s, ok := bot.conversations.get(key); ok {
 		asserts.True(t, s.Step >= 0 && s.Step < len(f.steps), "step stays within bounds")
 	}
 }
@@ -464,7 +462,7 @@ func TestFlow_ExpiredStateWithoutOnTimeoutConsumesAndReaps(t *testing.T) {
 	f := NewFlow("f").Ask("a", "a?").OnComplete(noopComplete) // no OnTimeout
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
-	bot.conversations.store.Set(key, ConversationState{
+	bot.conversations.set(key, ConversationState{
 		FlowID:    "f",
 		ExpiresAt: time.Now().Add(-time.Minute),
 		Answers:   map[string]string{},
@@ -472,7 +470,7 @@ func TestFlow_ExpiredStateWithoutOnTimeoutConsumesAndReaps(t *testing.T) {
 
 	handled := bot.conversations.advance(context.Background(), bot, msgFrom("u", "c", "late"))
 	asserts.True(t, handled, "an expired-state message is consumed even with no OnTimeout set")
-	_, ok := bot.conversations.store.Get(key)
+	_, ok := bot.conversations.get(key)
 	asserts.False(t, ok, "expired state is reaped on the nil-OnTimeout default path")
 }
 
@@ -489,7 +487,7 @@ func TestFlow_CancelWithoutOnCancelConsumesAndReaps(t *testing.T) {
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
 	handled := bot.conversations.advance(ctx, bot, msgFrom("u", "c", "cancel"))
 	asserts.True(t, handled, "the cancel word is consumed even with no OnCancel set")
-	_, ok := bot.conversations.store.Get(key)
+	_, ok := bot.conversations.get(key)
 	asserts.False(t, ok, "state is cleared on the nil-OnCancel default path")
 }
 
@@ -517,13 +515,13 @@ func TestFlow_SendFailureStillAdvancesAndLogs(t *testing.T) {
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
-	s, ok := bot.conversations.store.Get(key)
+	s, ok := bot.conversations.get(key)
 	asserts.True(t, ok, "flow starts despite a failed first-prompt send")
 	asserts.Equal(t, s.Step, 0, "on step 0 after start")
 
 	handled := bot.conversations.advance(ctx, bot, msgFrom("u", "c", "alpha"))
 	asserts.True(t, handled, "answer is consumed even though the prompt send fails")
-	s2, _ := bot.conversations.store.Get(key)
+	s2, _ := bot.conversations.get(key)
 	asserts.Equal(t, s2.Step, 1, "flow advances to step 1 despite the send error")
 
 	asserts.True(t, strings.Contains(buf.String(), "failed to send flow prompt"),

From f4bf5c6ee7354888d259de3033a746742e6d9352 Mon Sep 17 00:00:00 2001
From: Lucas <lucasabreuoliveira@gmail.com>
Date: Thu, 30 Jul 2026 12:54:59 +0200
Subject: [PATCH 2/2] test(core): route flow_test state access through locked
 helpers

get/set/del are documented as requiring the manager mutex; flow_test was
calling them unguarded. Add getConvState/setConvState that wrap the access
in withLock so every site honors the contract.
---
 internal/core/flow_test.go | 56 ++++++++++++++++++++++++--------------
 1 file changed, 36 insertions(+), 20 deletions(-)

diff --git a/internal/core/flow_test.go b/internal/core/flow_test.go
index e111564..8107cc8 100644
--- a/internal/core/flow_test.go
+++ b/internal/core/flow_test.go
@@ -16,6 +16,22 @@ import (
 
 func noopComplete(context.Context, *Bot, *Message, Answers) {}
 
+// getConvState / setConvState read and write conversation state through the
+// manager mutex, honoring the get/set "caller must hold m.mu" contract from test
+// code (get/set are unexported map ops that assume the lock is held).
+func getConvState(m *conversationManager, key string) (ConversationState, bool) {
+	var (
+		st ConversationState
+		ok bool
+	)
+	m.withLock(func() { st, ok = m.get(key) })
+	return st, ok
+}
+
+func setConvState(m *conversationManager, key string, st ConversationState) {
+	m.withLock(func() { m.set(key, st) })
+}
+
 func TestNewFlow_Defaults(t *testing.T) {
 	f := NewFlow("f")
 	asserts.Equal(t, f.cancelWord, "cancel", "default cancel word")
@@ -102,13 +118,13 @@ func TestFlow_ValidatorRePromptsSameStep(t *testing.T) {
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "bad")) // invalid
 
-	s, _ := bot.conversations.get(key)
+	s, _ := getConvState(bot.conversations, key)
 	asserts.Equal(t, s.Step, 0, "stayed on step 0 after an invalid answer")
 	_, stored := s.Answers["email"]
 	asserts.False(t, stored, "invalid answer is not stored")
 
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "a@b.com")) // valid
-	s2, _ := bot.conversations.get(key)
+	s2, _ := getConvState(bot.conversations, key)
 	asserts.Equal(t, s2.Step, 1, "advanced after a valid answer")
 
 	got := adapter.messages()
@@ -129,7 +145,7 @@ func TestFlow_EmptyAnswerRePrompts(t *testing.T) {
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "   ")) // whitespace = non-answer
 
-	s, _ := bot.conversations.get(key)
+	s, _ := getConvState(bot.conversations, key)
 	asserts.Equal(t, s.Step, 0, "stayed on step 0 for an empty answer")
 	asserts.Equal(t, len(s.Answers), 0, "empty answer is not stored")
 
@@ -154,7 +170,7 @@ func TestFlow_CancelBailsAndRunsOnCancel(t *testing.T) {
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "cancel"))
 
 	asserts.True(t, canceled, "OnCancel ran")
-	_, ok := bot.conversations.get(key)
+	_, ok := getConvState(bot.conversations, key)
 	asserts.False(t, ok, "state cleared on cancel")
 }
 
@@ -182,11 +198,11 @@ func TestFlow_TTLSlidesOnEachStep(t *testing.T) {
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
-	s1, _ := bot.conversations.get(key)
+	s1, _ := getConvState(bot.conversations, key)
 
 	time.Sleep(2 * time.Millisecond) // ensure the clock advances between steps
 	bot.conversations.advance(ctx, bot, msgFrom("u", "c", "alpha"))
-	s2, _ := bot.conversations.get(key)
+	s2, _ := getConvState(bot.conversations, key)
 
 	asserts.True(t, s2.ExpiresAt.After(s1.ExpiresAt), "TTL slides forward on each successful step")
 }
@@ -202,12 +218,12 @@ func TestFlow_TTLSlidesOnRejectedAnswer(t *testing.T) {
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
-	s1, _ := bot.conversations.get(key)
+	s1, _ := getConvState(bot.conversations, key)
 
 	time.Sleep(2 * time.Millisecond) // ensure the clock advances
 	handled := bot.conversations.advance(ctx, bot, msgFrom("u", "c", "bad"))
 	asserts.True(t, handled, "a rejected answer is consumed")
-	s2, ok := bot.conversations.get(key)
+	s2, ok := getConvState(bot.conversations, key)
 	asserts.True(t, ok, "state survives a rejected answer")
 	asserts.Equal(t, s2.Step, 0, "step does not advance on a rejected answer")
 	asserts.True(t, s2.ExpiresAt.After(s1.ExpiresAt), "TTL slides forward even when the answer is rejected")
@@ -223,7 +239,7 @@ func TestFlow_ExpiredStateTimesOut(t *testing.T) {
 		OnTimeout(func(context.Context, *Bot, *Message) { timedOut = true })
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
-	bot.conversations.set(key, ConversationState{
+	setConvState(bot.conversations, key, ConversationState{
 		FlowID:    "f",
 		ExpiresAt: time.Now().Add(-time.Minute),
 		Answers:   map[string]string{},
@@ -232,7 +248,7 @@ func TestFlow_ExpiredStateTimesOut(t *testing.T) {
 	handled := bot.conversations.advance(ctx, bot, msgFrom("u", "c", "late"))
 	asserts.True(t, handled, "an expired-state message is consumed")
 	asserts.True(t, timedOut, "OnTimeout ran")
-	_, ok := bot.conversations.get(key)
+	_, ok := getConvState(bot.conversations, key)
 	asserts.False(t, ok, "expired state is reaped")
 }
 
@@ -240,7 +256,7 @@ func TestFlow_UnregisteredFlowIDFallsThrough(t *testing.T) {
 	bot := New(SlackBotType, &recordingAdapter{})
 	key := conversationKey(msgFrom("u", "c", ""))
 
-	bot.conversations.set(key, ConversationState{
+	setConvState(bot.conversations, key, ConversationState{
 		FlowID:    "ghost",
 		ExpiresAt: time.Now().Add(time.Hour),
 		Answers:   map[string]string{},
@@ -248,7 +264,7 @@ func TestFlow_UnregisteredFlowIDFallsThrough(t *testing.T) {
 
 	handled := bot.conversations.advance(context.Background(), bot, msgFrom("u", "c", "hello"))
 	asserts.False(t, handled, "state for an unregistered flow falls through to the command table")
-	_, ok := bot.conversations.get(key)
+	_, ok := getConvState(bot.conversations, key)
 	asserts.False(t, ok, "stale state is reaped, not left to panic later")
 }
 
@@ -383,7 +399,7 @@ func TestFlow_OutOfRangeStepFallsThrough(t *testing.T) {
 	key := conversationKey(msgFrom("u", "c", ""))
 
 	// A durable store could load a state whose flow has since lost steps.
-	bot.conversations.set(key, ConversationState{
+	setConvState(bot.conversations, key, ConversationState{
 		FlowID:    "f",
 		Step:      99,
 		ExpiresAt: time.Now().Add(time.Hour),
@@ -392,7 +408,7 @@ func TestFlow_OutOfRangeStepFallsThrough(t *testing.T) {
 
 	handled := bot.conversations.advance(context.Background(), bot, msgFrom("u", "c", "hi"))
 	asserts.False(t, handled, "out-of-range step falls through to the command table")
-	_, ok := bot.conversations.get(key)
+	_, ok := getConvState(bot.conversations, key)
 	asserts.False(t, ok, "corrupt state is reaped, not left to wedge")
 
 	handled2 := bot.conversations.advance(context.Background(), bot, msgFrom("u", "c", "again"))
@@ -448,7 +464,7 @@ func TestFlow_ConcurrentAdvanceNoDeadlock(t *testing.T) {
 	}
 	wg.Wait()
 
-	if s, ok := bot.conversations.get(key); ok {
+	if s, ok := getConvState(bot.conversations, key); ok {
 		asserts.True(t, s.Step >= 0 && s.Step < len(f.steps), "step stays within bounds")
 	}
 }
@@ -462,7 +478,7 @@ func TestFlow_ExpiredStateWithoutOnTimeoutConsumesAndReaps(t *testing.T) {
 	f := NewFlow("f").Ask("a", "a?").OnComplete(noopComplete) // no OnTimeout
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
-	bot.conversations.set(key, ConversationState{
+	setConvState(bot.conversations, key, ConversationState{
 		FlowID:    "f",
 		ExpiresAt: time.Now().Add(-time.Minute),
 		Answers:   map[string]string{},
@@ -470,7 +486,7 @@ func TestFlow_ExpiredStateWithoutOnTimeoutConsumesAndReaps(t *testing.T) {
 
 	handled := bot.conversations.advance(context.Background(), bot, msgFrom("u", "c", "late"))
 	asserts.True(t, handled, "an expired-state message is consumed even with no OnTimeout set")
-	_, ok := bot.conversations.get(key)
+	_, ok := getConvState(bot.conversations, key)
 	asserts.False(t, ok, "expired state is reaped on the nil-OnTimeout default path")
 }
 
@@ -487,7 +503,7 @@ func TestFlow_CancelWithoutOnCancelConsumesAndReaps(t *testing.T) {
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
 	handled := bot.conversations.advance(ctx, bot, msgFrom("u", "c", "cancel"))
 	asserts.True(t, handled, "the cancel word is consumed even with no OnCancel set")
-	_, ok := bot.conversations.get(key)
+	_, ok := getConvState(bot.conversations, key)
 	asserts.False(t, ok, "state is cleared on the nil-OnCancel default path")
 }
 
@@ -515,13 +531,13 @@ func TestFlow_SendFailureStillAdvancesAndLogs(t *testing.T) {
 	asserts.NoError(t, bot.HandleFlow("^f$", f), "register")
 
 	bot.conversations.start(ctx, bot, msgFrom("u", "c", "f"), f)
-	s, ok := bot.conversations.get(key)
+	s, ok := getConvState(bot.conversations, key)
 	asserts.True(t, ok, "flow starts despite a failed first-prompt send")
 	asserts.Equal(t, s.Step, 0, "on step 0 after start")
 
 	handled := bot.conversations.advance(ctx, bot, msgFrom("u", "c", "alpha"))
 	asserts.True(t, handled, "answer is consumed even though the prompt send fails")
-	s2, _ := bot.conversations.get(key)
+	s2, _ := getConvState(bot.conversations, key)
 	asserts.Equal(t, s2.Step, 1, "flow advances to step 1 despite the send error")
 
 	asserts.True(t, strings.Contains(buf.String(), "failed to send flow prompt"),