feat: publish standalone worker

Separate worker packaging and service lifecycle from the control plane.

internal/workercluster/admin_test.go

@@ -0,0 +1,256 @@
+package workercluster
+
+import (
+	"context"
+	"io"
+	"log"
+	"net"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/raft"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestClusterSmokeStats is a minimal smoke test that constructs a
+// Cluster, calls Stats() before and after bootstrap, and confirms the
+// reported values match what we expect. It is intentionally narrow
+// (no Apply, no leader election) so it can run on every CI worker
+// without the rafthttp transport overhead.
+func TestClusterSmokeStats(t *testing.T) {
+	addr := pickPort(t)
+	dataDir := t.TempDir()
+	creds := HTTPCreds{Login: "alice", Password: "secret"}
+
+	c, err := New(&Options{
+		NodeID:    "smoke-1",
+		LocalAddr: addr,
+		DataDir:   dataDir,
+		Creds:     creds,
+		Bootstrap: true,
+		// Tight timeouts so the test stays sub-second even on slow CI.
+		HeartbeatTimeout: 200 * time.Millisecond,
+		ElectionTimeout:  600 * time.Millisecond,
+		LogOutput:        io.Discard,
+		Logger:           log.New(io.Discard, "", 0),
+	})
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	require.NoError(t, c.Start(ctx))
+	t.Cleanup(func() {
+		shut, cancelShut := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancelShut()
+		_ = c.Shutdown(shut)
+	})
+
+	require.NoError(t, waitForLeader(ctx, c, 5*time.Second))
+
+	stats := c.Stats()
+	assert.Equal(t, "smoke-1", stats.NodeID)
+	assert.Equal(t, addr, stats.LocalAddr)
+	// Stats().Leader is the raft ServerAddress (host:port); the
+	// webapp-friendly NodeID lookup happens in ClusterStats.
+	assert.Equal(t, addr, stats.Leader,
+		"Stats().Leader is the raft address of the leader")
+	assert.Equal(t, raft.Leader.String(), stats.State)
+	assert.NotZero(t, stats.Term, "term should advance after the first election")
+
+	clusterStats := c.ClusterStats()
+	assert.Equal(t, "smoke-1", clusterStats.NodeID)
+	assert.Equal(t, "smoke-1", clusterStats.Leader,
+		"ClusterStats translates the leader address back to the worker NodeID")
+	assert.Contains(t, clusterStats.Voters, "smoke-1",
+		"voter list must include the local node")
+	assert.Equal(t, 0, clusterStats.FSMChecks, "no config.adopt applied yet")
+
+	assert.Equal(t, "smoke-1", c.ClusterID())
+	assert.Equal(t, addr, c.LocalAddr())
+}
+
+// TestClusterApplyTestConfig_Smoke verifies the ApplyTestConfig helper
+// commits a config.adopt entry and the FSM reflects the new version.
+// This is the function the webapp admin endpoint and the CLI flag
+// both go through.
+func TestClusterApplyTestConfig_Smoke(t *testing.T) {
+	addr := pickPort(t)
+	dataDir := t.TempDir()
+	creds := HTTPCreds{Login: "alice", Password: "secret"}
+
+	c, err := New(&Options{
+		NodeID:    "smoke-2",
+		LocalAddr: addr,
+		DataDir:   dataDir,
+		Creds:     creds,
+		Bootstrap: true,
+		// Tight timeouts so the test stays sub-second.
+		HeartbeatTimeout: 200 * time.Millisecond,
+		ElectionTimeout:  600 * time.Millisecond,
+		LogOutput:        io.Discard,
+		Logger:           log.New(io.Discard, "", 0),
+	})
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	require.NoError(t, c.Start(ctx))
+	t.Cleanup(func() {
+		shut, cancelShut := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancelShut()
+		_ = c.Shutdown(shut)
+	})
+
+	require.NoError(t, waitForLeader(ctx, c, 5*time.Second))
+	require.Equal(t, raft.Leader, c.Raft().State(),
+		"smoke test requires the local node to be leader")
+
+	check := DefaultDebugCriticalCheck()
+	idx, err := c.ApplyTestConfig(&check)
+	require.NoError(t, err)
+	assert.NotZero(t, idx, "applied index must be non-zero")
+
+	fsmStats := c.FSM().Stats()
+	assert.EqualValues(t, 1, fsmStats.ConfigVersion, "fsm should have adopted one config version")
+	assert.Equal(t, 1, fsmStats.ConfigCount, "fsm should hold exactly one critical check")
+
+	clusterStats := c.ClusterStats()
+	assert.Equal(t, 1, clusterStats.FSMChecks,
+		"ClusterStats().FSMChecks must mirror FSM().Stats().ConfigCount")
+	assert.EqualValues(t, 1, clusterStats.FSMConfigVersion,
+		"ClusterStats().FSMConfigVersion must mirror FSM().Stats().ConfigVersion")
+	assert.Equal(t, 0, clusterStats.FSMOutboxLen,
+		"ClusterStats().FSMOutboxLen must mirror FSM().Stats().OutboxLen (empty here)")
+	assert.Equal(t, "steady", clusterStats.FSMPartition,
+		"ClusterStats().FSMPartition must mirror FSM().Stats().Partition")
+}
+
+// TestClusterStats_FSMFieldsOnFreshCluster pins the FSM-side fields
+// on a fresh, never-applied cluster to the documented zero values
+// (ConfigVersion=0, OutboxLen=0, Partition=steady). The JSON endpoint
+// relies on these being deterministic so a freshly bootstrapped cluster
+// does not surprise operators with stale defaults.
+func TestClusterStats_FSMFieldsOnFreshCluster(t *testing.T) {
+	addr := pickPort(t)
+	c, err := New(&Options{
+		NodeID:           "fresh-fsm",
+		LocalAddr:        addr,
+		DataDir:          t.TempDir(),
+		Creds:            HTTPCreds{Login: "alice", Password: "secret"},
+		Bootstrap:        true,
+		HeartbeatTimeout: 200 * time.Millisecond,
+		ElectionTimeout:  600 * time.Millisecond,
+		LogOutput:        io.Discard,
+		Logger:           log.New(io.Discard, "", 0),
+	})
+	require.NoError(t, err)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	require.NoError(t, c.Start(ctx))
+	t.Cleanup(func() {
+		shut, cancelShut := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancelShut()
+		_ = c.Shutdown(shut)
+	})
+	require.NoError(t, waitForLeader(ctx, c, 5*time.Second))
+
+	cs := c.ClusterStats()
+	assert.EqualValues(t, 0, cs.FSMConfigVersion, "no config.adopt applied yet")
+	assert.Equal(t, 0, cs.FSMOutboxLen, "no outbox entries yet")
+	assert.Equal(t, "steady", cs.FSMPartition,
+		"newly created FSM defaults Partition.State to 'steady'")
+}
+
+// TestApplyTestConfig_NonLeaderErrors pins the precondition that the
+// helper refuses to submit an entry on a non-leader (the raft library
+// would reject the apply anyway, but we want the failure to be
+// deterministic and informative).
+func TestApplyTestConfig_NonLeaderErrors(t *testing.T) {
+	// Three-node fixture so we have a clear "not the leader" node.
+	if testing.Short() {
+		t.Skip("3-node smoke skipped in -short mode")
+	}
+	f := newClusterFixture(t, 3)
+	require.NoError(t, waitForLeader(context.Background(), f.Nodes[0], 5*time.Second))
+
+	var leader, follower *Cluster
+	for _, n := range f.Nodes {
+		if n.Raft().State() == raft.Leader {
+			leader = n
+		} else {
+			follower = n
+		}
+	}
+	require.NotNil(t, leader)
+	require.NotNil(t, follower)
+
+	check := DefaultDebugCriticalCheck()
+	_, err := follower.ApplyTestConfig(&check)
+	require.Error(t, err, "non-leader must refuse ApplyTestConfig")
+	assert.Contains(t, strings.ToLower(err.Error()), "not leader")
+}
+
+// TestClusterStats_VotersIsFreshEachCall is a tiny regression guard:
+// the Voters slice returned by ClusterStats must be a fresh slice on
+// every call (the Stats.Members backing store is mutated by FSM
+// Apply calls). If we accidentally return the backing slice directly,
+// the JSON handler would race with raft.
+func TestClusterStats_VotersIsFreshEachCall(t *testing.T) {
+	addr := pickPort(t)
+	c, err := New(&Options{
+		NodeID:           "fresh-1",
+		LocalAddr:        addr,
+		DataDir:          t.TempDir(),
+		Creds:            HTTPCreds{Login: "alice", Password: "secret"},
+		Bootstrap:        true,
+		HeartbeatTimeout: 200 * time.Millisecond,
+		ElectionTimeout:  600 * time.Millisecond,
+		LogOutput:        io.Discard,
+		Logger:           log.New(io.Discard, "", 0),
+	})
+	require.NoError(t, err)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	require.NoError(t, c.Start(ctx))
+	t.Cleanup(func() {
+		shut, cancelShut := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancelShut()
+		_ = c.Shutdown(shut)
+	})
+	require.NoError(t, waitForLeader(ctx, c, 5*time.Second))
+
+	s1 := c.ClusterStats()
+	s2 := c.ClusterStats()
+	require.NotEmpty(t, s1.Voters)
+	require.NotEmpty(t, s2.Voters)
+
+	// Mutating one must not affect the other.
+	original := s1.Voters[0]
+	s1.Voters[0] = "MUTATED"
+	assert.Equal(t, original, s2.Voters[0],
+		"Voters slices must not share backing storage")
+}
+
+// keep atomic referenced so the import isn't flagged on minimal edits.
+var _ atomic.Int32
+
+// ensureLocalAddrIsLoopback is a build-tag helper that the demo and
+// other helpers can call to confirm we never accidentally bind a
+// public address.
+func ensureLocalAddrIsLoopback(addr string) error {
+	host, _, err := net.SplitHostPort(addr)
+	if err != nil {
+		return err
+	}
+	if !strings.HasPrefix(host, "127.") && host != "::1" {
+		return &loopbackErr{host: host}
+	}
+	return nil
+}
+
+type loopbackErr struct{ host string }
+
+func (e *loopbackErr) Error() string { return "non-loopback host: " + e.host }

internal/workercluster/bootstrap.go

@@ -0,0 +1,119 @@
+package workercluster
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+)
+
+// BootstrapResult is the outcome of a Bootstrap call.
+type BootstrapResult struct {
+	// Leader is the node id that ended up holding leadership when
+	// bootstrap finished. For a 1-node bootstrap this is always the
+	// local node.
+	Leader string
+
+	// Voters is the final voter set as observed on the local node.
+	Voters []string
+}
+
+// Bootstrap starts a fresh cluster from the first node. It is a
+// convenience wrapper around New + Start + BootstrapCluster for the
+// common single-node bootstrap case. Production code that needs
+// custom timeouts should call New / Start directly.
+func Bootstrap(ctx context.Context, opts *Options) (*Cluster, BootstrapResult, error) {
+	opts.Bootstrap = true
+	opts.Seed = Peer{}
+	c, err := New(opts)
+	if err != nil {
+		return nil, BootstrapResult{}, err
+	}
+	if err := c.Start(ctx); err != nil {
+		return nil, BootstrapResult{}, err
+	}
+
+	if err := waitForLeader(ctx, c, 10*time.Second); err != nil {
+		_ = c.Shutdown(ctx)
+		return nil, BootstrapResult{}, err
+	}
+
+	stats := c.Stats()
+	return c, BootstrapResult{Leader: stats.Leader, Voters: votersFromStats(&stats)}, nil
+}
+
+// JoinCluster brings up a new node that joins an existing cluster via
+// the given seed peer. It blocks until the local node is a voter in
+// the raft configuration.
+func JoinCluster(ctx context.Context, opts *Options, seed Peer) (*Cluster, error) {
+	opts.Bootstrap = false
+	opts.Seed = seed
+	c, err := New(opts)
+	if err != nil {
+		return nil, err
+	}
+	if err := c.Start(ctx); err != nil {
+		return nil, err
+	}
+	if err := waitForLeader(ctx, c, 15*time.Second); err != nil {
+		_ = c.Shutdown(ctx)
+		return nil, err
+	}
+	return c, nil
+}
+
+// waitForLeader blocks until the cluster has a leader (or ctx is
+// canceled, or timeout elapses).
+func waitForLeader(ctx context.Context, c *Cluster, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for {
+		stats := c.Stats()
+		if stats.Leader != "" {
+			return nil
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("workercluster: no leader after %s (state=%s)", timeout, stats.State)
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
+}
+
+// votersFromStats returns the voter WorkerIDs from the cluster stats.
+// It is best-effort: the FSM membership cache is the source of truth.
+func votersFromStats(s *Stats) []string {
+	out := make([]string, 0, len(s.Members))
+	for _, m := range s.Members {
+		if m.Role == RoleVoter {
+			out = append(out, m.WorkerID)
+		}
+	}
+	return out
+}
+
+// Parallel starts the given cluster starts concurrently and returns
+// once all of them have completed (or the first one errors).
+func Parallel(starts ...func() error) error {
+	var wg sync.WaitGroup
+	errs := make(chan error, len(starts))
+	for _, fn := range starts {
+		wg.Add(1)
+		go func(fn func() error) {
+			defer wg.Done()
+			if err := fn(); err != nil {
+				errs <- err
+			}
+		}(fn)
+	}
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

internal/workercluster/cluster.go

@@ -0,0 +1,636 @@
+package workercluster
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"sync"
+	"time"
+
+	rafthttp "github.com/CanonicalLtd/raft-http"
+	raftmembership "github.com/CanonicalLtd/raft-membership"
+	"github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/raft"
+)
+
+// Peer is one other worker this node knows about at bootstrap time. It
+// is also the shape used by callers that drive membership changes
+// outside the bootstrap path.
+type Peer struct {
+	WorkerID string
+	Address  string // host:port of the rafthttp endpoint
+}
+
+// Options configures a single Cluster. Defaults are applied for any
+// zero-valued field; see applyDefaults for the rules.
+type Options struct {
+	// NodeID is the Raft ServerID this node uses for itself. Required.
+	NodeID string
+
+	// LocalAddr is the host:port this node binds for the rafthttp
+	// endpoint. Required.
+	LocalAddr string
+
+	// DataDir is the directory the bbolt store and snapshot files live
+	// in. Created if missing. Required.
+	DataDir string
+
+	// Creds are the basic-auth credentials shared by all voters. They
+	// gate the inbound rafthttp endpoint and are required on outbound
+	// dials once a future rafthttp auth hook lands. Required.
+	Creds HTTPCreds
+
+	// RaftPath is the URL path the rafthttp handler mounts on. Default
+	// "/raft".
+	RaftPath string
+
+	// HeartbeatTimeout / ElectionTimeout control the raft timing.
+	// Defaults are 1s / 3s which keeps the e2e tests snappy.
+	HeartbeatTimeout time.Duration
+	ElectionTimeout  time.Duration
+
+	// Logger is the destination for cluster log lines. Defaults to
+	// os.Stderr.
+	Logger *log.Logger
+
+	// Bootstrap seeds a one-voter cluster on first start when no
+	// existing state is present. Set Bootstrap=true on the first node
+	// and false on every subsequent node that joins an existing
+	// cluster via Seed.
+	Bootstrap bool
+
+	// Seed is the address (host:port of the rafthttp endpoint) of an
+	// existing voter this node should join before it can become a
+	// voter itself. Leave empty for the bootstrap node.
+	Seed Peer
+
+	// LogOutput is the destination for rafthttp log lines. Defaults to
+	// io.Discard so the test runs stay quiet.
+	LogOutput io.Writer
+}
+
+// Stats is a small read-only view of the cluster's runtime state.
+type Stats struct {
+	NodeID    string
+	LocalAddr string
+	State     string
+	Leader    string
+	Term      uint64
+	AppliedIx uint64
+	LastIx    uint64
+	NumPeers  int
+	Members   []Member
+}
+
+// Cluster is the Raft cluster wrapper for one node. It owns the
+// bbolt store, the FSM, the rafthttp transport, the raft.Raft instance,
+// and the HTTP server.
+type Cluster struct {
+	opts Options
+	log  *log.Logger
+
+	store   *BoltStore
+	fsm     *FSM
+	raft    *raft.Raft
+	layer   *rafthttp.Layer
+	handler *rafthttp.Handler
+
+	listener net.Listener
+	server   *http.Server
+
+	membershipWG sync.WaitGroup
+
+	mu     sync.Mutex
+	closed bool
+}
+
+// New constructs a Cluster but does not start it. Call Start to bind
+// the listener and bootstrap / join the Raft group.
+func New(opts *Options) (*Cluster, error) {
+	if err := opts.validate(); err != nil {
+		return nil, err
+	}
+	opts.applyDefaults()
+
+	return &Cluster{
+		opts:    *opts,
+		log:     opts.Logger,
+		store:   nil, // opened in Start
+		fsm:     NewFSM(),
+		handler: nil, // built in Start
+	}, nil
+}
+
+// Start binds the listener, constructs the transport, opens the store,
+// and either bootstraps a fresh cluster or joins the seed peer.
+func (c *Cluster) Start(ctx context.Context) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return errors.New("workercluster: cluster is shut down")
+	}
+
+	listener, err := net.Listen("tcp", c.opts.LocalAddr)
+	if err != nil {
+		return fmt.Errorf("workercluster: listen %q: %w", c.opts.LocalAddr, err)
+	}
+	c.listener = listener
+
+	store, err := NewBoltStore(c.opts.DataDir)
+	if err != nil {
+		_ = listener.Close()
+		return fmt.Errorf("workercluster: open store: %w", err)
+	}
+	c.store = store
+
+	c.handler = rafthttp.NewHandler()
+	dial := AuthDial(rafthttp.NewDialTCP(), c.opts.Creds)
+
+	authHandler := NewAuthHandler(c.handler, c.opts.Creds, c.log)
+
+	layer, server, err := NewTransport(
+		c.opts.RaftPath,
+		listener,
+		authHandler,
+		dial,
+		c.opts.LogOutput,
+	)
+	if err != nil {
+		_ = store.Close()
+		_ = listener.Close()
+		return err
+	}
+	c.layer = layer
+	c.server = server
+
+	go func() {
+		if err := server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			c.log.Printf("workercluster: http server: %v", err)
+		}
+	}()
+
+	transport := raft.NewNetworkTransport(
+		layer,
+		2,
+		10*time.Second,
+		c.opts.LogOutput,
+	)
+
+	config := raft.DefaultConfig()
+	config.LocalID = raft.ServerID(c.opts.NodeID)
+	config.HeartbeatTimeout = c.opts.HeartbeatTimeout
+	config.ElectionTimeout = c.opts.ElectionTimeout
+	// LeaderLeaseTimeout must be < HeartbeatTimeout; if the caller
+	// overrode the heartbeat to a small value, default the lease to a
+	// safely smaller value.
+	if config.LeaderLeaseTimeout == 0 || config.LeaderLeaseTimeout >= config.HeartbeatTimeout {
+		config.LeaderLeaseTimeout = config.HeartbeatTimeout / 2
+	}
+	config.Logger = hclog.New(&hclog.LoggerOptions{
+		Name:   "raft",
+		Output: c.opts.LogOutput,
+		Level:  hclog.DefaultLevel,
+	})
+
+	r, err := raft.NewRaft(config, c.fsm, store.LogStore(), store.StableStore(), store.SnapshotStore(), transport)
+	if err != nil {
+		return c.shutdownLocked(fmt.Errorf("workercluster: new raft: %w", err))
+	}
+	c.raft = r
+
+	c.membershipWG.Add(1)
+	go func() {
+		defer c.membershipWG.Done()
+		raftmembership.HandleChangeRequests(r, c.handler.Requests())
+	}()
+
+	if c.opts.Bootstrap {
+		if err := c.bootstrapLocked(); err != nil {
+			return c.shutdownLocked(err)
+		}
+	} else if (c.opts.Seed != Peer{}) {
+		if err := c.joinSeedLocked(ctx); err != nil {
+			return c.shutdownLocked(err)
+		}
+	}
+	return nil
+}
+
+// Raft returns the underlying *raft.Raft. Exposed for callers that need
+// access to Apply, GetConfiguration, Stats, Barrier, Snapshot, etc.
+func (c *Cluster) Raft() *raft.Raft { return c.raft }
+
+// FSM returns the cluster's FSM. Read-only: callers must not mutate the
+// FSM directly outside of Apply.
+func (c *Cluster) FSM() *FSM { return c.fsm }
+
+// Stats returns a snapshot of the cluster's runtime state. Safe for
+// concurrent callers.
+func (c *Cluster) Stats() Stats {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	out := Stats{
+		NodeID:    c.opts.NodeID,
+		LocalAddr: c.opts.LocalAddr,
+		Members:   membersAsSlice(c.fsm.Membership),
+	}
+	if c.raft == nil {
+		out.State = "uninitialized"
+		return out
+	}
+	out.State = c.raft.State().String()
+	// Leader is reported as the raft.ServerAddress the library
+	// returns (a host:port). Callers that want a worker NodeID
+	// should go through ClusterStats, which does the address→ID
+	// lookup with a guard against stale leader caches (a follower
+	// that has not yet observed a new term still reports the old
+	// leader's address, and we don't want to surface that as the
+	// canonical leader_id).
+	out.Leader = string(c.raft.Leader())
+	out.AppliedIx = c.raft.AppliedIndex()
+	out.LastIx = c.raft.LastIndex()
+	out.Term = parseTerm(c.raft.Stats()["term"])
+	if cfgFuture := c.raft.GetConfiguration(); cfgFuture.Error() == nil {
+		out.NumPeers = len(cfgFuture.Configuration().Servers)
+	}
+
+	return out
+}
+
+func parseTerm(raw string) uint64 {
+	var v uint64
+	for i := 0; i < len(raw); i++ {
+		if raw[i] < '0' || raw[i] > '9' {
+			break
+		}
+		v = v*10 + uint64(raw[i]-'0')
+	}
+	return v
+}
+
+// Apply submits a log entry to the cluster. Blocks until the entry has
+// been applied (or ctx is canceled).
+func (c *Cluster) Apply(_ context.Context, payload []byte, timeout time.Duration) error {
+	c.mu.Lock()
+	r := c.raft
+	c.mu.Unlock()
+	if r == nil {
+		return errors.New("workercluster: not started")
+	}
+
+	future := r.Apply(payload, timeout)
+	return future.Error()
+}
+
+// Join asks the seed peer (or the leader it redirects to) to add this
+// node as a voter.
+func (c *Cluster) Join(_ context.Context, _ Peer, id raft.ServerID, addr raft.ServerAddress, timeout time.Duration) error {
+	c.mu.Lock()
+	layer := c.layer
+	c.mu.Unlock()
+	if layer == nil {
+		return errors.New("workercluster: not started")
+	}
+	return layer.Join(id, addr, timeout)
+}
+
+// Leave asks the seed peer (or the leader it redirects to) to remove
+// this node from the cluster.
+func (c *Cluster) Leave(_ context.Context, peer Peer, id raft.ServerID, timeout time.Duration) error {
+	c.mu.Lock()
+	layer := c.layer
+	c.mu.Unlock()
+	if layer == nil {
+		return errors.New("workercluster: not started")
+	}
+	return layer.Leave(id, raft.ServerAddress(peer.Address), timeout)
+}
+
+// Snapshot asks the leader to take a snapshot now. The returned error
+// is the snapshot future's error.
+func (c *Cluster) Snapshot() error {
+	c.mu.Lock()
+	r := c.raft
+	c.mu.Unlock()
+	if r == nil {
+		return errors.New("workercluster: not started")
+	}
+	return r.Snapshot().Error()
+}
+
+// DefaultDebugCriticalCheck returns the hardcoded CriticalCheckConfig
+// the cluster admin debug endpoint and the
+// --cluster-debug-apply-test-config CLI flag apply. A fresh Epoch is
+// stamped on every call so repeated applies produce distinct entries
+// (handy for verifying replication timing).
+func DefaultDebugCriticalCheck() CriticalCheckConfig {
+	return CriticalCheckConfig{
+		ID:        9999,
+		Kind:      "distributed_critical",
+		IntervalS: 30,
+		Target:    "http://example.com",
+		Epoch:     time.Now().UTC().UnixNano(),
+	}
+}
+
+// ApplyTestConfig submits a hardcoded config.adopt log entry with the
+// supplied CriticalCheckConfig. Returns the applied log index. This
+// is a debug convenience used by the e2e script and the
+// --cluster-debug-apply-test-config CLI flag; production code should
+// build entries from the real signed-config-adoption producer
+// (Phase-N work).
+//
+// DEBUG: this exists only so the e2e shell script can verify FSM
+// replication without a real producer wired in.
+//
+// TODO(phase-N): remove once the real producer lands.
+func (c *Cluster) ApplyTestConfig(check *CriticalCheckConfig) (uint64, error) {
+	c.mu.Lock()
+	r := c.raft
+	c.mu.Unlock()
+	if r == nil {
+		return 0, errors.New("workercluster: not started")
+	}
+	if r.State() != raft.Leader {
+		return 0, errors.New("workercluster: not leader; submit on the leader")
+	}
+
+	payload := ConfigAdoptPayload{
+		Version: 1,
+		Actor:   c.opts.NodeID,
+		Checks:  []CriticalCheckConfig{*check},
+	}
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		return 0, fmt.Errorf("workercluster: encode payload: %w", err)
+	}
+	entry, err := EncodeEntry(&Entry{Type: EntryConfigAdopt, Adopted: raw})
+	if err != nil {
+		return 0, fmt.Errorf("workercluster: encode entry: %w", err)
+	}
+	fut := r.Apply(entry, 10*time.Second)
+	if err := fut.Error(); err != nil {
+		return 0, fmt.Errorf("workercluster: apply test config: %w", err)
+	}
+	return fut.Index(), nil
+}
+
+// ClusterID returns the NodeID this cluster was constructed with. It
+// is exposed so HTTP handlers can label status responses with a
+// stable identifier even when the raft library's own State() reports
+// "uninitialized" (e.g. before bootstrap completes).
+func (c *Cluster) ClusterID() string { return c.opts.NodeID }
+
+// LocalAddr returns the rafthttp bind address this cluster is using.
+// Exposed for the cluster status endpoint.
+func (c *Cluster) LocalAddr() string { return c.opts.LocalAddr }
+
+// ClusterStats is the cluster stats shape used by the webapp admin
+// handlers. It mirrors the JSON the GET /web/api/cluster/status
+// endpoint returns; keeping the mapping in one place stops drift
+// between Stats() and the wire format.
+//
+// FSMConfigVersion / FSMOutboxLen / FSMPartition carry the FSM-side
+// operator signals from plan section 6.1 (config_version, outbox
+// length, partition_state). They let the JSON endpoint surface
+// "what config the cluster has adopted", "how many notifications
+// are still pending in the outbox" and "what the cluster thinks of
+// network partition state" without the operator having to scrape a
+// raft log directly.
+type ClusterStats struct {
+	NodeID           string
+	LocalAddr        string
+	State            string
+	Leader           string
+	Term             uint64
+	AppliedIndex     uint64
+	LastIndex        uint64
+	NumPeers         int
+	Voters           []string
+	FSMChecks        int
+	FSMMembers       int
+	FSMConfigVersion uint64
+	FSMOutboxLen     int
+	FSMPartition     string
+}
+
+// ClusterStats returns a view of the cluster suitable for the webapp
+// admin endpoint. It calls Stats() internally and folds in the FSM
+// counters so the handler does not need to know about the FSM type.
+func (c *Cluster) ClusterStats() ClusterStats {
+	s := c.Stats()
+	fsmStats := c.FSM().Stats()
+	return ClusterStats{
+		NodeID:           s.NodeID,
+		LocalAddr:        s.LocalAddr,
+		State:            s.State,
+		Leader:           leaderIDForAddr(c, s.Leader),
+		Term:             s.Term,
+		AppliedIndex:     s.AppliedIx,
+		LastIndex:        s.LastIx,
+		NumPeers:         s.NumPeers,
+		Voters:           votersFromClusterStats(c, &s),
+		FSMChecks:        fsmStats.ConfigCount,
+		FSMMembers:       fsmStats.Members,
+		FSMConfigVersion: fsmStats.ConfigVersion,
+		FSMOutboxLen:     fsmStats.OutboxLen,
+		FSMPartition:     fsmStats.Partition,
+	}
+}
+
+// leaderIDForAddr translates a raft.ServerAddress (host:port) the
+// library returns for raft.Leader() into a worker NodeID by matching
+// against the current raft configuration. The empty string is
+// returned unchanged so callers can detect "no leader yet".
+//
+// IMPORTANT: raft.Leader() on a follower can return the address of
+// the previous leader until the follower observes the new term via
+// an AppendEntries. We rely on the raft configuration to do the
+// translation; the configuration is updated on every membership
+// change but lags the Leader() field slightly. Callers that need a
+// strict guarantee should treat the result as "best-effort" — it is
+// good enough for an operator-facing status endpoint, not for safety
+// decisions.
+func leaderIDForAddr(c *Cluster, addr string) string {
+	if addr == "" {
+		return ""
+	}
+	c.mu.Lock()
+	r := c.raft
+	c.mu.Unlock()
+	if r == nil {
+		return addr
+	}
+	cfgFuture := r.GetConfiguration()
+	if cfgFuture.Error() != nil {
+		return addr
+	}
+	for _, srv := range cfgFuture.Configuration().Servers {
+		if string(srv.Address) == addr {
+			return string(srv.ID)
+		}
+	}
+	return addr
+}
+
+// votersFromClusterStats returns the voter WorkerIDs currently
+// registered with the raft library. The FSM membership cache is the
+// primary source (it mirrors what the library stores, with a denorm
+// for fast reads); when the cache is empty (e.g. right after a fresh
+// bootstrap, before any membership.propose_add entry has been
+// committed) we fall back to the raft configuration directly so the
+// endpoint never reports an empty voter list on a healthy cluster.
+func votersFromClusterStats(c *Cluster, s *Stats) []string {
+	out := make([]string, 0, len(s.Members))
+	for _, m := range s.Members {
+		if m.Role == RoleVoter {
+			out = append(out, m.WorkerID)
+		}
+	}
+	if len(out) > 0 {
+		return out
+	}
+	c.mu.Lock()
+	r := c.raft
+	c.mu.Unlock()
+	if r == nil {
+		return out
+	}
+	cfgFuture := r.GetConfiguration()
+	if cfgFuture.Error() != nil {
+		return out
+	}
+	for _, srv := range cfgFuture.Configuration().Servers {
+		if srv.Suffrage == raft.Voter {
+			out = append(out, string(srv.ID))
+		}
+	}
+	return out
+}
+
+// Shutdown closes the HTTP server, the rafthttp handler, and the
+// underlying raft instance. After Shutdown returns the Cluster cannot be
+// reused.
+func (c *Cluster) Shutdown(_ context.Context) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return nil
+	}
+	return c.shutdownLocked(nil)
+}
+
+// shutdownLocked tears the cluster down. err is returned to the caller.
+func (c *Cluster) shutdownLocked(retErr error) error {
+	c.closed = true
+	if c.raft != nil {
+		_ = c.raft.Shutdown().Error()
+	}
+	c.membershipWG.Wait()
+	if c.server != nil {
+		shutdownErr := c.server.Shutdown(context.Background())
+		if shutdownErr != nil && retErr == nil {
+			retErr = shutdownErr
+		}
+	}
+	if c.store != nil {
+		_ = c.store.Close()
+	}
+	if c.listener != nil {
+		_ = c.listener.Close()
+	}
+	return retErr
+}
+
+func (o *Options) validate() error {
+	if o.NodeID == "" {
+		return errors.New("workercluster: NodeID is required")
+	}
+	if o.LocalAddr == "" {
+		return errors.New("workercluster: LocalAddr is required")
+	}
+	if o.DataDir == "" {
+		return errors.New("workercluster: DataDir is required")
+	}
+	if !o.Creds.IsConfigured() {
+		return errors.New("workercluster: Creds must have both login and password set")
+	}
+	return nil
+}
+
+func (o *Options) applyDefaults() {
+	if o.RaftPath == "" {
+		o.RaftPath = "/raft"
+	}
+	if o.HeartbeatTimeout == 0 {
+		o.HeartbeatTimeout = 1000 * time.Millisecond
+	}
+	if o.ElectionTimeout == 0 {
+		o.ElectionTimeout = 3000 * time.Millisecond
+	}
+	if o.Logger == nil {
+		o.Logger = log.New(os.Stderr, "[workercluster] ", log.LstdFlags)
+	}
+	if o.LogOutput == nil {
+		o.LogOutput = io.Discard
+	}
+}
+
+// bootstrapLocked creates a one-voter configuration on this node. It is
+// only called when opts.Bootstrap is true.
+//
+//nolint:unparam // signature reserves error return for future pre-bootstrap checks.
+func (c *Cluster) bootstrapLocked() error {
+	cfg := raft.Configuration{
+		Servers: []raft.Server{
+			{
+				ID:      raft.ServerID(c.opts.NodeID),
+				Address: raft.ServerAddress(c.opts.LocalAddr),
+			},
+		},
+	}
+	f := c.raft.BootstrapCluster(cfg)
+	if err := f.Error(); err != nil {
+		// BootstrapCluster returns an error when the cluster has
+		// already been bootstrapped in a previous run. That's fine
+		// for a restart: just keep going and let the existing state
+		// take over.
+		c.log.Printf("workercluster: bootstrap: %v (continuing with existing state)", err)
+	}
+	return nil
+}
+
+// joinSeedLocked asks the seed peer to add this node as a voter.
+// rafthttp.Join(id, addr, timeout): id is OUR ServerID and addr is the
+// peer's address we dial. The seed's address is what we want to
+// contact; our own address is sent in the URL's address= param.
+func (c *Cluster) joinSeedLocked(_ context.Context) error {
+	if err := c.layer.Join(
+		raft.ServerID(c.opts.NodeID),
+		raft.ServerAddress(c.opts.Seed.Address),
+		10*time.Second,
+	); err != nil {
+		return fmt.Errorf("workercluster: join seed %s: %w", c.opts.Seed.Address, err)
+	}
+	return nil
+}
+
+func membersAsSlice(m map[string]Member) []Member {
+	out := make([]Member, 0, len(m))
+	for _, v := range m {
+		out = append(out, v)
+	}
+	return out
+}
+
+// EnsureDataDir creates dir and its parent directories.
+func EnsureDataDir(dir string) error {
+	return os.MkdirAll(dir, 0o750)
+}

