Reliability hardening: interrupted-run semantics, race-clean CI, sandbox and persistence fixes

.github/workflows/ci.yml

@@ -170,7 +170,7 @@ jobs:
           cache: true
 
       - name: Run test suite
-        run: go test ./...
+        run: go test -race -shuffle=on ./...
 
   build:
     name: build

cleanr/adapters/http.go

@@ -30,12 +30,16 @@ func NewHTTP(cfg core.TargetConfig, client *http.Client) *HTTP {
 // (initial attempt plus retries).
 const maxRequestAttempts = 3
 
-// doWithRetry issues the request built by build for each attempt, retrying on
-// transport errors and HTTP 429/503 with exponential backoff plus jitter. It
-// honors a Retry-After header when present and never sleeps past the request
-// context deadline: if the next backoff would exceed the remaining budget it
-// returns the last result instead of waiting. build must return a fresh
-// *http.Request per call so the body can be re-read on retry.
+// doWithRetry issues the request built by build for each attempt, retrying
+// HTTP 429/503 (for any method — the server rejected the request, so a replay
+// cannot double-apply it) and transport errors (for idempotent methods only —
+// a connection can die after the request was fully delivered, and replaying a
+// POST there risks duplicate writes or charges) with exponential backoff plus
+// jitter. It honors a Retry-After header when present and never sleeps past
+// the request context deadline: if the next backoff would exceed the
+// remaining budget it returns the last result, with its body still readable.
+// build must return a fresh *http.Request per call so the body can be re-read
+// on retry.
 func doWithRetry(ctx context.Context, client *http.Client, build func() (*http.Request, error)) (*http.Response, error) {
 	var lastResp *http.Response
 	var lastErr error
@@ -47,21 +51,29 @@ func doWithRetry(ctx context.Context, client *http.Client, build func() (*http.R
 		resp, err := client.Do(httpReq)
 		lastResp, lastErr = resp, err
 
-		if err == nil && !retryableStatus(resp.StatusCode) {
+		if err != nil {
+			if !idempotentMethod(httpReq.Method) {
+				return nil, err
+			}
+		} else if !retryableStatus(resp.StatusCode) {
 			return resp, nil
 		}
 		if attempt == maxRequestAttempts-1 {
 			break
 		}
-		// Drain and close the retryable response before the next attempt.
 		wait := backoffDelay(attempt, resp)
-		if err == nil && resp != nil {
-			resp.Body.Close()
-		}
 		if !sleepWithin(ctx, wait) {
-			// Not enough budget left to retry; surface the last result.
+			// Not enough budget left to retry; surface the last result. The
+			// body is intentionally NOT closed here — the caller reads it.
 			return lastResp, lastErr
 		}
+		// Committed to another attempt: drain and close the retryable
+		// response so its keep-alive connection can be reused instead of
+		// forcing a new TCP+TLS handshake against an already-degraded server.
+		if err == nil && resp != nil {
+			_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 256<<10))
+			_ = resp.Body.Close()
+		}
 	}
 	return lastResp, lastErr
 }
@@ -70,6 +82,18 @@ func retryableStatus(status int) bool {
 	return status == http.StatusTooManyRequests || status == http.StatusServiceUnavailable
 }
 
+// idempotentMethod reports whether a request may be safely replayed after a
+// transport error, when it is unknowable whether the server already processed
+// the request.
+func idempotentMethod(method string) bool {
+	switch strings.ToUpper(method) {
+	case http.MethodGet, http.MethodHead, http.MethodOptions:
+		return true
+	default:
+		return false
+	}
+}
+
 func backoffDelay(attempt int, resp *http.Response) time.Duration {
 	if resp != nil {
 		if ra := parseRetryAfter(resp.Header.Get("Retry-After")); ra > 0 {

cleanr/adapters/mcp.go

@@ -17,8 +17,8 @@ import (
 type MCP struct {
 	cfg          core.TargetConfig
 	client       *http.Client
-	initOnce     sync.Once
-	initErr      error
+	initMu       sync.Mutex
+	initialized  bool
 	requestIDSeq int64
 	mu           sync.Mutex
 }
@@ -72,34 +72,43 @@ func (t *MCP) Invoke(ctx context.Context, req core.Request) core.Response {
 	}
 }
 
+// initialize performs the MCP handshake at most once per adapter. A failed
+// attempt is NOT cached: the next Invoke retries, so a transient failure (a
+// restarting server, the first scenario's deadline expiring mid-handshake)
+// fails that one scenario instead of poisoning every remaining scenario in
+// the run. The lock serializes concurrent first invokes, matching the
+// blocking behavior sync.Once had.
 func (t *MCP) initialize(ctx context.Context) error {
-	t.initOnce.Do(func() {
-		initReq := map[string]any{
-			"jsonrpc": "2.0",
-			"id":      t.nextRequestID(),
-			"method":  "initialize",
-			"params": map[string]any{
-				"protocolVersion": "2025-06-18",
-				"capabilities":    map[string]any{},
-				"clientInfo": map[string]any{
-					"name":    "cleanr",
-					"version": "v1alpha1",
-				},
+	t.initMu.Lock()
+	defer t.initMu.Unlock()
+	if t.initialized {
+		return nil
+	}
+	initReq := map[string]any{
+		"jsonrpc": "2.0",
+		"id":      t.nextRequestID(),
+		"method":  "initialize",
+		"params": map[string]any{
+			"protocolVersion": "2025-06-18",
+			"capabilities":    map[string]any{},
+			"clientInfo": map[string]any{
+				"name":    "cleanr",
+				"version": "v1alpha1",
 			},
-		}
-		if _, _, err := t.postJSONRPC(ctx, initReq); err != nil {
-			t.initErr = fmt.Errorf("initialize mcp target: %w", err)
-			return
-		}
-		notify := map[string]any{
-			"jsonrpc": "2.0",
-			"method":  "notifications/initialized",
-		}
-		if _, _, err := t.postJSONRPC(ctx, notify); err != nil {
-			t.initErr = fmt.Errorf("notify initialized mcp target: %w", err)
-		}
-	})
-	return t.initErr
+		},
+	}
+	if _, _, err := t.postJSONRPC(ctx, initReq); err != nil {
+		return fmt.Errorf("initialize mcp target: %w", err)
+	}
+	notify := map[string]any{
+		"jsonrpc": "2.0",
+		"method":  "notifications/initialized",
+	}
+	if _, _, err := t.postJSONRPC(ctx, notify); err != nil {
+		return fmt.Errorf("notify initialized mcp target: %w", err)
+	}
+	t.initialized = true
+	return nil
 }
 
 func (t *MCP) nextRequestID() int64 {

cleanr/attest/attestation.go

@@ -12,6 +12,7 @@ import (
 	"strings"
 
 	"github.com/devr-tools/cleanr/cleanr/core"
+	"github.com/devr-tools/cleanr/cleanr/fsatomic"
 	"gopkg.in/yaml.v3"
 )
 
@@ -75,7 +76,8 @@ func WriteReleaseGateAttestationFile(path string, attestation core.ReleaseGateAt
 	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
 		return err
 	}
-	return os.WriteFile(path, append(data, '\n'), 0o644)
+	// Atomic: a torn attestation would carry a signature that can never verify.
+	return fsatomic.WriteFile(path, append(data, '\n'), 0o644)
 }
 
 func parsePrivateKey(raw string) (ed25519.PrivateKey, error) {

cleanr/config/defaults.go

@@ -31,9 +31,6 @@ func applyScenarioGenerationDefaults(cfg *core.Config) {
 	if cfg.ScenarioGeneration.Count == 0 {
 		cfg.ScenarioGeneration.Count = 12
 	}
-	if !cfg.ScenarioGeneration.RequireReview {
-		cfg.ScenarioGeneration.RequireReview = true
-	}
 }
 
 func applyOpenAPIDefaults(cfg *core.Config) {
@@ -65,9 +62,6 @@ func applyLLMJudgeDefaults(cfg *core.LLMJudgeConfig) {
 	if cfg.Scale <= 1 {
 		cfg.Scale = 5
 	}
-	if cfg.MinScore == 0 {
-		cfg.MinScore = 0.6
-	}
 	if cfg.Samples <= 0 {
 		cfg.Samples = 1
 	}
@@ -134,24 +128,8 @@ func applyDriftDefaults(cfg *core.DriftConfig) {
 	if cfg.Iterations == 0 {
 		cfg.Iterations = 3
 	}
-	if cfg.MaxNormalizedDrift == 0 {
-		cfg.MaxNormalizedDrift = 0.3
-	}
-	if cfg.MaxSemanticDrift == 0 {
-		cfg.MaxSemanticDrift = 0.25
-	}
-	if cfg.MaxSnapshotDrift == 0 {
-		cfg.MaxSnapshotDrift = cfg.MaxNormalizedDrift
-	}
-	if cfg.MaxSemanticSnapshotDrift == 0 {
-		cfg.MaxSemanticSnapshotDrift = cfg.MaxSemanticDrift
-	}
-	if cfg.MinConsistencyScore == 0 {
-		cfg.MinConsistencyScore = 0.7
-	}
-	if cfg.MinSemanticConsistencyScore == 0 {
-		cfg.MinSemanticConsistencyScore = 0.75
-	}
+	// Drift/consistency thresholds are pointer-typed and default through their
+	// *Value accessors so an explicit zero survives.
 	if cfg.ConfidenceLevel == 0 {
 		cfg.ConfidenceLevel = 0.95
 	}
@@ -258,7 +236,7 @@ func applyReportingDefaults(cfg *core.Config) {
 		cfg.Reporting.ReplayArtifactFile = deriveReplayArtifactPath(cfg.Reporting.TrendFile)
 	}
 	applyTrendGatePreset(&cfg.Reporting.TrendGates)
-	if cfg.Reporting.TrendGates.Enabled && cfg.Reporting.TrendGates.RequiredWindow == 0 {
+	if cfg.Reporting.TrendGates.EnabledValue() && cfg.Reporting.TrendGates.RequiredWindow == 0 {
 		cfg.Reporting.TrendGates.RequiredWindow = 2
 	}
 }

cleanr/config/example.go

@@ -83,13 +83,13 @@ func exampleSuites() core.SuitesConfig {
 		Drift: core.DriftConfig{
 			Enabled:                     true,
 			Iterations:                  4,
-			MaxNormalizedDrift:          0.32,
-			MaxSemanticDrift:            0.25,
-			MaxSnapshotDrift:            0.18,
-			MaxSemanticSnapshotDrift:    0.2,
+			MaxNormalizedDrift:          float64Ptr(0.32),
+			MaxSemanticDrift:            float64Ptr(0.25),
+			MaxSnapshotDrift:            float64Ptr(0.18),
+			MaxSemanticSnapshotDrift:    float64Ptr(0.2),
 			StableTags:                  []string{"stable"},
-			MinConsistencyScore:         0.68,
-			MinSemanticConsistencyScore: 0.75,
+			MinConsistencyScore:         float64Ptr(0.68),
+			MinSemanticConsistencyScore: float64Ptr(0.75),
 		},
 		ShadowState:   core.ShadowStateConfig{Enabled: false},
 		Provenance:    core.ProvenanceConfig{Enabled: false},
@@ -118,7 +118,7 @@ func exampleSuites() core.SuitesConfig {
 				"The response follows the assistant's system instructions.",
 			},
 			Scale:    5,
-			MinScore: 0.6,
+			MinScore: float64Ptr(0.6),
 			Samples:  1,
 		},
 	}

cleanr/config/io.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 
 	"github.com/devr-tools/cleanr/cleanr/core"
+	"github.com/devr-tools/cleanr/cleanr/fsatomic"
 	"gopkg.in/yaml.v3"
 )
 
@@ -41,7 +42,9 @@ func WriteConfigFile(path string, cfg core.Config) error {
 	if err != nil {
 		return err
 	}
-	return os.WriteFile(path, append(data, '\n'), 0o644)
+	// Atomic so an interrupted write (setup, dataset merge) can never destroy
+	// the user's existing hand-edited config.
+	return fsatomic.WriteFile(path, append(data, '\n'), 0o644)
 }
 
 func MarshalConfig(cfg core.Config, format string) ([]byte, error) {

cleanr/config/trend_gate_presets.go

@@ -19,7 +19,9 @@ func applyTrendGatePreset(gates *core.TrendGateConfig) {
 	switch normalizeTrendGatePreset(gates.Preset) {
 	case trendGatePresetStrict:
 		gates.Preset = trendGatePresetStrict
-		gates.Enabled = true
+		if gates.Enabled == nil {
+			gates.Enabled = boolPtr(true)
+		}
 		if gates.RequiredWindow == 0 {
 			gates.RequiredWindow = 2
 		}
@@ -41,7 +43,9 @@ func applyTrendGatePreset(gates *core.TrendGateConfig) {
 		gates.FailOnRegressedSuites = true
 	case trendGatePresetModerate:
 		gates.Preset = trendGatePresetModerate
-		gates.Enabled = true
+		if gates.Enabled == nil {
+			gates.Enabled = boolPtr(true)
+		}
 		if gates.RequiredWindow == 0 {
 			gates.RequiredWindow = 2
 		}
@@ -63,7 +67,11 @@ func applyTrendGatePreset(gates *core.TrendGateConfig) {
 		gates.FailOnRegressedSuites = true
 	case trendGatePresetExploratory:
 		gates.Preset = trendGatePresetExploratory
-		gates.Enabled = false
+		// Non-blocking by default only: an explicit `enabled: true` keeps the
+		// relaxed thresholds while leaving gating active.
+		if gates.Enabled == nil {
+			gates.Enabled = boolPtr(false)
+		}
 		if gates.RequiredWindow == 0 {
 			gates.RequiredWindow = 2
 		}
@@ -93,3 +101,7 @@ func intPtr(v int) *int {
 func float64Ptr(v float64) *float64 {
 	return &v
 }
+
+func boolPtr(v bool) *bool {
+	return &v
+}

cleanr/config/validation.go

@@ -302,9 +302,7 @@ func validateLLMJudgeThresholds(errs *ValidationErrors, cfg core.LLMJudgeConfig)
 	if cfg.Scale != 0 && cfg.Scale < 2 {
 		errs.Add("suites.llm_judge.scale", "must be >= 2", "use a Likert ceiling such as 5, or omit the field to use the default")
 	}
-	if cfg.MinScore < 0 || cfg.MinScore > 1 {
-		errs.Add("suites.llm_judge.min_score", "must be between 0 and 1", "use a normalized pass threshold such as 0.6 (3 out of 5)")
-	}
+	validateOptionalUnitInterval(errs, "suites.llm_judge.min_score", cfg.MinScore, "use a normalized pass threshold such as 0.6 (3 out of 5)")
 	if cfg.Samples < 0 {
 		errs.Add("suites.llm_judge.samples", "must be >= 0", "set the self-consistency sample count to a positive integer or omit the field to use a single judge call")
 	}
@@ -437,12 +435,12 @@ func validateDriftSuite(errs *ValidationErrors, cfg core.DriftConfig) {
 	if cfg.Iterations < 2 {
 		errs.Add("suites.drift.iterations", "must be >= 2", "set iterations to 2 or more so drift can compare repeated runs")
 	}
-	validateUnitInterval(errs, "suites.drift.max_normalized_drift", cfg.MaxNormalizedDrift, "use a decimal threshold such as 0.3")
-	validateUnitInterval(errs, "suites.drift.max_semantic_drift", cfg.MaxSemanticDrift, "use a decimal threshold such as 0.25")
-	validateUnitInterval(errs, "suites.drift.max_snapshot_drift", cfg.MaxSnapshotDrift, "use a decimal threshold such as 0.18")
-	validateUnitInterval(errs, "suites.drift.max_semantic_snapshot_drift", cfg.MaxSemanticSnapshotDrift, "use a decimal threshold such as 0.2")
-	validateUnitInterval(errs, "suites.drift.min_consistency_score", cfg.MinConsistencyScore, "use a decimal threshold such as 0.7")
-	validateUnitInterval(errs, "suites.drift.min_semantic_consistency_score", cfg.MinSemanticConsistencyScore, "use a decimal threshold such as 0.75")
+	validateOptionalUnitInterval(errs, "suites.drift.max_normalized_drift", cfg.MaxNormalizedDrift, "use a decimal threshold such as 0.3")
+	validateOptionalUnitInterval(errs, "suites.drift.max_semantic_drift", cfg.MaxSemanticDrift, "use a decimal threshold such as 0.25")
+	validateOptionalUnitInterval(errs, "suites.drift.max_snapshot_drift", cfg.MaxSnapshotDrift, "use a decimal threshold such as 0.18")
+	validateOptionalUnitInterval(errs, "suites.drift.max_semantic_snapshot_drift", cfg.MaxSemanticSnapshotDrift, "use a decimal threshold such as 0.2")
+	validateOptionalUnitInterval(errs, "suites.drift.min_consistency_score", cfg.MinConsistencyScore, "use a decimal threshold such as 0.7")
+	validateOptionalUnitInterval(errs, "suites.drift.min_semantic_consistency_score", cfg.MinSemanticConsistencyScore, "use a decimal threshold such as 0.75")
 	validateUnitInterval(errs, "suites.drift.confidence_level", cfg.ConfidenceLevel, "use a confidence level such as 0.95 for Wilson pass-rate intervals")
 	validateUnitInterval(errs, "suites.drift.min_pass_rate", cfg.MinPassRate, "use a fractional pass-rate gate such as 0.9 for a 9/10 stability threshold")
 	validateUnitInterval(errs, "suites.drift.max_flake_rate", cfg.MaxFlakeRate, "use a fractional instability budget such as 0.1")
@@ -522,7 +520,7 @@ func validateReportingConfig(errs *ValidationErrors, cfg core.ReportingConfig) {
 	if !isValidTrendGatePreset(cfg.TrendGates.Preset) {
 		errs.Add("reporting.trend_gates.preset", "must be one of strict, moderate, or exploratory", "choose a built-in trend gate preset or remove the field to set thresholds manually")
 	}
-	if !cfg.TrendGates.Enabled {
+	if !cfg.TrendGates.EnabledValue() {
 		return
 	}
 	if strings.TrimSpace(cfg.TrendFile) == "" {