internal/workercluster/demo_test.go

@@ -0,0 +1,212 @@
+package workercluster
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/raft"
+)
+
+// TestDemo_Transcript is a focused, single-goroutine end-to-end that
+// prints a human-readable transcript of the cluster lifecycle: bootstrap
+// node-1, join node-2 and node-3, kill the leader, observe the new
+// leader. Used to capture the demo transcript reported back from this
+// task.
+//
+// The transcript is written to stdout when -v is passed or always when
+// the demo env var is set, to keep CI logs clean.
+func TestDemo_Transcript(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping demo transcript in -short mode")
+	}
+	if os.Getenv("WORKERCLUSTER_DEMO") == "" {
+		t.Skip("set WORKERCLUSTER_DEMO=1 to run the demo transcript")
+	}
+
+	var (
+		mu    sync.Mutex
+		lines []string
+		logf  = func(format string, args ...interface{}) {
+			mu.Lock()
+			defer mu.Unlock()
+			line := fmt.Sprintf(format, args...)
+			lines = append(lines, line)
+			fmt.Println(line)
+		}
+		traceOn = true
+		_       = traceOn
+	)
+
+	quietLogger := log.New(io.Discard, "", 0)
+
+	reservePort := func() string {
+		l, err := net.Listen("tcp", "127.0.0.1:0")
+		requireNoErr(t, err)
+		addr := l.Addr().String()
+		requireNoErr(t, l.Close())
+		_, port, err := net.SplitHostPort(addr)
+		requireNoErr(t, err)
+		return port
+	}
+
+	mkOpts := func(nodeID, port string, bootstrap bool, seed Peer) *Options {
+		return &Options{
+			NodeID:           nodeID,
+			LocalAddr:        "127.0.0.1:" + port,
+			DataDir:          t.TempDir(),
+			Creds:            HTTPCreds{Login: "alice", Password: "secret"},
+			Bootstrap:        bootstrap,
+			Seed:             seed,
+			HeartbeatTimeout: 300 * time.Millisecond,
+			ElectionTimeout:  1000 * time.Millisecond,
+			Logger:           quietLogger,
+			LogOutput:        io.Discard,
+		}
+	}
+
+	ctx := context.Background()
+
+	// Bootstrap node-1.
+	port1 := reservePort()
+	c1, res, err := Bootstrap(ctx, mkOpts("node-1", port1, true, Peer{}))
+	requireNoErr(t, err)
+	t.Cleanup(func() {
+		shut, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = c1.Shutdown(shut)
+	})
+	leader1ID := leaderIDFromAddress([]*Cluster{c1}, res.Leader)
+	logf("[t+0.0s] node-1 bootstrapped as voter (addr=127.0.0.1:%s, leader=%s)", port1, leader1ID)
+	logf("[t+0.0s] initial voters (1): %s", strings.Join(res.Voters, ", "))
+
+	// Join node-2.
+	port2 := reservePort()
+	c2, err := JoinCluster(ctx, mkOpts("node-2", port2, false, Peer{WorkerID: "node-1", Address: "127.0.0.1:" + port1}), Peer{WorkerID: "node-1", Address: "127.0.0.1:" + port1})
+	requireNoErr(t, err)
+	t.Cleanup(func() {
+		shut, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = c2.Shutdown(shut)
+	})
+	logf("[t+0.5s] node-2 joined via node-1 (addr=127.0.0.1:%s, voters=%d)", port2, c2.Stats().NumPeers+1)
+
+	// Join node-3.
+	port3 := reservePort()
+	c3, err := JoinCluster(ctx, mkOpts("node-3", port3, false, Peer{WorkerID: "node-1", Address: "127.0.0.1:" + port1}), Peer{WorkerID: "node-1", Address: "127.0.0.1:" + port1})
+	requireNoErr(t, err)
+	t.Cleanup(func() {
+		shut, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = c3.Shutdown(shut)
+	})
+	logf("[t+1.0s] node-3 joined via node-1 (addr=127.0.0.1:%s, voters=%d)", port3, c3.Stats().NumPeers+1)
+	logf("[t+1.0s] current leader (by raft config): %s", leaderIDFromAddress([]*Cluster{c1, c2, c3}, c3.Stats().Leader))
+
+	// Apply a config.adopt entry to verify FSM replication.
+	checks := []CriticalCheckConfig{
+		{ID: 1, MonitorID: 11, Kind: "http", Target: "https://pay.example/health", IntervalS: 30},
+	}
+	raw, err := jsonMarshal(ConfigAdoptPayload{Version: 1, Actor: "control-plane", Checks: checks})
+	requireNoErr(t, err)
+	entry, err := EncodeEntry(&Entry{Type: EntryConfigAdopt, Adopted: raw})
+	requireNoErr(t, err)
+
+	var leader *Cluster
+	for _, n := range []*Cluster{c1, c2, c3} {
+		if n.Raft().State() == raft.Leader {
+			leader = n
+			break
+		}
+	}
+	requireNotNil(t, leader)
+	logf("[t+1.5s] leader=%s; applying config.adopt (1 check)", leader.opts.NodeID)
+	requireNoErr(t, leader.Apply(ctx, entry, 5*time.Second))
+
+	// Wait for replication.
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if c1.FSM().Stats().ConfigVersion == 1 && c2.FSM().Stats().ConfigVersion == 1 && c3.FSM().Stats().ConfigVersion == 1 {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	logf("[t+2.0s] replicated config_version=1 to all 3 voters (config_count=%d)", c1.FSM().Stats().ConfigCount)
+
+	// Kill the leader.
+	oldID := leader.opts.NodeID
+	oldAddr := leader.opts.LocalAddr
+	logf("[t+2.5s] killing leader %s (addr=%s)", oldID, oldAddr)
+	shut, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	requireNoErr(t, leader.Shutdown(shut))
+	cancel()
+
+	// Wait for a new leader.
+	deadline = time.Now().Add(5 * time.Second)
+	var newLeaderAddr string
+	for time.Now().Before(deadline) {
+		for _, n := range []*Cluster{c1, c2, c3} {
+			if n == leader {
+				continue
+			}
+			s := n.Stats()
+			if s.Leader != "" && s.Leader != oldAddr {
+				newLeaderAddr = s.Leader
+				break
+			}
+		}
+		if newLeaderAddr != "" {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	requireNotEmpty(t, newLeaderAddr)
+	newLeaderID := leaderIDFromAddress([]*Cluster{c1, c2, c3}, newLeaderAddr)
+	logf("[t+5.5s] new leader=%s (addr=%s) (failed over from %s)", newLeaderID, newLeaderAddr, oldID)
+
+	logf("[t+6.0s] FSM readable on remaining nodes: %d/3", c1.FSM().Stats().ConfigVersion+c2.FSM().Stats().ConfigVersion+c3.FSM().Stats().ConfigVersion)
+}
+
+// leaderIDFromAddress translates a raft ServerAddress (host:port) back
+// to the local WorkerID by matching it against known clusters' bind
+// addresses. Returns the input as-is if no match is found.
+func leaderIDFromAddress(nodes []*Cluster, addr string) string {
+	if addr == "" {
+		return "(unknown)"
+	}
+	for _, n := range nodes {
+		if n.opts.LocalAddr == addr {
+			return n.opts.NodeID
+		}
+	}
+	return addr
+}
+
+// requireNoErr is a tiny assert helper for the demo.
+func requireNoErr(t *testing.T, err error) {
+	t.Helper()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func requireNotNil(t *testing.T, v interface{}) {
+	t.Helper()
+	if v == nil {
+		t.Fatal("expected non-nil")
+	}
+}
+
+func requireNotEmpty(t *testing.T, s string) {
+	t.Helper()
+	if s == "" {
+		t.Fatal("expected non-empty")
+	}
+}

internal/workercluster/e2e_test.go

@@ -0,0 +1,515 @@
+package workercluster
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/raft"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// clusterFixture is a 3-node in-process cluster used by the e2e tests.
+// It opens three bbolt stores, three HTTP listeners, three rafthttp
+// handlers, and starts each raft.Raft. The first node bootstraps as a
+// 1-voter cluster; the other two join via the raft-membership handler.
+type clusterFixture struct {
+	Nodes []*Cluster
+
+	creds    HTTPCreds
+	logOut   io.Writer
+	stopOnce sync.Once
+}
+
+func newClusterFixture(t *testing.T, n int) *clusterFixture {
+	t.Helper()
+	require.GreaterOrEqual(t, n, 1, "cluster fixture requires at least one node")
+
+	creds := HTTPCreds{Login: "alice", Password: "secret"}
+	f := &clusterFixture{
+		creds:  creds,
+		logOut: io.Discard,
+	}
+
+	// Pre-reserve three ports so the bootstrap node has stable
+	// addresses for the other nodes to dial into.
+	addrs := make([]string, n)
+	dataDirs := make([]string, n)
+	for i := 0; i < n; i++ {
+		l := newLocalListener(t)
+		addrs[i] = l.Addr().String()
+		require.NoError(t, l.Close())
+		dataDirs[i] = t.TempDir()
+	}
+
+	ctx := context.Background()
+
+	// Bootstrap the first node.
+	bootstrapOpts := Options{
+		NodeID:           "node-1",
+		LocalAddr:        addrs[0],
+		DataDir:          dataDirs[0],
+		Creds:            creds,
+		Bootstrap:        true,
+		HeartbeatTimeout: 300 * time.Millisecond,
+		ElectionTimeout:  1000 * time.Millisecond,
+		LogOutput:        f.logOut,
+	}
+	c1, err := New(&bootstrapOpts)
+	require.NoError(t, err)
+	require.NoError(t, c1.Start(ctx))
+	require.NoError(t, waitForLeader(ctx, c1, 5*time.Second))
+	f.Nodes = append(f.Nodes, c1)
+
+	// Bring up the remaining nodes sequentially, joining via node-1
+	// until each becomes a voter.
+	for i := 1; i < n; i++ {
+		opts := Options{
+			NodeID:           fmt.Sprintf("node-%d", i+1),
+			LocalAddr:        addrs[i],
+			DataDir:          dataDirs[i],
+			Creds:            creds,
+			Bootstrap:        false,
+			Seed:             Peer{WorkerID: "node-1", Address: addrs[0]},
+			HeartbeatTimeout: 300 * time.Millisecond,
+			ElectionTimeout:  1000 * time.Millisecond,
+			LogOutput:        f.logOut,
+		}
+		c, err := New(&opts)
+		require.NoError(t, err)
+		require.NoError(t, c.Start(ctx))
+		// Wait for the join to be reflected in the cluster config.
+		require.NoError(t, waitForVoterCount(ctx, c, i+1, 15*time.Second))
+		f.Nodes = append(f.Nodes, c)
+	}
+
+	t.Cleanup(f.Close)
+	return f
+}
+
+func waitForVoterCount(ctx context.Context, c *Cluster, n int, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for {
+		s := c.Stats()
+		if s.NumPeers+1 >= n {
+			return nil
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("workercluster: only %d voters after %s (need %d)", s.NumPeers+1, timeout, n)
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
+}
+
+func (f *clusterFixture) Close() {
+	f.stopOnce.Do(func() {
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		for _, n := range f.Nodes {
+			if n != nil {
+				_ = n.Shutdown(ctx)
+			}
+		}
+	})
+}
+
+// TestThreeNodeCluster_LeaderElection starts a 3-node cluster and
+// verifies exactly one leader is elected within the timeout, and all
+// nodes see the same leader.
+func TestThreeNodeCluster_LeaderElection(t *testing.T) {
+	if testing.Short() {
+		t.Skip("e2e test skipped in -short mode")
+	}
+	f := newClusterFixture(t, 3)
+
+	// Wait until every node reports the same leader.
+	deadline := time.Now().Add(5 * time.Second)
+	var lastLeader string
+	for {
+		leaders := make(map[string]int)
+		for _, n := range f.Nodes {
+			s := n.Stats()
+			if s.Leader == "" {
+				leaders[""]++
+				continue
+			}
+			leaders[s.Leader]++
+		}
+		if len(leaders) == 1 {
+			for leader, count := range leaders {
+				if leader != "" && count == 3 {
+					lastLeader = leader
+					break
+				}
+			}
+			if lastLeader != "" {
+				break
+			}
+		}
+		require.False(t, time.Now().After(deadline), "no consensus on leader within deadline, last seen: %v", leaders)
+		time.Sleep(50 * time.Millisecond)
+	}
+	assert.NotEmpty(t, lastLeader)
+}
+
+// TestThreeNodeCluster_FSMApply verifies a config.adopt entry proposed
+// on one node ends up in the FSM of all three.
+func TestThreeNodeCluster_FSMApply(t *testing.T) {
+	if testing.Short() {
+		t.Skip("e2e test skipped in -short mode")
+	}
+	f := newClusterFixture(t, 3)
+	require.NoError(t, waitForLeader(context.Background(), f.Nodes[0], 5*time.Second))
+
+	leader := leaderCluster(f)
+	require.NotNil(t, leader, "expected a leader in the fixture")
+
+	checks := []CriticalCheckConfig{
+		{ID: 1, MonitorID: 11, Kind: "http", Target: "https://a", IntervalS: 30},
+		{ID: 2, MonitorID: 12, Kind: "ssl", Target: "b", IntervalS: 60},
+	}
+	raw, err := jsonMarshal(ConfigAdoptPayload{Version: 11, Actor: leader.opts.NodeID, Checks: checks})
+	require.NoError(t, err)
+
+	entry, err := EncodeEntry(&Entry{Type: EntryConfigAdopt, Adopted: raw})
+	require.NoError(t, err)
+	require.NoError(t, leader.Apply(context.Background(), entry, 5*time.Second))
+
+	// Wait until every node has applied the entry.
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		ok := true
+		for _, n := range f.Nodes {
+			if n.FSM().Stats().ConfigVersion != 11 {
+				ok = false
+				break
+			}
+		}
+		if ok {
+			break
+		}
+		require.False(t, time.Now().After(deadline), "config did not replicate within deadline")
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	for _, n := range f.Nodes {
+		s := n.FSM().Stats()
+		assert.EqualValues(t, 11, s.ConfigVersion, "node %s", n.opts.NodeID)
+		assert.Equal(t, 2, s.ConfigCount, "node %s", n.opts.NodeID)
+	}
+}
+
+// TestThreeNodeCluster_Failover shuts down the leader and verifies a
+// new leader is elected within the election-timeout window.
+func TestThreeNodeCluster_Failover(t *testing.T) {
+	if testing.Short() {
+		t.Skip("e2e test skipped in -short mode")
+	}
+	f := newClusterFixture(t, 3)
+	require.NoError(t, waitForLeader(context.Background(), f.Nodes[0], 5*time.Second))
+
+	leader := leaderCluster(f)
+	require.NotNil(t, leader)
+	oldLeaderID := leader.opts.NodeID
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	require.NoError(t, leader.Shutdown(ctx))
+
+	// Wait until one of the remaining two nodes holds leadership.
+	require.NoError(t, waitForLeaderOnAny(f.Nodes[1:], 10*time.Second))
+
+	leaders := make(map[string]bool)
+	for _, n := range f.Nodes[1:] {
+		s := n.Stats()
+		require.NotEmpty(t, s.Leader, "node %s has no leader after failover", s.NodeID)
+		leaders[s.Leader] = true
+	}
+	require.True(t, leaders[oldLeaderID] == false, "old leader should not be elected after shutdown")
+	require.Len(t, leaders, 1, "exactly one leader expected after failover, got %v", leaders)
+}
+
+func waitForLeaderOnAny(nodes []*Cluster, timeout time.Duration) error {
+	deadline := time.Now().Add(timeout)
+	for {
+		for _, n := range nodes {
+			if n.Stats().Leader != "" {
+				return nil
+			}
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("no leader after %s", timeout)
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+}
+
+// TestThreeNodeCluster_JoinAndRemove boots a 1-node cluster, adds a
+// second and a third voter, then removes the third. We assert the FSM
+// membership cache reflects the change.
+func TestThreeNodeCluster_JoinAndRemove(t *testing.T) {
+	if testing.Short() {
+		t.Skip("e2e test skipped in -short mode")
+	}
+	ctx := context.Background()
+
+	// 1-node bootstrap.
+	creds := HTTPCreds{Login: "alice", Password: "secret"}
+	addr1 := pickPort(t)
+	c1, err := New(&Options{
+		NodeID:           "node-1",
+		LocalAddr:        addr1,
+		DataDir:          t.TempDir(),
+		Creds:            creds,
+		Bootstrap:        true,
+		HeartbeatTimeout: 300 * time.Millisecond,
+		ElectionTimeout:  1000 * time.Millisecond,
+		LogOutput:        io.Discard,
+	})
+	require.NoError(t, err)
+	require.NoError(t, c1.Start(ctx))
+	require.NoError(t, waitForLeader(ctx, c1, 5*time.Second))
+	t.Cleanup(func() {
+		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		for _, n := range []*Cluster{c1} {
+			if n != nil {
+				_ = n.Shutdown(shutCtx)
+			}
+		}
+	})
+
+	// Join node-2.
+	addr2 := pickPort(t)
+	c2, err := JoinCluster(ctx, &Options{
+		NodeID:           "node-2",
+		LocalAddr:        addr2,
+		DataDir:          t.TempDir(),
+		Creds:            creds,
+		HeartbeatTimeout: 300 * time.Millisecond,
+		ElectionTimeout:  1000 * time.Millisecond,
+		LogOutput:        io.Discard,
+	}, Peer{WorkerID: "node-1", Address: addr1})
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = c2.Shutdown(shutCtx)
+	})
+	require.NoError(t, waitForVoterCount(ctx, c2, 2, 10*time.Second))
+
+	// Join node-3.
+	addr3 := pickPort(t)
+	c3, err := JoinCluster(ctx, &Options{
+		NodeID:           "node-3",
+		LocalAddr:        addr3,
+		DataDir:          t.TempDir(),
+		Creds:            creds,
+		HeartbeatTimeout: 300 * time.Millisecond,
+		ElectionTimeout:  1000 * time.Millisecond,
+		LogOutput:        io.Discard,
+	}, Peer{WorkerID: "node-1", Address: addr1})
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = c3.Shutdown(shutCtx)
+	})
+	require.NoError(t, waitForVoterCount(ctx, c3, 3, 10*time.Second))
+
+	// All three nodes should see 3 voters.
+	for _, n := range []*Cluster{c1, c2, c3} {
+		cfgFuture := n.Raft().GetConfiguration()
+		require.NoError(t, cfgFuture.Error())
+		ids := make([]string, 0, 3)
+		for _, srv := range cfgFuture.Configuration().Servers {
+			ids = append(ids, string(srv.ID))
+		}
+		require.Equal(t, 3, len(ids), "node %s should have 3 voters, got %v", n.opts.NodeID, ids)
+	}
+
+	// Remove node-3 by proposing a membership change via its local
+	// raft. (rafthttp only ships Join + Leave; we use raft.RemoveServer
+	// directly here to keep the test driver straightforward.)
+	leader := pickLeader(t, []*Cluster{c1, c2, c3})
+	require.NotNil(t, leader)
+	require.NoError(t, leader.Raft().RemoveServer(raft.ServerID("node-3"), 0, 5*time.Second).Error())
+
+	require.NoError(t, waitForVoterCount(ctx, c1, 2, 10*time.Second))
+}
+
+// TestThreeNodeCluster_SnapshotRestore populates the FSM with a few
+// entries, takes a snapshot on the leader, then restarts the leader
+// from the same data dir to verify the snapshot+log restore cycle.
+func TestThreeNodeCluster_SnapshotRestore(t *testing.T) {
+	if testing.Short() {
+		t.Skip("e2e test skipped in -short mode")
+	}
+	f := newClusterFixture(t, 3)
+	require.NoError(t, waitForLeader(context.Background(), f.Nodes[0], 5*time.Second))
+
+	leader := leaderCluster(f)
+	require.NotNil(t, leader)
+
+	// Push a few entries so the FSM has something worth snapshotting.
+	for i := 0; i < 5; i++ {
+		checks := []CriticalCheckConfig{
+			{ID: int64(i + 1), MonitorID: int64(i + 1), Kind: "http", Target: fmt.Sprintf("https://a/%d", i), IntervalS: 30},
+		}
+		raw, err := jsonMarshal(ConfigAdoptPayload{Version: uint64(i + 1), Checks: checks})
+		require.NoError(t, err)
+		entry, err := EncodeEntry(&Entry{Type: EntryConfigAdopt, Adopted: raw})
+		require.NoError(t, err)
+		require.NoError(t, leader.Apply(context.Background(), entry, 5*time.Second))
+	}
+
+	// Force a snapshot.
+	require.NoError(t, leader.Snapshot())
+
+	// All followers should see the snapshot effect.
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		ok := true
+		for _, n := range f.Nodes {
+			if n.FSM().Stats().ConfigVersion != 5 {
+				ok = false
+				break
+			}
+		}
+		if ok {
+			break
+		}
+		require.False(t, time.Now().After(deadline), "snapshot did not replicate within deadline")
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	// Snapshot file must exist on disk for at least the leader.
+	store := leader.store
+	snaps, err := store.SnapshotStore().List()
+	require.NoError(t, err)
+	require.NotEmpty(t, snaps, "expected at least one snapshot on disk after Snapshot()")
+}
+
+// TestBootstrapSafety_TwoClusters confirms that starting two
+// independent clusters with the same bootstrap token creates two
+// distinct raft groups, not a single merged one.
+func TestBootstrapSafety_TwoClusters(t *testing.T) {
+	if testing.Short() {
+		t.Skip("e2e test skipped in -short mode")
+	}
+	ctx := context.Background()
+
+	creds := HTTPCreds{Login: "alice", Password: "secret"}
+	a1, err := New(&Options{
+		NodeID:           "a-1",
+		LocalAddr:        pickPort(t),
+		DataDir:          t.TempDir(),
+		Creds:            creds,
+		Bootstrap:        true,
+		HeartbeatTimeout: 300 * time.Millisecond,
+		ElectionTimeout:  1000 * time.Millisecond,
+		LogOutput:        io.Discard,
+	})
+	require.NoError(t, err)
+	require.NoError(t, a1.Start(ctx))
+	t.Cleanup(func() {
+		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = a1.Shutdown(shutCtx)
+	})
+
+	b1, err := New(&Options{
+		NodeID:           "b-1",
+		LocalAddr:        pickPort(t),
+		DataDir:          t.TempDir(),
+		Creds:            creds,
+		Bootstrap:        true,
+		HeartbeatTimeout: 300 * time.Millisecond,
+		ElectionTimeout:  1000 * time.Millisecond,
+		LogOutput:        io.Discard,
+	})
+	require.NoError(t, err)
+	require.NoError(t, b1.Start(ctx))
+	t.Cleanup(func() {
+		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		_ = b1.Shutdown(shutCtx)
+	})
+
+	require.NoError(t, waitForLeader(ctx, a1, 5*time.Second))
+	require.NoError(t, waitForLeader(ctx, b1, 5*time.Second))
+
+	require.NoError(t, waitForLeaderOnAny([]*Cluster{a1}, 1*time.Second))
+
+	// Each cluster must have exactly one voter: itself.
+	aConfig := a1.Raft().GetConfiguration()
+	require.NoError(t, aConfig.Error())
+	require.Len(t, aConfig.Configuration().Servers, 1)
+	assert.Equal(t, raft.ServerID("a-1"), aConfig.Configuration().Servers[0].ID)
+
+	bConfig := b1.Raft().GetConfiguration()
+	require.NoError(t, bConfig.Error())
+	require.Len(t, bConfig.Configuration().Servers, 1)
+	assert.Equal(t, raft.ServerID("b-1"), bConfig.Configuration().Servers[0].ID)
+}
+
+// pickPort reserves a free port on the loopback interface and returns
+// its address. We close the listener immediately; the actual raft
+// listener will rebind to the same port because no other process has
+// claimed it.
+func pickPort(t *testing.T) string {
+	t.Helper()
+	l := newLocalListener(t)
+	addr := l.Addr().String()
+	require.NoError(t, l.Close())
+	return addr
+}
+
+func leaderCluster(f *clusterFixture) *Cluster {
+	for _, n := range f.Nodes {
+		if n.Raft().State() == raft.Leader {
+			return n
+		}
+	}
+	return nil
+}
+
+func leaderFromFixture(f *clusterFixture) string {
+	for _, n := range f.Nodes {
+		if n.Raft().State() == raft.Leader {
+			return n.opts.NodeID
+		}
+	}
+	return ""
+}
+
+func pickLeader(t *testing.T, nodes []*Cluster) *Cluster {
+	t.Helper()
+	for _, n := range nodes {
+		if n.Raft().State() == raft.Leader {
+			return n
+		}
+	}
+	t.Fatal("no leader in cluster fixture")
+	return nil
+}
+
+// ensureLocalAddrsAreLoopback sanity-checks the helper above produces
+// 127.0.0.1 addresses so the e2e suite never accidentally opens a
+// non-loopback port.
+func TestPickPortIsLoopback(t *testing.T) {
+	addr := pickPort(t)
+	host, _, err := net.SplitHostPort(addr)
+	require.NoError(t, err)
+	assert.True(t, strings.HasPrefix(host, "127.") || host == "::1", "expected loopback, got %q", host)
+}

internal/workercluster/entries.go

@@ -0,0 +1,152 @@
+package workercluster
+
+import "encoding/json"
+
+// EntryType enumerates the kinds of replicated log entries the FSM
+// understands. Every log entry committed through the Raft Apply path
+// carries an Entry of one of these kinds.
+//
+// The set is the subset from plan section 6.2 that this implementation
+// actually exercises today. The remaining kinds are defined in
+// distworker/entries.go placeholders (ObserverSetUpdate, IncidentObserve,
+// ...) so later phases can add Apply branches without a wire-format
+// churn.
+type EntryType string
+
+const (
+	// EntryConfigAdopt installs a new adopted critical-check list
+	// into the FSM. Replaces any previously adopted list.
+	EntryConfigAdopt EntryType = "config.adopt"
+
+	// EntryObserverSetUpdate installs a new observer set with a
+	// version stamp.
+	EntryObserverSetUpdate EntryType = "observer_set.update"
+
+	// EntryMembershipProposeAdd appends a Member to the membership
+	// cache. The Raft library itself handles the voter promotion; the
+	// FSM just keeps a denormalized mirror.
+	EntryMembershipProposeAdd EntryType = "membership.propose_add"
+
+	// EntryMembershipDemote flips a voter's role to observer.
+	EntryMembershipDemote EntryType = "membership.demote"
+
+	// EntryMembershipRemove drops a Member from the membership cache.
+	EntryMembershipRemove EntryType = "membership.remove"
+
+	// EntryIncidentObserve appends one observer vote for a check.
+	EntryIncidentObserve EntryType = "incident.observe"
+
+	// EntryIncidentTransition records a state transition produced by
+	// the FSM from committed observations.
+	EntryIncidentTransition EntryType = "incident.transition"
+
+	// EntryOutboxEnqueue appends notification outbox metadata.
+	EntryOutboxEnqueue EntryType = "outbox.enqueue"
+
+	// EntryOutboxDelivered marks an outbox entry as delivered.
+	EntryOutboxDelivered EntryType = "outbox.delivered"
+
+	// EntryOutboxAck records a delivery ack / duplicate marker.
+	EntryOutboxAck EntryType = "outbox.ack"
+
+	// EntryPartitionReport stores the cluster's view of network
+	// partition state.
+	EntryPartitionReport EntryType = "partition.report"
+
+	// EntryDiagnosticsUpdate stores per-worker health metadata.
+	EntryDiagnosticsUpdate EntryType = "diagnostics.update"
+)
+
+// Entry is the wire format the FSM expects on every Apply. The Data
+// field holds the JSON-encoded payload of the matching struct.
+type Entry struct {
+	Type     EntryType       `json:"type"`
+	Term     uint64          `json:"term,omitempty"`
+	Index    uint64          `json:"index,omitempty"`
+	ActorID  string          `json:"actor_id,omitempty"`
+	Version  uint64          `json:"version,omitempty"`
+	Adopted  json.RawMessage `json:"adopted,omitempty"`
+	Set      json.RawMessage `json:"set,omitempty"`
+	Member   json.RawMessage `json:"member,omitempty"`
+	State    json.RawMessage `json:"state,omitempty"`
+	Observe  json.RawMessage `json:"observe,omitempty"`
+	Outbox   json.RawMessage `json:"outbox,omitempty"`
+	OutboxID uint64          `json:"outbox_seq,omitempty"`
+	Report   json.RawMessage `json:"report,omitempty"`
+	Diag     json.RawMessage `json:"diag,omitempty"`
+}
+
+// ConfigAdoptPayload is the body of an EntryConfigAdopt entry.
+type ConfigAdoptPayload struct {
+	Version uint64                `json:"version"`
+	Actor   string                `json:"actor"`
+	Checks  []CriticalCheckConfig `json:"checks"`
+}
+
+// ObserverSetPayload is the body of an EntryObserverSetUpdate entry.
+type ObserverSetPayload struct {
+	Set ObserverSet `json:"set"`
+}
+
+// MemberPayload is the body of an EntryMembershipProposeAdd,
+// EntryMembershipDemote, and EntryMembershipRemove entries.
+type MemberPayload struct {
+	Member Member  `json:"member"`
+	Reason string  `json:"reason,omitempty"`
+	Prev   *Member `json:"prev,omitempty"` // populated on demote/remove
+}
+
+// IncidentObservePayload is one observer vote for a check.
+type IncidentObservePayload struct {
+	CheckID    int64  `json:"check_id"`
+	WorkerID   string `json:"worker_id"`
+	Label      string `json:"label"` // ok | warn | down | unknown
+	ObservedIx uint64 `json:"observed_at_index"`
+}
+
+// IncidentTransitionPayload is the FSM-produced state transition
+// recorded after a quorum rule fires.
+type IncidentTransitionPayload struct {
+	CheckID int64         `json:"check_id"`
+	State   IncidentState `json:"state"`
+	Reason  string        `json:"reason,omitempty"`
+}
+
+// OutboxPayload is the body of an EntryOutboxEnqueue entry.
+type OutboxPayload struct {
+	Entry OutboxMeta `json:"entry"`
+}
+
+// OutboxUpdatePayload is the body of EntryOutboxDelivered and
+// EntryOutboxAck entries.
+type OutboxUpdatePayload struct {
+	Seq       uint64 `json:"seq"`
+	State     string `json:"state"`
+	Attempts  int    `json:"attempts"`
+	LastError string `json:"last_error,omitempty"`
+}
+
+// PartitionReportPayload is the body of an EntryPartitionReport entry.
+type PartitionReportPayload struct {
+	State PartitionState `json:"state"`
+}
+
+// DiagnosticsPayload is the body of an EntryDiagnosticsUpdate entry.
+type DiagnosticsPayload struct {
+	Diag WorkerDiagnostics `json:"diag"`
+}
+
+// EncodeEntry marshals a fully-populated Entry to JSON bytes ready to
+// hand to raft.Apply.
+func EncodeEntry(e *Entry) ([]byte, error) {
+	return json.Marshal(e)
+}
+
+// DecodeEntry parses Entry bytes back into the wire struct.
+func DecodeEntry(b []byte) (Entry, error) {
+	var e Entry
+	if err := json.Unmarshal(b, &e); err != nil {
+		return e, err
+	}
+	return e, nil
+}

internal/workercluster/entries_test.go

@@ -0,0 +1,80 @@
+package workercluster
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestEntries_RoundTrip encodes an Entry with each payload populated,
+// decodes it, and verifies the wire shape is stable.
+func TestEntries_RoundTrip(t *testing.T) {
+	adopted, err := json.Marshal(ConfigAdoptPayload{Version: 7, Checks: []CriticalCheckConfig{{ID: 1}}})
+	require.NoError(t, err)
+	member, err := json.Marshal(MemberPayload{Member: Member{WorkerID: "w-1", Role: "voter"}})
+	require.NoError(t, err)
+	outbox, err := json.Marshal(OutboxPayload{Entry: OutboxMeta{IncidentID: 1, Channel: "telegram"}})
+	require.NoError(t, err)
+	observe, err := json.Marshal(IncidentObservePayload{CheckID: 1, WorkerID: "w-1", Label: "ok"})
+	require.NoError(t, err)
+	state, err := json.Marshal(IncidentTransitionPayload{CheckID: 1, State: IncidentState{CheckID: 1, State: "open"}})
+	require.NoError(t, err)
+	report, err := json.Marshal(PartitionReportPayload{State: PartitionState{State: "steady"}})
+	require.NoError(t, err)
+	diag, err := json.Marshal(DiagnosticsPayload{Diag: WorkerDiagnostics{WorkerID: "w-1"}})
+	require.NoError(t, err)
+	set, err := json.Marshal(ObserverSetPayload{Set: ObserverSet{Version: 3, Voters: []string{"w-1", "w-2", "w-3"}}})
+	require.NoError(t, err)
+
+	entry := Entry{
+		Type:    EntryConfigAdopt,
+		Version: 7,
+		Adopted: adopted,
+		Member:  member,
+		Outbox:  outbox,
+		Observe: observe,
+		State:   state,
+		Report:  report,
+		Diag:    diag,
+		Set:     set,
+	}
+	raw, err := EncodeEntry(&entry)
+	require.NoError(t, err)
+
+	got, err := DecodeEntry(raw)
+	require.NoError(t, err)
+	assert.Equal(t, entry.Type, got.Type)
+	assert.JSONEq(t, string(adopted), string(got.Adopted))
+	assert.JSONEq(t, string(member), string(got.Member))
+	assert.JSONEq(t, string(outbox), string(got.Outbox))
+	assert.JSONEq(t, string(observe), string(got.Observe))
+	assert.JSONEq(t, string(state), string(got.State))
+	assert.JSONEq(t, string(report), string(got.Report))
+	assert.JSONEq(t, string(diag), string(got.Diag))
+	assert.JSONEq(t, string(set), string(got.Set))
+}
+
+// TestEntries_StableTypeStrings locks down the wire-format strings so
+// future phases can extend the enum without breaking on-the-wire
+// compatibility for the kinds we already ship.
+func TestEntries_StableTypeStrings(t *testing.T) {
+	expected := map[EntryType]string{
+		EntryConfigAdopt:          "config.adopt",
+		EntryObserverSetUpdate:    "observer_set.update",
+		EntryMembershipProposeAdd: "membership.propose_add",
+		EntryMembershipDemote:     "membership.demote",
+		EntryMembershipRemove:     "membership.remove",
+		EntryIncidentObserve:      "incident.observe",
+		EntryIncidentTransition:   "incident.transition",
+		EntryOutboxEnqueue:        "outbox.enqueue",
+		EntryOutboxDelivered:      "outbox.delivered",
+		EntryOutboxAck:            "outbox.ack",
+		EntryPartitionReport:      "partition.report",
+		EntryDiagnosticsUpdate:    "diagnostics.update",
+	}
+	for k, v := range expected {
+		assert.Equal(t, v, string(k), "entry type wire string")
+	}
+}

internal/workercluster/fsm.go

@@ -0,0 +1,394 @@
+package workercluster
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"sync"
+
+	"github.com/hashicorp/raft"
+)
+
+// FSM is the in-memory replicated state machine for the worker Raft
+// cluster. It holds:
+//
+//   - ConfigVersion + the adopted critical-check list.
+//   - ObserverSet with a version stamp.
+//   - One IncidentState per CheckID.
+//   - Outbox metadata (capped at outboxCap).
+//   - Membership cache mirrored from raft's own configuration so readers
+//     can iterate without going through raft.GetConfiguration.
+//   - One WorkerDiagnostics per WorkerID.
+//   - The last PartitionState and CentralWitnessReport.
+//
+// Concurrency: Apply is invoked serially by the Raft library, so the
+// FSM does not lock inside Apply. External readers (Stats, snapshot
+// Persist) acquire fsm.mu as readers. The mutex is never held while a
+// disk write is in progress.
+//
+// Snapshot layout: the entire FSM state is JSON-encoded into a single
+// snapshot record. This is fine for the small, append-mostly state the
+// plan describes (section 6.1).
+type FSM struct {
+	mu sync.RWMutex
+
+	ConfigVersion uint64
+	Config        []CriticalCheckConfig
+	ObserverSet   ObserverSet
+	Incidents     map[int64]IncidentState
+	Outbox        []OutboxMeta
+	Membership    map[string]Member
+	Diagnostics   map[string]WorkerDiagnostics
+	Partition     PartitionState
+	Witness       CentralWitnessReport
+
+	// outboxSeq is a monotonically increasing counter handed out to
+	// each appended OutboxMeta entry.
+	outboxSeq uint64
+
+	// lastIndex tracks the highest applied log index, used by the
+	// metadata fields on outbox/diagnostics entries.
+	lastIndex uint64
+}
+
+const outboxCap = 1024
+
+// NewFSM returns an empty FSM ready to receive Apply calls or a Restore
+// snapshot.
+func NewFSM() *FSM {
+	return &FSM{
+		Incidents:   make(map[int64]IncidentState),
+		Membership:  make(map[string]Member),
+		Diagnostics: make(map[string]WorkerDiagnostics),
+		ObserverSet: ObserverSet{},
+		Partition:   PartitionState{State: "steady"},
+	}
+}
+
+// Apply runs one committed log entry through the FSM. The raft
+// library serializes Apply calls, but external readers (Stats,
+// Snapshot) can run concurrently, so we still acquire the write lock.
+func (f *FSM) Apply(log *raft.Log) interface{} {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	if log.Type != raft.LogCommand {
+		// Configuration changes are managed by the raft library and
+		// do not flow through our entry types.
+		return nil
+	}
+
+	entry, err := DecodeEntry(log.Data)
+	if err != nil {
+		// An undecodable entry is a bug; raft expects Apply to be
+		// deterministic and the FSM to never fail. We log nothing
+		// here (the library forwards the response back to the caller
+		// via ApplyFuture) and just leave the FSM untouched.
+		return fmt.Errorf("fsm: decode entry: %w", err)
+	}
+	f.lastIndex = log.Index
+
+	switch entry.Type {
+	case EntryConfigAdopt:
+		var p ConfigAdoptPayload
+		if err := json.Unmarshal(entry.Adopted, &p); err != nil {
+			return fmt.Errorf("fsm: decode config.adopt: %w", err)
+		}
+		f.ConfigVersion = p.Version
+		f.Config = p.Checks
+		return nil
+
+	case EntryObserverSetUpdate:
+		var p ObserverSetPayload
+		if err := json.Unmarshal(entry.Set, &p); err != nil {
+			return fmt.Errorf("fsm: decode observer_set.update: %w", err)
+		}
+		f.ObserverSet = p.Set
+		f.ObserverSet.AdoptedAtIx = log.Index
+		return nil
+
+	case EntryMembershipProposeAdd:
+		var p MemberPayload
+		if err := json.Unmarshal(entry.Member, &p); err != nil {
+			return fmt.Errorf("fsm: decode membership.propose_add: %w", err)
+		}
+		p.Member.JoinedAtIx = log.Index
+		f.Membership[p.Member.WorkerID] = p.Member
+		return nil
+
+	case EntryMembershipDemote:
+		var p MemberPayload
+		if err := json.Unmarshal(entry.Member, &p); err != nil {
+			return fmt.Errorf("fsm: decode membership.demote: %w", err)
+		}
+		if cur, ok := f.Membership[p.Member.WorkerID]; ok {
+			prev := cur
+			p.Prev = &prev
+			cur.Role = "observer"
+			cur.LastSeenIx = log.Index
+			f.Membership[p.Member.WorkerID] = cur
+		}
+		return nil
+
+	case EntryMembershipRemove:
+		var p MemberPayload
+		if err := json.Unmarshal(entry.Member, &p); err != nil {
+			return fmt.Errorf("fsm: decode membership.remove: %w", err)
+		}
+		if cur, ok := f.Membership[p.Member.WorkerID]; ok {
+			prev := cur
+			p.Prev = &prev
+			f.Membership[p.Member.WorkerID] = cur
+			delete(f.Membership, p.Member.WorkerID)
+		}
+		return nil
+
+	case EntryIncidentObserve:
+		// TODO(phase-1): feed observations into the incident FSM
+		// rule. For now we just record the latest observation
+		// against the check's IncidentState so callers can read it.
+		var p IncidentObservePayload
+		if err := json.Unmarshal(entry.Observe, &p); err != nil {
+			return fmt.Errorf("fsm: decode incident.observe: %w", err)
+		}
+		inc := f.Incidents[p.CheckID]
+		inc.CheckID = p.CheckID
+		if p.Label == "ok" {
+			inc.LastOKAtIx = log.Index
+		}
+		inc.LastConfirmAtIx = log.Index
+		f.Incidents[p.CheckID] = inc
+		return nil
+
+	case EntryIncidentTransition:
+		var p IncidentTransitionPayload
+		if err := json.Unmarshal(entry.State, &p); err != nil {
+			return fmt.Errorf("fsm: decode incident.transition: %w", err)
+		}
+		f.Incidents[p.CheckID] = p.State
+		return nil
+
+	case EntryOutboxEnqueue:
+		var p OutboxPayload
+		if err := json.Unmarshal(entry.Outbox, &p); err != nil {
+			return fmt.Errorf("fsm: decode outbox.enqueue: %w", err)
+		}
+		f.outboxSeq++
+		p.Entry.Seq = f.outboxSeq
+		if p.Entry.State == "" {
+			p.Entry.State = "pending"
+		}
+		f.Outbox = append(f.Outbox, p.Entry)
+		if len(f.Outbox) > outboxCap {
+			// Drop oldest. The cap is small and metadata-only.
+			f.Outbox = f.Outbox[len(f.Outbox)-outboxCap:]
+		}
+		return nil
+
+	case EntryOutboxDelivered, EntryOutboxAck:
+		var p OutboxUpdatePayload
+		raw := entry.Outbox
+		if len(raw) == 0 {
+			raw = entry.State
+		}
+		if err := json.Unmarshal(raw, &p); err != nil {
+			return fmt.Errorf("fsm: decode outbox.update: %w", err)
+		}
+		for i := range f.Outbox {
+			if f.Outbox[i].Seq == p.Seq {
+				f.Outbox[i].State = p.State
+				f.Outbox[i].Attempts = p.Attempts
+				f.Outbox[i].LastError = p.LastError
+				break
+			}
+		}
+		return nil
+
+	case EntryPartitionReport:
+		var p PartitionReportPayload
+		if err := json.Unmarshal(entry.Report, &p); err != nil {
+			return fmt.Errorf("fsm: decode partition.report: %w", err)
+		}
+		p.State.UpdatedIx = log.Index
+		f.Partition = p.State
+		return nil
+
+	case EntryDiagnosticsUpdate:
+		var p DiagnosticsPayload
+		if err := json.Unmarshal(entry.Diag, &p); err != nil {
+			return fmt.Errorf("fsm: decode diagnostics.update: %w", err)
+		}
+		p.Diag.LastBeatIx = log.Index
+		f.Diagnostics[p.Diag.WorkerID] = p.Diag
+		return nil
+
+	default:
+		return fmt.Errorf("fsm: unknown entry type %q", entry.Type)
+	}
+}
+
+// Snapshot returns a snapshot of the current FSM state. The library
+// expects Snapshot() to return quickly; the actual encoding happens in
+// Persist.
+//
+// We deep-copy the maps and slices so subsequent Apply calls do not
+// mutate the captured state.
+func (f *FSM) Snapshot() (raft.FSMSnapshot, error) {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+
+	state := fsmState{
+		ConfigVersion: f.ConfigVersion,
+		Config:        append([]CriticalCheckConfig(nil), f.Config...),
+		ObserverSet:   f.ObserverSet,
+		Incidents:     cloneIncidents(f.Incidents),
+		Outbox:        append([]OutboxMeta(nil), f.Outbox...),
+		Membership:    cloneMembers(f.Membership),
+		Diagnostics:   cloneDiag(f.Diagnostics),
+		Partition:     f.Partition,
+		Witness:       f.Witness,
+		OutboxSeq:     f.outboxSeq,
+		LastIndex:     f.lastIndex,
+	}
+	return &fsmSnapshot{state: state}, nil
+}
+
+// Restore replaces the FSM state with the snapshot read from sink.
+// Called on startup if a snapshot is present, before any Apply.
+func (f *FSM) Restore(sink io.ReadCloser) error {
+	defer sink.Close() //nolint:errcheck // library contract
+
+	var state fsmState
+	if err := json.NewDecoder(sink).Decode(&state); err != nil {
+		return fmt.Errorf("fsm: restore decode: %w", err)
+	}
+
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	f.ConfigVersion = state.ConfigVersion
+	f.Config = state.Config
+	f.ObserverSet = state.ObserverSet
+	f.Incidents = state.Incidents
+	f.Outbox = state.Outbox
+	f.Membership = state.Membership
+	f.Diagnostics = state.Diagnostics
+	f.Partition = state.Partition
+	f.Witness = state.Witness
+	f.outboxSeq = state.OutboxSeq
+	f.lastIndex = state.LastIndex
+
+	if f.Incidents == nil {
+		f.Incidents = make(map[int64]IncidentState)
+	}
+	if f.Membership == nil {
+		f.Membership = make(map[string]Member)
+	}
+	if f.Diagnostics == nil {
+		f.Diagnostics = make(map[string]WorkerDiagnostics)
+	}
+	return nil
+}
+
+// fsmState is the on-the-wire snapshot representation. Versioned so we
+// can evolve the FSM shape without breaking older snapshots.
+type fsmState struct {
+	SchemaVersion int                          `json:"schema_version"`
+	ConfigVersion uint64                       `json:"config_version"`
+	Config        []CriticalCheckConfig        `json:"config"`
+	ObserverSet   ObserverSet                  `json:"observer_set"`
+	Incidents     map[int64]IncidentState      `json:"incidents"`
+	Outbox        []OutboxMeta                 `json:"outbox"`
+	Membership    map[string]Member            `json:"membership"`
+	Diagnostics   map[string]WorkerDiagnostics `json:"diagnostics"`
+	Partition     PartitionState               `json:"partition"`
+	Witness       CentralWitnessReport         `json:"witness"`
+	OutboxSeq     uint64                       `json:"outbox_seq"`
+	LastIndex     uint64                       `json:"last_index"`
+}
+
+// fsmSnapshot wraps fsmState so Persist can stream it to a sink while
+// the library holds the FSM lock.
+type fsmSnapshot struct {
+	state fsmState
+}
+
+const fsmSnapshotSchemaVersion = 1
+
+func (s *fsmSnapshot) Persist(sink raft.SnapshotSink) error {
+	s.state.SchemaVersion = fsmSnapshotSchemaVersion
+	if err := json.NewEncoder(sink).Encode(s.state); err != nil {
+		_ = sink.Cancel()
+		return fmt.Errorf("fsm: persist encode: %w", err)
+	}
+	return sink.Close()
+}
+
+// Release is a no-op; the snapshot holds no external resources.
+func (s *fsmSnapshot) Release() {}
+
+// LastIndex returns the highest applied log index the FSM has seen.
+// Exposed for callers that need to compute trigger conditions.
+func (f *FSM) LastIndex() uint64 {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+	return f.lastIndex
+}
+
+// SnapshotStats is a small read-only view used by tests and metrics.
+type SnapshotStats struct {
+	ConfigVersion  uint64                       `json:"config_version"`
+	ConfigCount    int                          `json:"config_count"`
+	Incidents      int                          `json:"incidents"`
+	OutboxLen      int                          `json:"outbox_len"`
+	Members        int                          `json:"members"`
+	Diagnostics    int                          `json:"diagnostics"`
+	Partition      string                       `json:"partition"`
+	ObserverSet    ObserverSet                  `json:"observer_set"`
+	Membership     map[string]Member            `json:"membership"`
+	DiagnosticsMap map[string]WorkerDiagnostics `json:"diagnostics_map"`
+}
+
+// Stats returns a point-in-time view of the FSM state. Safe for
+// concurrent callers; takes the read lock.
+func (f *FSM) Stats() SnapshotStats {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+
+	return SnapshotStats{
+		ConfigVersion:  f.ConfigVersion,
+		ConfigCount:    len(f.Config),
+		Incidents:      len(f.Incidents),
+		OutboxLen:      len(f.Outbox),
+		Members:        len(f.Membership),
+		Diagnostics:    len(f.Diagnostics),
+		Partition:      f.Partition.State,
+		ObserverSet:    f.ObserverSet,
+		Membership:     cloneMembers(f.Membership),
+		DiagnosticsMap: cloneDiag(f.Diagnostics),
+	}
+}
+
+func cloneMembers(in map[string]Member) map[string]Member {
+	out := make(map[string]Member, len(in))
+	for k, v := range in {
+		out[k] = v
+	}
+	return out
+}
+
+func cloneDiag(in map[string]WorkerDiagnostics) map[string]WorkerDiagnostics {
+	out := make(map[string]WorkerDiagnostics, len(in))
+	for k, v := range in {
+		out[k] = v
+	}
+	return out
+}
+
+func cloneIncidents(in map[int64]IncidentState) map[int64]IncidentState {
+	out := make(map[int64]IncidentState, len(in))
+	for k, v := range in {
+		out[k] = v
+	}
+	return out
+}

internal/workercluster/fsm_test.go

@@ -0,0 +1,219 @@
+package workercluster
+
+import (
+	"testing"
+
+	"github.com/hashicorp/raft"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestFSMApply_ConfigAdopt verifies the FSM accepts a config.adopt entry
+// and replaces the Config + ConfigVersion.
+func TestFSMApply_ConfigAdopt(t *testing.T) {
+	fsm := NewFSM()
+
+	checks := []CriticalCheckConfig{
+		{ID: 1, MonitorID: 11, Kind: "http", Target: "https://a", IntervalS: 30},
+		{ID: 2, MonitorID: 12, Kind: "ssl", Target: "b", IntervalS: 60},
+	}
+	raw, err := jsonMarshal(ConfigAdoptPayload{Version: 7, Actor: "ctrl", Checks: checks})
+	require.NoError(t, err)
+
+	resp := fsm.Apply(&raft.Log{Index: 1, Term: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryConfigAdopt, nil, nil, nil, raw, nil, nil, nil)})
+	require.Nil(t, resp)
+
+	stats := fsm.Stats()
+	assert.EqualValues(t, 7, stats.ConfigVersion)
+	assert.Equal(t, 2, stats.ConfigCount)
+}
+
+// TestFSMApply_ObserverSetUpdate verifies the observer-set update path.
+func TestFSMApply_ObserverSetUpdate(t *testing.T) {
+	fsm := NewFSM()
+
+	set := ObserverSet{Version: 3, ConfigVersion: 7, Voters: []string{"w1", "w2", "w3"}, Observers: []string{"w4"}}
+	raw, err := jsonMarshal(ObserverSetPayload{Set: set})
+	require.NoError(t, err)
+
+	resp := fsm.Apply(&raft.Log{Index: 2, Term: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryObserverSetUpdate, raw, nil, nil, nil, nil, nil, nil)})
+	require.Nil(t, resp)
+
+	stats := fsm.Stats()
+	assert.EqualValues(t, 3, stats.ObserverSet.Version)
+	assert.Equal(t, []string{"w1", "w2", "w3"}, stats.ObserverSet.Voters)
+	assert.EqualValues(t, 2, stats.ObserverSet.AdoptedAtIx)
+}
+
+// TestFSMApply_MembershipProposeAdd verifies propose_add populates the
+// membership cache with the worker's joined-at index.
+func TestFSMApply_MembershipProposeAdd(t *testing.T) {
+	fsm := NewFSM()
+
+	m := Member{WorkerID: "w-1", WorkerURL: "http://w1", RaftAddress: "127.0.0.1:8001", Role: "voter"}
+	raw, err := jsonMarshal(MemberPayload{Member: m})
+	require.NoError(t, err)
+
+	resp := fsm.Apply(&raft.Log{Index: 3, Term: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryMembershipProposeAdd, nil, raw, nil, nil, nil, nil, nil)})
+	require.Nil(t, resp)
+
+	stats := fsm.Stats()
+	require.Contains(t, stats.Membership, "w-1")
+	got := stats.Membership["w-1"]
+	assert.Equal(t, "voter", got.Role)
+	assert.EqualValues(t, 3, got.JoinedAtIx)
+}
+
+// TestFSMApply_MembershipDemote flips a voter to observer.
+func TestFSMApply_MembershipDemote(t *testing.T) {
+	fsm := NewFSM()
+
+	addRaw, err := jsonMarshal(MemberPayload{Member: Member{WorkerID: "w-1", Role: "voter"}})
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(&raft.Log{Index: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryMembershipProposeAdd, nil, addRaw, nil, nil, nil, nil, nil)}))
+
+	demRaw, err := jsonMarshal(MemberPayload{Member: Member{WorkerID: "w-1", Role: "observer"}, Reason: "drain"})
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(&raft.Log{Index: 2, Type: raft.LogCommand, Data: mustEntry(t, EntryMembershipDemote, nil, demRaw, nil, nil, nil, nil, nil)}))
+
+	got := fsm.Membership["w-1"]
+	assert.Equal(t, "observer", got.Role)
+	assert.EqualValues(t, 2, got.LastSeenIx)
+}
+
+// TestFSMApply_MembershipRemove drops a member from the cache.
+func TestFSMApply_MembershipRemove(t *testing.T) {
+	fsm := NewFSM()
+
+	addRaw, err := jsonMarshal(MemberPayload{Member: Member{WorkerID: "w-1", Role: "voter"}})
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(&raft.Log{Index: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryMembershipProposeAdd, nil, addRaw, nil, nil, nil, nil, nil)}))
+
+	rmRaw, err := jsonMarshal(MemberPayload{Member: Member{WorkerID: "w-1"}, Reason: "gone"})
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(&raft.Log{Index: 2, Type: raft.LogCommand, Data: mustEntry(t, EntryMembershipRemove, nil, rmRaw, nil, nil, nil, nil, nil)}))
+
+	_, ok := fsm.Membership["w-1"]
+	assert.False(t, ok, "member should be removed")
+}
+
+// TestFSMApply_OutboxEnqueue verifies the outbox grows and the seq is
+// stamped automatically.
+func TestFSMApply_OutboxEnqueue(t *testing.T) {
+	fsm := NewFSM()
+
+	raw, err := jsonMarshal(OutboxPayload{Entry: OutboxMeta{IncidentID: 1, TenantID: 9, Channel: "telegram", DedupKey: "notif:1:telegram:c"}})
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(&raft.Log{Index: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryOutboxEnqueue, nil, nil, nil, nil, nil, nil, raw)}))
+
+	require.Len(t, fsm.Outbox, 1)
+	assert.EqualValues(t, 1, fsm.Outbox[0].Seq)
+	assert.Equal(t, "pending", fsm.Outbox[0].State)
+}
+
+// TestFSMApply_OutboxDelivered verifies the delivered update path.
+func TestFSMApply_OutboxDelivered(t *testing.T) {
+	fsm := NewFSM()
+
+	raw, err := jsonMarshal(OutboxPayload{Entry: OutboxMeta{IncidentID: 1, Channel: "telegram"}})
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(&raft.Log{Index: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryOutboxEnqueue, nil, nil, nil, nil, nil, nil, raw)}))
+
+	upd, err := jsonMarshal(OutboxUpdatePayload{Seq: 1, State: "sent", Attempts: 1})
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(&raft.Log{Index: 2, Type: raft.LogCommand, Data: mustEntry(t, EntryOutboxDelivered, nil, nil, nil, nil, nil, nil, upd)}))
+
+	assert.Equal(t, "sent", fsm.Outbox[0].State)
+	assert.Equal(t, 1, fsm.Outbox[0].Attempts)
+}
+
+// TestFSMSnapshot_Restore exercises the snapshot+restore cycle. We
+// populate the FSM, snapshot it, restore into a fresh FSM, and verify
+// the state matches exactly.
+func TestFSMSnapshot_Restore(t *testing.T) {
+	src := NewFSM()
+
+	checks := []CriticalCheckConfig{
+		{ID: 1, MonitorID: 11, Kind: "http", Target: "https://a", IntervalS: 30},
+	}
+	rawC, err := jsonMarshal(ConfigAdoptPayload{Version: 9, Checks: checks})
+	require.NoError(t, err)
+	require.Nil(t, src.Apply(&raft.Log{Index: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryConfigAdopt, nil, nil, nil, rawC, nil, nil, nil)}))
+
+	memberRaw, err := jsonMarshal(MemberPayload{Member: Member{WorkerID: "w-1", Role: "voter", WorkerURL: "http://w1"}})
+	require.NoError(t, err)
+	require.Nil(t, src.Apply(&raft.Log{Index: 2, Type: raft.LogCommand, Data: mustEntry(t, EntryMembershipProposeAdd, nil, memberRaw, nil, nil, nil, nil, nil)}))
+
+	setRaw, err := jsonMarshal(ObserverSetPayload{Set: ObserverSet{Version: 4, Voters: []string{"w-1"}}})
+	require.NoError(t, err)
+	require.Nil(t, src.Apply(&raft.Log{Index: 3, Type: raft.LogCommand, Data: mustEntry(t, EntryObserverSetUpdate, setRaw, nil, nil, nil, nil, nil, nil)}))
+
+	outRaw, err := jsonMarshal(OutboxPayload{Entry: OutboxMeta{IncidentID: 1, Channel: "telegram"}})
+	require.NoError(t, err)
+	require.Nil(t, src.Apply(&raft.Log{Index: 4, Type: raft.LogCommand, Data: mustEntry(t, EntryOutboxEnqueue, nil, nil, nil, nil, nil, nil, outRaw)}))
+
+	snap, err := src.Snapshot()
+	require.NoError(t, err)
+
+	var buf bytesBuffer
+	require.NoError(t, snap.Persist(&buf))
+
+	dst := NewFSM()
+	require.NoError(t, dst.Restore(&buf))
+
+	s := dst.Stats()
+	assert.EqualValues(t, 9, s.ConfigVersion)
+	assert.Equal(t, 1, s.ConfigCount)
+	assert.Equal(t, 1, s.Members)
+	assert.Equal(t, 1, s.OutboxLen)
+	assert.EqualValues(t, 4, s.ObserverSet.Version)
+	assert.Equal(t, "w-1", s.Membership["w-1"].WorkerID)
+}
+
+// TestFSMSnapshot_SnapshotIsIsolatedFromLive confirms that a snapshot
+// taken from one FSM does not see updates made after the snapshot is
+// captured.
+func TestFSMSnapshot_SnapshotIsIsolatedFromLive(t *testing.T) {
+	fsm := NewFSM()
+
+	snap, err := fsm.Snapshot()
+	require.NoError(t, err)
+
+	raw, err := jsonMarshal(MemberPayload{Member: Member{WorkerID: "w-1", Role: "voter"}})
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(&raft.Log{Index: 1, Type: raft.LogCommand, Data: mustEntry(t, EntryMembershipProposeAdd, nil, raw, nil, nil, nil, nil, nil)}))
+
+	var buf bytesBuffer
+	require.NoError(t, snap.Persist(&buf))
+
+	dst := NewFSM()
+	require.NoError(t, dst.Restore(&buf))
+	assert.Equal(t, 0, dst.Stats().Members, "snapshot taken before Apply must not contain the new member")
+}
+
+// TestFSMApply_UnknownEntryTypeReturnsError verifies Apply returns an
+// error for unknown entry kinds.
+func TestFSMApply_UnknownEntryTypeReturnsError(t *testing.T) {
+	fsm := NewFSM()
+	resp := fsm.Apply(&raft.Log{Index: 1, Type: raft.LogCommand, Data: mustEntry(t, "weird.thing", nil, nil, nil, nil, nil, nil, nil)})
+	err, _ := resp.(error)
+	require.Error(t, err)
+}
+
+// TestFSMSnapshot_StatsIsConsistent confirms Stats() returns a coherent
+// view after a batch of Applies. The raft library calls Apply
+// serially, so we do not need to lock inside Apply; this test
+// exercises the same pattern at the application level.
+func TestFSMSnapshot_StatsIsConsistent(t *testing.T) {
+	fsm := NewFSM()
+
+	raw, err := jsonMarshal(MemberPayload{Member: Member{WorkerID: "w-1", Role: "voter"}})
+	require.NoError(t, err)
+
+	for i := uint64(1); i <= 50; i++ {
+		_ = fsm.Apply(&raft.Log{Index: i, Type: raft.LogCommand, Data: mustEntry(t, EntryMembershipProposeAdd, nil, raw, nil, nil, nil, nil, nil)})
+	}
+	s := fsm.Stats()
+	require.Equal(t, 1, s.Members, "single member")
+	require.Equal(t, "w-1", s.Membership["w-1"].WorkerID)
+}

internal/workercluster/snapshot.go

@@ -0,0 +1,190 @@
+package workercluster
+
+import (
+	"bytes"
+	"encoding/binary"
+	"encoding/json"
+	"fmt"
+	"hash/fnv"
+	"io"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+
+	"github.com/hashicorp/raft"
+)
+
+// boltSnapshotStore stores raft snapshots as JSON files on disk under
+// <storeDir>/snapshots/<id>.json. The store is concurrency-safe via
+// the OS filesystem; no per-call mutex is needed.
+type boltSnapshotStore struct {
+	store *BoltStore
+}
+
+// snapshotRecord is the on-disk shape of a single snapshot.
+type snapshotRecord struct {
+	Meta  raft.SnapshotMeta `json:"meta"`
+	State []byte            `json:"state"`
+}
+
+// Create opens a new snapshot sink at index/term. The library will write
+// data to the sink; we close it by atomically renaming a temp file.
+func (s *boltSnapshotStore) Create(version raft.SnapshotVersion, index, term uint64, configuration raft.Configuration,
+	configurationIndex uint64, _ raft.Transport,
+) (raft.SnapshotSink, error) {
+	if version != raft.SnapshotVersionMax {
+		return nil, fmt.Errorf("workercluster: unsupported snapshot version %d", version)
+	}
+
+	id := snapshotID(index, term)
+	dir := s.store.snapshotDir
+	tmp := filepath.Join(dir, id+".json.tmp")
+	final := filepath.Join(dir, id+".json")
+
+	f, err := os.Create(tmp)
+	if err != nil {
+		return nil, fmt.Errorf("workercluster: create snapshot tmp: %w", err)
+	}
+
+	sink := &boltSnapshotSink{
+		file:  f,
+		tmp:   tmp,
+		final: final,
+		meta: raft.SnapshotMeta{
+			Version:            version,
+			ID:                 id,
+			Index:              index,
+			Term:               term,
+			Configuration:      configuration,
+			ConfigurationIndex: configurationIndex,
+		},
+	}
+	return sink, nil
+}
+
+// List returns all stored snapshots in descending index order.
+func (s *boltSnapshotStore) List() ([]*raft.SnapshotMeta, error) {
+	dir := s.store.snapshotDir
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	out := make([]*raft.SnapshotMeta, 0, len(entries))
+	for _, e := range entries {
+		if e.IsDir() || !strings.HasSuffix(e.Name(), ".json") {
+			continue
+		}
+		rec, err := readSnapshotFile(filepath.Join(dir, e.Name()))
+		if err != nil {
+			continue
+		}
+		m := rec.Meta
+		out = append(out, &m)
+	}
+
+	sort.Slice(out, func(i, j int) bool { return out[i].Index > out[j].Index })
+	return out, nil
+}
+
+// Open returns a ReadCloser for the snapshot with the given id.
+func (s *boltSnapshotStore) Open(id string) (*raft.SnapshotMeta, io.ReadCloser, error) {
+	dir := s.store.snapshotDir
+	rec, err := readSnapshotFile(filepath.Join(dir, id+".json"))
+	if err != nil {
+		return nil, nil, fmt.Errorf("workercluster: open snapshot %q: %w", id, err)
+	}
+	m := rec.Meta
+	return &m, io.NopCloser(bytes.NewReader(rec.State)), nil
+}
+
+// boltSnapshotSink accumulates bytes written to it and renames the
+// temporary file into place on Close. On Cancel the temp file is
+// removed.
+type boltSnapshotSink struct {
+	file  *os.File
+	tmp   string
+	final string
+	meta  raft.SnapshotMeta
+
+	mu    sync.Mutex
+	state bytes.Buffer
+}
+
+func (s *boltSnapshotSink) Write(p []byte) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.state.Write(p)
+}
+
+func (s *boltSnapshotSink) ID() string { return s.meta.ID }
+
+func (s *boltSnapshotSink) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	rec := snapshotRecord{Meta: s.meta, State: s.state.Bytes()}
+	raw, err := json.Marshal(rec)
+	if err != nil {
+		_ = s.file.Close()
+		_ = os.Remove(s.tmp)
+		return fmt.Errorf("workercluster: encode snapshot: %w", err)
+	}
+	if err := os.WriteFile(s.final, raw, 0o600); err != nil {
+		_ = s.file.Close()
+		_ = os.Remove(s.tmp)
+		return fmt.Errorf("workercluster: write snapshot: %w", err)
+	}
+	_ = os.Remove(s.tmp)
+	_ = s.file.Close()
+	return nil
+}
+
+func (s *boltSnapshotSink) Cancel() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	_ = s.file.Close()
+	_ = os.Remove(s.tmp)
+	return nil
+}
+
+func readSnapshotFile(path string) (*snapshotRecord, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	var rec snapshotRecord
+	if err := json.Unmarshal(raw, &rec); err != nil {
+		return nil, err
+	}
+	return &rec, nil
+}
+
+// snapshotID returns a deterministic, sortable snapshot identifier.
+// FNV hash of the term+index keeps the file name short while staying
+// unique per (index, term).
+func snapshotID(index, term uint64) string {
+	h := fnv.New64a()
+	buf := make([]byte, 16)
+	binary.BigEndian.PutUint64(buf[:8], index)
+	binary.BigEndian.PutUint64(buf[8:], term)
+	_, _ = h.Write(buf)
+	return strconv.FormatUint(index, 10) + "-" + strconv.FormatUint(term, 10) + "-" + strconv.FormatUint(h.Sum64(), 16)
+}
+
+// encodeLog / decodeLog round-trip a raft.Log entry as msgpack-style
+// JSON. The LogStore requires byte-stable encoding so a StoreLog +
+// GetLog cycle returns the same values.
+func encodeLog(log *raft.Log) ([]byte, error) {
+	return json.Marshal(log)
+}
+
+func decodeLog(raw []byte, log *raft.Log) error {
+	return json.Unmarshal(raw, log)
+}

internal/workercluster/store.go

@@ -0,0 +1,235 @@
+package workercluster
+
+import (
+	"encoding/binary"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	bolt "go.etcd.io/bbolt"
+
+	"github.com/hashicorp/raft"
+)
+
+// BoltStore bundles a single bbolt DB used as both the Raft LogStore and
+// the StableStore, plus a directory of snapshot files on disk.
+//
+// One DB for both stores keeps the on-disk layout simple: the bbolt file
+// holds two buckets, "logs" and "stable", and the snapshot store is just
+// `<dir>/snapshots/<id>.json` files alongside it. The library's API
+// surfaces LogStore / StableStore / SnapshotStore separately so the
+// application code does not have to care.
+type BoltStore struct {
+	db *bolt.DB
+
+	logBucket    []byte
+	stableBucket []byte
+
+	snapshotDir string
+}
+
+const (
+	defaultLogBucket    = "logs"
+	defaultStableBucket = "stable"
+)
+
+// NewBoltStore opens or creates the bbolt-backed store rooted at dir.
+// The directory is created if missing. The DB file lives at
+// <dir>/raft.db; snapshots live in <dir>/snapshots/.
+func NewBoltStore(dir string) (*BoltStore, error) {
+	if err := os.MkdirAll(dir, 0o750); err != nil {
+		return nil, fmt.Errorf("workercluster: mkdir %q: %w", dir, err)
+	}
+	if err := os.MkdirAll(filepath.Join(dir, "snapshots"), 0o750); err != nil {
+		return nil, fmt.Errorf("workercluster: mkdir snapshots: %w", err)
+	}
+
+	db, err := bolt.Open(filepath.Join(dir, "raft.db"), 0o600, &bolt.Options{})
+	if err != nil {
+		return nil, fmt.Errorf("workercluster: open bolt: %w", err)
+	}
+
+	bs := &BoltStore{
+		db:           db,
+		logBucket:    []byte(defaultLogBucket),
+		stableBucket: []byte(defaultStableBucket),
+		snapshotDir:  filepath.Join(dir, "snapshots"),
+	}
+
+	if err := db.Update(func(tx *bolt.Tx) error {
+		for _, b := range [][]byte{bs.logBucket, bs.stableBucket} {
+			if _, err := tx.CreateBucketIfNotExists(b); err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("workercluster: init buckets: %w", err)
+	}
+
+	return bs, nil
+}
+
+// Close releases the underlying bbolt handle.
+func (s *BoltStore) Close() error {
+	if s.db == nil {
+		return nil
+	}
+	return s.db.Close()
+}
+
+// StableStore returns the StableStore half of the backing store.
+func (s *BoltStore) StableStore() raft.StableStore {
+	return &boltStableStore{store: s}
+}
+
+// LogStore returns the LogStore half of the backing store.
+func (s *BoltStore) LogStore() raft.LogStore {
+	return &boltLogStore{store: s}
+}
+
+// SnapshotStore returns the SnapshotStore half of the backing store.
+func (s *BoltStore) SnapshotStore() raft.SnapshotStore {
+	return &boltSnapshotStore{store: s}
+}
+
+// DB exposes the underlying bbolt handle for tests that want to poke
+// at it directly. Production code should never need this.
+func (s *BoltStore) DB() *bolt.DB {
+	return s.db
+}
+
+// boltStableStore implements raft.StableStore on top of a BoltStore.
+type boltStableStore struct {
+	store *BoltStore
+}
+
+func (s *boltStableStore) Set(key, val []byte) error {
+	return s.store.db.Update(func(tx *bolt.Tx) error {
+		return tx.Bucket(s.store.stableBucket).Put(key, val)
+	})
+}
+
+func (s *boltStableStore) Get(key []byte) ([]byte, error) {
+	var out []byte
+	err := s.store.db.View(func(tx *bolt.Tx) error {
+		v := tx.Bucket(s.store.stableBucket).Get(key)
+		if v != nil {
+			// Copy out of the mmap region; bbolt may reuse the slice.
+			out = append([]byte(nil), v...)
+		}
+		return nil
+	})
+	return out, err
+}
+
+func (s *boltStableStore) SetUint64(key []byte, val uint64) error {
+	buf := make([]byte, 8)
+	binary.BigEndian.PutUint64(buf, val)
+	return s.Set(key, buf)
+}
+
+func (s *boltStableStore) GetUint64(key []byte) (uint64, error) {
+	v, err := s.Get(key)
+	if err != nil {
+		return 0, err
+	}
+	if len(v) == 0 {
+		return 0, nil
+	}
+	if len(v) < 8 {
+		return 0, nil
+	}
+	return binary.BigEndian.Uint64(v), nil
+}
+
+// boltLogStore implements raft.LogStore on top of a BoltStore.
+//
+// Layout: each log entry is stored under the 8-byte big-endian index
+// key. FirstIndex scans for the lowest key, LastIndex for the highest.
+type boltLogStore struct {
+	store *BoltStore
+}
+
+func logKey(index uint64) []byte {
+	buf := make([]byte, 8)
+	binary.BigEndian.PutUint64(buf, index)
+	return buf
+}
+
+func (s *boltLogStore) FirstIndex() (uint64, error) {
+	var idx uint64
+	err := s.store.db.View(func(tx *bolt.Tx) error {
+		c := tx.Bucket(s.store.logBucket).Cursor()
+		k, _ := c.First()
+		if k == nil {
+			idx = 0
+			return nil
+		}
+		idx = binary.BigEndian.Uint64(k)
+		return nil
+	})
+	return idx, err
+}
+
+func (s *boltLogStore) LastIndex() (uint64, error) {
+	var idx uint64
+	err := s.store.db.View(func(tx *bolt.Tx) error {
+		c := tx.Bucket(s.store.logBucket).Cursor()
+		k, _ := c.Last()
+		if k == nil {
+			idx = 0
+			return nil
+		}
+		idx = binary.BigEndian.Uint64(k)
+		return nil
+	})
+	return idx, err
+}
+
+func (s *boltLogStore) GetLog(index uint64, log *raft.Log) error {
+	err := s.store.db.View(func(tx *bolt.Tx) error {
+		raw := tx.Bucket(s.store.logBucket).Get(logKey(index))
+		if raw == nil {
+			return raft.ErrLogNotFound
+		}
+		return decodeLog(raw, log)
+	})
+	return err
+}
+
+func (s *boltLogStore) StoreLog(log *raft.Log) error {
+	return s.StoreLogs([]*raft.Log{log})
+}
+
+func (s *boltLogStore) StoreLogs(logs []*raft.Log) error {
+	if len(logs) == 0 {
+		return nil
+	}
+	return s.store.db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.store.logBucket)
+		for _, log := range logs {
+			raw, err := encodeLog(log)
+			if err != nil {
+				return err
+			}
+			if err := b.Put(logKey(log.Index), raw); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
+func (s *boltLogStore) DeleteRange(lo, hi uint64) error {
+	return s.store.db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.store.logBucket)
+		for i := lo; i <= hi; i++ {
+			if err := b.Delete(logKey(i)); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}

internal/workercluster/store_test.go

@@ -0,0 +1,173 @@
+package workercluster
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hashicorp/raft"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestStoreRoundTrip_Log writes a single log entry, reads it back, and
+// verifies every field matches. This is the contract the raft library
+// relies on.
+func TestStoreRoundTrip_Log(t *testing.T) {
+	s, err := NewBoltStore(t.TempDir())
+	require.NoError(t, err)
+	defer s.Close() //nolint:errcheck
+
+	store := s.LogStore()
+
+	in := &raft.Log{
+		Index:      5,
+		Term:       3,
+		Type:       raft.LogCommand,
+		Data:       []byte(`{"type":"config.adopt","adopted":{"version":1}}`),
+		AppendedAt: nowRFC3339(),
+	}
+	require.NoError(t, store.StoreLog(in))
+
+	got := &raft.Log{}
+	require.NoError(t, store.GetLog(5, got))
+	assert.Equal(t, in.Index, got.Index)
+	assert.Equal(t, in.Term, got.Term)
+	assert.Equal(t, in.Type, got.Type)
+	assert.Equal(t, in.Data, got.Data)
+	assert.True(t, in.AppendedAt.Equal(got.AppendedAt))
+}
+
+// TestStoreRoundTrip_FirstLastIndex exercises the index range after
+// StoreLog and DeleteRange.
+func TestStoreRoundTrip_FirstLastIndex(t *testing.T) {
+	s, err := NewBoltStore(t.TempDir())
+	require.NoError(t, err)
+	defer s.Close() //nolint:errcheck
+
+	store := s.LogStore()
+
+	first, err := store.FirstIndex()
+	require.NoError(t, err)
+	assert.EqualValues(t, 0, first)
+
+	last, err := store.LastIndex()
+	require.NoError(t, err)
+	assert.EqualValues(t, 0, last)
+
+	for i := uint64(10); i <= 20; i++ {
+		require.NoError(t, store.StoreLog(&raft.Log{Index: i, Term: 1, Type: raft.LogCommand, Data: []byte{byte(i)}}))
+	}
+
+	first, err = store.FirstIndex()
+	require.NoError(t, err)
+	assert.EqualValues(t, 10, first)
+	last, err = store.LastIndex()
+	require.NoError(t, err)
+	assert.EqualValues(t, 20, last)
+
+	// Delete a middle range.
+	require.NoError(t, store.DeleteRange(12, 15))
+
+	first, err = store.FirstIndex()
+	require.NoError(t, err)
+	assert.EqualValues(t, 10, first)
+	last, err = store.LastIndex()
+	require.NoError(t, err)
+	assert.EqualValues(t, 20, last, "LastIndex should still be the highest stored")
+
+	// The deleted slot should now report ErrLogNotFound.
+	err = store.GetLog(13, &raft.Log{})
+	assert.ErrorIs(t, err, raft.ErrLogNotFound)
+
+	// Surviving slot still readable.
+	got := &raft.Log{}
+	require.NoError(t, store.GetLog(11, got))
+	assert.Equal(t, byte(11), got.Data[0])
+}
+
+// TestStoreRoundTrip_StableStore checks the StableStore contract: Set /
+// Get / SetUint64 / GetUint64.
+func TestStoreRoundTrip_StableStore(t *testing.T) {
+	s, err := NewBoltStore(t.TempDir())
+	require.NoError(t, err)
+	defer s.Close() //nolint:errcheck
+
+	ss := s.StableStore()
+
+	require.NoError(t, ss.SetUint64([]byte("term"), 7))
+	require.NoError(t, ss.Set([]byte("voted_for"), []byte("w-1")))
+
+	term, err := ss.GetUint64([]byte("term"))
+	require.NoError(t, err)
+	assert.EqualValues(t, 7, term)
+
+	voted, err := ss.Get([]byte("voted_for"))
+	require.NoError(t, err)
+	assert.Equal(t, []byte("w-1"), voted)
+
+	// Unknown key returns zero value, not error.
+	missing, err := ss.GetUint64([]byte("nope"))
+	require.NoError(t, err)
+	assert.EqualValues(t, 0, missing)
+}
+
+// TestStoreRoundTrip_SnapshotStore_CreateOpenList exercises the
+// SnapshotStore lifecycle: Create, write, Close, then List + Open.
+func TestStoreRoundTrip_SnapshotStore_CreateOpenList(t *testing.T) {
+	s, err := NewBoltStore(t.TempDir())
+	require.NoError(t, err)
+	defer s.Close() //nolint:errcheck
+
+	ss := s.SnapshotStore()
+
+	cfg := raft.Configuration{Servers: []raft.Server{{ID: "w-1", Address: "127.0.0.1:1", Suffrage: raft.Voter}}}
+	sink, err := ss.Create(raft.SnapshotVersionMax, 100, 4, cfg, 50, nil)
+	require.NoError(t, err)
+
+	body := []byte(`{"hello":"world","config_count":3}`)
+	_, err = sink.Write(body)
+	require.NoError(t, err)
+	require.NoError(t, sink.Close())
+
+	list, err := ss.List()
+	require.NoError(t, err)
+	require.Len(t, list, 1)
+	assert.EqualValues(t, 100, list[0].Index)
+	assert.EqualValues(t, 4, list[0].Term)
+
+	meta, r, err := ss.Open(list[0].ID)
+	require.NoError(t, err)
+	defer r.Close() //nolint:errcheck
+
+	buf := make([]byte, len(body))
+	n, err := r.Read(buf)
+	require.NoError(t, err)
+	assert.Equal(t, len(body), n)
+	assert.Equal(t, body, buf[:n])
+
+	assert.EqualValues(t, 100, meta.Index)
+}
+
+// TestStoreRoundTrip_SnapshotStore_DeleteRangeAfterSnapshot ensures the
+// log store + snapshot store coexist on disk under the same bbolt file.
+func TestStoreRoundTrip_SnapshotStore_DeleteRangeAfterSnapshot(t *testing.T) {
+	s, err := NewBoltStore(t.TempDir())
+	require.NoError(t, err)
+	defer s.Close() //nolint:errcheck
+
+	ls := s.LogStore()
+	for i := uint64(1); i <= 50; i++ {
+		require.NoError(t, ls.StoreLog(&raft.Log{Index: i, Term: 1, Type: raft.LogCommand}))
+	}
+	require.NoError(t, ls.DeleteRange(1, 25))
+
+	first, err := ls.FirstIndex()
+	require.NoError(t, err)
+	assert.EqualValues(t, 26, first)
+}
+
+// nowRFC3339 returns a fixed-format RFC3339 timestamp for tests so
+// append log entries have a deterministic AppendedAt.
+func nowRFC3339() time.Time {
+	return time.Date(2026, 6, 27, 12, 0, 0, 0, time.UTC)
+}

internal/workercluster/testhelpers_test.go

@@ -0,0 +1,56 @@
+package workercluster
+
+import (
+	"bytes"
+	"encoding/json"
+	"testing"
+)
+
+// jsonMarshal marshals v to JSON or fails the test.
+func jsonMarshal(v interface{}) ([]byte, error) {
+	return json.Marshal(v)
+}
+
+// bytesBuffer is a minimal in-memory raft.SnapshotSink that captures
+// the bytes written to it. Used by snapshot Persist() in tests.
+type bytesBuffer struct {
+	bytes.Buffer
+}
+
+func (b *bytesBuffer) Close() error  { return nil }
+func (b *bytesBuffer) Cancel() error { return nil }
+func (b *bytesBuffer) ID() string    { return "test-snapshot" }
+
+// mustEntry encodes an Entry struct with the given field values into
+// JSON. It exists so the test bodies stay readable. All payload fields
+// are []byte; pass nil to leave them unset.
+func mustEntry(
+	t *testing.T,
+	typ EntryType,
+	setJSON, memberJSON, observeJSON, adoptedJSON, configJSON, stateJSON, outboxJSON []byte,
+) []byte {
+	t.Helper()
+
+	entry := Entry{
+		Type:    typ,
+		Set:     rawOrEmpty(setJSON),
+		Member:  rawOrEmpty(memberJSON),
+		Observe: rawOrEmpty(observeJSON),
+		Adopted: rawOrEmpty(adoptedJSON),
+		Report:  rawOrEmpty(configJSON),
+		State:   rawOrEmpty(stateJSON),
+		Outbox:  rawOrEmpty(outboxJSON),
+	}
+	raw, err := json.Marshal(entry)
+	if err != nil {
+		t.Fatalf("encode entry: %v", err)
+	}
+	return raw
+}
+
+func rawOrEmpty(b []byte) json.RawMessage {
+	if len(b) == 0 {
+		return nil
+	}
+	return json.RawMessage(b)
+}

internal/workercluster/transport.go

@@ -0,0 +1,287 @@
+package workercluster
+
+import (
+	"bytes"
+	"crypto/subtle"
+	"encoding/base64"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"sync"
+	"time"
+
+	rafthttp "github.com/CanonicalLtd/raft-http"
+)
+
+// HTTPCreds is the basic-auth credential pair workers use to gate the
+// rafthttp endpoint. The values come from distworker.HTTPConfig.Login
+// and distworker.HTTPConfig.Password (env vars WORKER_LOGIN and
+// WORKER_PASSWORD).
+type HTTPCreds struct {
+	Login    string
+	Password string
+}
+
+// IsConfigured returns true when both login and password are set.
+// NewTransport and NewHandler refuse to start otherwise.
+func (c HTTPCreds) IsConfigured() bool {
+	return c.Login != "" && c.Password != ""
+}
+
+// AuthDial returns a rafthttp.Dial function that wraps the inner
+// connection so the Authorization header is injected on every HTTP
+// request the rafthttp library writes over it.
+//
+// rafthttp builds its own http.Request without Authorization (GET for
+// the stream upgrade, POST/DELETE for membership changes). It then
+// writes the request to the net.Conn returned by Dial. We can't inject
+// at the request layer; we have to do it at the connection layer.
+//
+// Implementation: the wrapper buffers the first Write, looks for the
+// end-of-headers marker (\r\n\r\n), inserts an Authorization header
+// just before it, then forwards the augmented buffer plus any further
+// writes to the inner conn.
+//
+// TODO(security): replace with a rafthttp fork that supports an
+// outbound Authorization header or use NewDialTLS with mTLS client
+// certs once the worker identity model in plan section 7.1 lands.
+func AuthDial(inner rafthttp.Dial, creds HTTPCreds) rafthttp.Dial {
+	if !creds.IsConfigured() {
+		panic("workercluster: AuthDial requires both login and password")
+	}
+	return func(addr string, timeout time.Duration) (net.Conn, error) {
+		conn, err := inner(addr, timeout)
+		if err != nil {
+			return nil, err
+		}
+		return &authInjectingConn{
+			Conn: conn,
+			auth: "Basic " + base64.StdEncoding.EncodeToString([]byte(creds.Login+":"+creds.Password)),
+		}, nil
+	}
+}
+
+// authInjectingConn wraps a net.Conn and rewrites the first HTTP
+// request written to it so it carries an Authorization header.
+//
+// State machine:
+//
+//   - injected = false: incoming bytes are appended to buf until we see
+//     the end-of-headers marker (\r\n\r\n).
+//   - once we see \r\n\r\n, we insert the Authorization header just
+//     before it, drain the buffer to the inner conn, and switch to
+//     passthrough.
+//   - if too much data arrives without a header terminator (e.g. a
+//     very large POST body), we forward as-is; the auth handler will
+//     reject the request.
+//   - subsequent writes pass through unchanged.
+//
+// This is sufficient for rafthttp: every HTTP request it writes is a
+// self-contained, single-shot request over a fresh connection.
+type authInjectingConn struct {
+	net.Conn
+
+	auth string
+
+	mu       sync.Mutex
+	buf      []byte
+	injected bool
+}
+
+func (a *authInjectingConn) Write(p []byte) (int, error) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	if !a.injected {
+		a.buf = append(a.buf, p...)
+		if len(a.buf) > maxAuthHeaderBuffer {
+			// Too much data before we saw the header terminator;
+			// bail out and forward as-is. The request will be
+			// rejected by the auth handler on the other side,
+			// which is the correct failure mode.
+			a.injected = true
+			if _, err := a.Conn.Write(a.buf); err != nil {
+				return 0, err
+			}
+			a.buf = nil
+			return len(p), nil
+		}
+		if idx := bytes.Index(a.buf, []byte("\r\n\r\n")); idx >= 0 {
+			// Split around the header terminator.
+			head := a.buf[:idx]
+			rest := a.buf[idx:]
+			newBuf := make([]byte, 0, len(a.buf)+len(a.auth)+32)
+			newBuf = append(newBuf, head...)
+			newBuf = append(newBuf, []byte("\r\nAuthorization: ")...)
+			newBuf = append(newBuf, []byte(a.auth)...)
+			newBuf = append(newBuf, rest...)
+			a.buf = newBuf
+			a.injected = true
+			n, err := a.Conn.Write(a.buf)
+			a.buf = nil
+			return n, err
+		}
+		return len(p), nil
+	}
+	return a.Conn.Write(p)
+}
+
+// maxAuthHeaderBuffer caps the bytes we'll buffer waiting for the
+// header terminator. 64 KiB is well past any reasonable rafthttp
+// request and large enough to absorb the headers + a small body.
+const maxAuthHeaderBuffer = 64 * 1024
+
+// NewAuthHandler wraps an inner rafthttp.Handler with an HTTP basic-auth
+// check. Requests without matching credentials are rejected with 401
+// before the rafthttp path runs.
+//
+// The auth check uses crypto/subtle.ConstantTimeCompare to avoid timing
+// leaks on the credential comparison.
+func NewAuthHandler(inner *rafthttp.Handler, creds HTTPCreds, logger *log.Logger) http.Handler {
+	if logger == nil {
+		logger = log.Default()
+	}
+	if !creds.IsConfigured() {
+		// We panic on construction rather than at request time so a
+		// misconfigured worker fails fast at startup.
+		panic("workercluster: NewAuthHandler requires both login and password")
+	}
+
+	expectedUser := []byte(creds.Login)
+	expectedPass := []byte(creds.Password)
+
+	return &authHandler{inner: inner, expectedUser: expectedUser, expectedPass: expectedPass, logger: logger}
+}
+
+type authHandler struct {
+	inner        *rafthttp.Handler
+	expectedUser []byte
+	expectedPass []byte
+	logger       *log.Logger
+}
+
+func (a *authHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	user, pass, ok := r.BasicAuth()
+	if !ok ||
+		subtle.ConstantTimeCompare([]byte(user), a.expectedUser) != 1 ||
+		subtle.ConstantTimeCompare([]byte(pass), a.expectedPass) != 1 {
+		a.logger.Printf("[WARN] raft-http: rejected %s %s from %s: bad credentials",
+			r.Method, r.URL.Path, r.RemoteAddr)
+		w.Header().Set("WWW-Authenticate", `Basic realm="raft"`)
+		http.Error(w, "unauthorized", http.StatusUnauthorized)
+		return
+	}
+	a.inner.ServeHTTP(w, r)
+}
+
+// Unwrap exposes the inner rafthttp.Handler so callers like
+// unwrapRafthttpHandler can find it.
+func (a *authHandler) Unwrap() http.Handler { return a.inner }
+
+// NewTransport builds a rafthttp Layer + NetworkTransport pair bound to
+// the given listener. The returned Layer is ready to hand to
+// raft.NewNetworkTransport. Close must be called on shutdown to drain
+// the Layer's HTTP handler.
+//
+// dial is the rafthttp.Dial used to connect to peers; if nil, the
+// rafthttp.NewDialTCP default is used. handler is an http.Handler that
+// owns the rafthttp endpoint; production callers pass the auth wrapper
+// from NewAuthHandler. The inbound listener is started by the caller
+// because the listener needs to be running before peers can dial in.
+func NewTransport(
+	raftPath string,
+	listener net.Listener,
+	handler http.Handler,
+	dial rafthttp.Dial,
+	logOutput io.Writer,
+) (*rafthttp.Layer, *http.Server, error) {
+	if raftPath == "" {
+		raftPath = "/raft"
+	}
+	if listener == nil {
+		return nil, nil, fmt.Errorf("workercluster: listener is required")
+	}
+	if handler == nil {
+		return nil, nil, fmt.Errorf("workercluster: handler is required")
+	}
+	if dial == nil {
+		dial = rafthttp.NewDialTCP()
+	}
+	logger := log.New(logOutput, "[raft-http] ", log.LstdFlags)
+
+	realHandler, ok := unwrapRafthttpHandler(handler)
+	if !ok {
+		return nil, nil, fmt.Errorf("workercluster: handler must wrap a *rafthttp.Handler")
+	}
+	layer := rafthttp.NewLayerWithLogger(raftPath, listener.Addr(), realHandler, dial, logger)
+
+	server := &http.Server{
+		Handler:           handler,
+		ReadHeaderTimeout: 10 * time.Second,
+	}
+
+	return layer, server, nil
+}
+
+// unwrapRafthttpHandler walks a chain of http.Handler wrappers and
+// returns the *rafthttp.Handler at the bottom. Production wraps it in
+// NewAuthHandler; tests can wrap it in additional middleware.
+func unwrapRafthttpHandler(h http.Handler) (*rafthttp.Handler, bool) {
+	for {
+		switch v := h.(type) {
+		case *rafthttp.Handler:
+			return v, true
+		case interface{ Unwrap() http.Handler }:
+			h = v.Unwrap()
+		default:
+			return nil, false
+		}
+	}
+}
+
+// CheckBasicAuth is a small helper used by transport_test.go to confirm
+// the auth wrapper rejects bad creds and accepts good ones without
+// needing the rafthttp Library state.
+func CheckBasicAuth(h http.Handler, r *http.Request, login, password string) bool {
+	r.Header.Set("Authorization", basicAuthHeader(login, password))
+	rec := &recordingResponseWriter{header: http.Header{}}
+	h.ServeHTTP(rec, r)
+	return rec.status == http.StatusOK
+}
+
+// basicAuthHeader returns the value of an HTTP Basic Authorization
+// header for the given user/password pair. Exported for tests; production
+// code uses Go's r.BasicAuth() helper.
+func basicAuthHeader(user, pass string) string {
+	const prefix = "Basic "
+	value := user + ":" + pass
+	return prefix + base64Encode(value)
+}
+
+// base64Encode is a tiny indirection so tests do not import encoding/base64
+// directly; the production call sites use Go's standard library.
+func base64Encode(s string) string {
+	return base64.StdEncoding.EncodeToString([]byte(s))
+}
+
+// recordingResponseWriter is a minimal http.ResponseWriter for tests.
+type recordingResponseWriter struct {
+	header http.Header
+	body   []byte
+	status int
+}
+
+func (w *recordingResponseWriter) Header() http.Header { return w.header }
+func (w *recordingResponseWriter) Write(b []byte) (int, error) {
+	w.body = append(w.body, b...)
+	if w.status == 0 {
+		w.status = http.StatusOK
+	}
+	return len(b), nil
+}
+
+func (w *recordingResponseWriter) WriteHeader(status int) {
+	w.status = status
+}

internal/workercluster/transport_test.go

@@ -0,0 +1,133 @@
+package workercluster
+
+import (
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	rafthttp "github.com/CanonicalLtd/raft-http"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newLocalListener opens a TCP listener on the loopback interface
+// using an ephemeral port. Used by transport and e2e tests.
+func newLocalListener(t *testing.T) net.Listener {
+	t.Helper()
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	return l
+}
+
+// TestTransport_Auth_RejectsMissingCreds verifies that a request
+// without an Authorization header is rejected with 401.
+func TestTransport_Auth_RejectsMissingCreds(t *testing.T) {
+	h := rafthttp.NewHandler()
+	wrapped := NewAuthHandler(h, HTTPCreds{Login: "alice", Password: "secret"}, nil)
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest("GET", "/raft", nil)
+	wrapped.ServeHTTP(rec, req)
+
+	assert.Equal(t, http.StatusUnauthorized, rec.Code)
+	assert.Equal(t, `Basic realm="raft"`, rec.Header().Get("WWW-Authenticate"))
+}
+
+// TestTransport_Auth_RejectsWrongCreds verifies that a request with
+// incorrect credentials is rejected with 401.
+func TestTransport_Auth_RejectsWrongCreds(t *testing.T) {
+	h := rafthttp.NewHandler()
+	wrapped := NewAuthHandler(h, HTTPCreds{Login: "alice", Password: "secret"}, nil)
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest("GET", "/raft", nil)
+	req.SetBasicAuth("alice", "wrong")
+	wrapped.ServeHTTP(rec, req)
+
+	assert.Equal(t, http.StatusUnauthorized, rec.Code)
+}
+
+// TestTransport_Auth_AcceptsCorrectCreds verifies that a request with
+// matching credentials is forwarded to the inner rafthttp handler. The
+// rafthttp GET path expects an Upgrade header; without one it returns
+// 400, but the important point is that the auth wrapper does not block
+// the request before reaching the inner handler.
+func TestTransport_Auth_AcceptsCorrectCreds(t *testing.T) {
+	h := rafthttp.NewHandler()
+	wrapped := NewAuthHandler(h, HTTPCreds{Login: "alice", Password: "secret"}, nil)
+
+	rec := httptest.NewRecorder()
+	req := httptest.NewRequest("GET", "/raft", nil)
+	req.SetBasicAuth("alice", "secret")
+	wrapped.ServeHTTP(rec, req)
+
+	assert.NotEqual(t, http.StatusUnauthorized, rec.Code, "auth wrapper should not block good credentials")
+	assert.Equal(t, http.StatusBadRequest, rec.Code, "rafthttp expects Upgrade:raft header; with auth, 400 confirms we reached the inner handler")
+}
+
+// TestTransport_Auth_PanicsOnEmptyCreds verifies the wrapper refuses
+// to construct without both login and password.
+func TestTransport_Auth_PanicsOnEmptyCreds(t *testing.T) {
+	assert.Panics(t, func() {
+		NewAuthHandler(rafthttp.NewHandler(), HTTPCreds{Login: "", Password: ""}, nil)
+	})
+}
+
+// TestTransport_Auth_TimingSafe verifies that the wrapper uses
+// crypto/subtle.ConstantTimeCompare rather than ==. We assert this
+// indirectly: with wrong creds the response is always 401 regardless of
+// how close the password is to the real one.
+func TestTransport_Auth_TimingSafe(t *testing.T) {
+	h := rafthttp.NewHandler()
+	wrapped := NewAuthHandler(h, HTTPCreds{Login: "alice", Password: "secret"}, nil)
+
+	for _, pw := range []string{"s", "se", "sec", "secr", "secre", "secret", "secretX"} {
+		rec := httptest.NewRecorder()
+		req := httptest.NewRequest("GET", "/raft", nil)
+		req.SetBasicAuth("alice", pw)
+		wrapped.ServeHTTP(rec, req)
+		if pw == "secret" {
+			assert.NotEqual(t, http.StatusUnauthorized, rec.Code, "exact password should pass auth")
+		} else {
+			assert.Equal(t, http.StatusUnauthorized, rec.Code, "password %q should fail auth", pw)
+		}
+	}
+}
+
+// TestTransport_HTTPCreds_IsConfigured verifies the IsConfigured
+// contract: both set, or neither.
+func TestTransport_HTTPCreds_IsConfigured(t *testing.T) {
+	cases := []struct {
+		creds HTTPCreds
+		want  bool
+	}{
+		{HTTPCreds{Login: "u", Password: "p"}, true},
+		{HTTPCreds{Login: "u"}, false},
+		{HTTPCreds{Password: "p"}, false},
+		{HTTPCreds{}, false},
+	}
+	for _, tc := range cases {
+		assert.Equal(t, tc.want, tc.creds.IsConfigured(), "%+v", tc.creds)
+	}
+}
+
+// TestTransport_NewTransportRequiresArgs guards the constructor's
+// invariants.
+func TestTransport_NewTransportRequiresArgs(t *testing.T) {
+	h := rafthttp.NewHandler()
+	l := newLocalListener(t)
+
+	_, _, err := NewTransport("", l, nil, nil, nil)
+	require.Error(t, err)
+
+	_, _, err = NewTransport("", nil, h, nil, nil)
+	require.Error(t, err)
+
+	layer, srv, err := NewTransport("/raft", l, h, nil, nil)
+	require.NoError(t, err)
+	require.NotNil(t, layer)
+	require.NotNil(t, srv)
+	require.NoError(t, srv.Close())
+	require.NoError(t, l.Close())
+}

internal/workercluster/types.go

@@ -0,0 +1,138 @@
+// Package workercluster implements a worker-to-worker Raft consensus
+// cluster for the RSMon distributed worker subsystem.
+//
+// The package is an early subset of the plan in
+// docs/distributed/worker-to-worker-raft.md. It provides:
+//
+//   - A small in-memory FSM (CriticalCheckConfig list, membership cache,
+//     incident/outbox placeholders).
+//   - A bbolt-backed LogStore / StableStore / SnapshotStore.
+//   - An HTTP/WebSocket transport (rafthttp) with HTTP basic auth on
+//     both inbound and outbound connections.
+//   - A 3-voter cluster bootstrap / join / leave flow.
+//
+// Out of scope for this package (see // TODO(phase-N): comments in the
+// relevant files): the distributed_critical check kind itself, signed
+// config adoption, envelope-encrypted credentials, snapshot encryption,
+// and the external central witness.
+package workercluster
+
+// Worker role strings used in Member.Role and ObserverSet.Voters /
+// Observers. Centralized so lints (goconst) and callers agree on the
+// canonical spelling.
+const (
+	RoleVoter            = "voter"
+	RoleObserver         = "observer"
+	RoleVoterAndObserver = "voter+observer"
+)
+
+// CriticalCheckConfig is one adopted critical check entry. The cluster
+// runs every entry from this list on every voter+observer worker.
+//
+// TODO(phase-1): wire this into the checks/c* dispatcher once the
+// distributed_critical kind lands.
+type CriticalCheckConfig struct {
+	ID         int64  `json:"id"`
+	MonitorID  int64  `json:"monitor_id"`
+	Kind       string `json:"kind"`
+	Target     string `json:"target"`
+	IntervalS  int    `json:"interval_seconds"`
+	Epoch      int64  `json:"epoch"` // monotonic per config_version
+	ConfigHash string `json:"config_hash"`
+}
+
+// IncidentState is the per-check incident lifecycle. The values are the
+// names used by the plan section 9.1.
+//
+// TODO(phase-1): implement the hysteresis + flap_suppression rules
+// described in section 8.5 of the plan.
+type IncidentState struct {
+	CheckID         int64  `json:"check_id"`
+	State           string `json:"state"` // clear | observing | open | resolving
+	OpenedAtIndex   uint64 `json:"opened_at_index,omitempty"`
+	LastConfirmAtIx uint64 `json:"last_confirm_at_index,omitempty"`
+	LastOKAtIx      uint64 `json:"last_ok_at_index,omitempty"`
+	Confirmations   int    `json:"confirmations"`
+}
+
+// OutboxMeta is the metadata-only notification outbox entry. The FSM
+// keeps only metadata; provider delivery happens outside Raft safety.
+//
+// TODO(phase-1): wire the outbox entries to sender/ once the commit-
+// before-notify flow in section 9.4 lands.
+type OutboxMeta struct {
+	Seq        uint64 `json:"seq"`
+	IncidentID int64  `json:"incident_id"`
+	TenantID   int64  `json:"tenant_id"`
+	Channel    string `json:"channel"`
+	ContactRef string `json:"contact_ref"`
+	DedupKey   string `json:"dedup_key"`
+	State      string `json:"state"` // pending | sent | failed | dead
+	Attempts   int    `json:"attempts"`
+	LastError  string `json:"last_error,omitempty"`
+}
+
+// Member is one worker in the cluster. The Membership cache mirrors what
+// the raft library itself stores, but is denormalized here so callers can
+// read the current set without going through raft.GetConfiguration.
+//
+// TODO(phase-1): fold in region_code + build_version + drain_state.
+type Member struct {
+	WorkerID    string `json:"worker_id"`
+	WorkerURL   string `json:"worker_url"`
+	RaftAddress string `json:"raft_address"`
+	Role        string `json:"role"` // voter | observer | voter+observer
+	JoinedAtIx  uint64 `json:"joined_at_index"`
+	LastSeenIx  uint64 `json:"last_seen_index,omitempty"`
+	BuildVer    string `json:"build_version,omitempty"`
+}
+
+// ObserverSet is the versioned observer-set object described in plan
+// section 6.5. The hash is a placeholder for the adoption check; the
+// real signed-config adoption flow lands in phase 1.
+//
+// TODO(phase-1): replace Hash with a real content hash + Ed25519
+// signature once section 11.3 lands.
+type ObserverSet struct {
+	Version       uint64   `json:"version"`
+	ConfigVersion uint64   `json:"config_version"`
+	AdoptedAtIx   uint64   `json:"adopted_at_index"`
+	Voters        []string `json:"voters"`
+	Observers     []string `json:"observers"`
+	Hash          string   `json:"hash"`
+}
+
+// WorkerDiagnostics is a compact per-worker health snapshot. The FSM
+// stores the most recent report for each worker.
+//
+// TODO(phase-1): publish these to VictoriaMetrics via internal/influx
+// once section 15 of the plan lands.
+type WorkerDiagnostics struct {
+	WorkerID    string `json:"worker_id"`
+	DiskFreePct int    `json:"disk_free_pct"`
+	ClockSkewMs int64  `json:"clock_skew_ms"`
+	LastBeatIx  uint64 `json:"last_beat_index,omitempty"`
+}
+
+// PartitionState mirrors section 10.2 of the plan. The cluster writes
+// its current view via partition.report entries.
+//
+// TODO(phase-1): wire the external central witness (section 10.3) into
+// this object.
+type PartitionState struct {
+	State     string `json:"state"` // steady | degraded | partitioned | healing | ...
+	UpdatedIx uint64 `json:"updated_at_index"`
+	Reason    string `json:"reason,omitempty"`
+}
+
+// CentralWitnessReport is the last external witness view. Empty until
+// the control plane starts pushing reports.
+//
+// TODO(phase-1): accept incoming witness reports from the control plane
+// over the existing worker websocket; section 10.3 of the plan.
+type CentralWitnessReport struct {
+	ReachableVoters    []string `json:"reachable_voters"`
+	ReachableObservers []string `json:"reachable_observers"`
+	SplitBrain         bool     `json:"split_brain_detected"`
+	ReportedAtIx       uint64   `json:"reported_at_index"`
+}