feat: classify accepted-blocking refusal eligibility

docs/invariants.md

@@ -418,6 +418,17 @@ Each refusal is a preflight **error with a stated reason** — never a warning,
   fields; the classification registry (`pkg/plan/refusal.go`, `pkg/migrate/refusal_registry.go`)
   is checked for completeness against the production closed sets by
   `TestRefusalRegistryIsComplete`. *Source:* [refusal classes](refusal-classes.md).
+- **RF-8** — An in-process refused verdict retains its full typed refusal proof, and
+  `Verdict.Refusal` hands that proof out only when it validates under RF-7 and still matches
+  the verdict's exported refusal fields. A verdict decoded from JSON reconstructs class,
+  reason, owner, and cause, but cannot recover the unexported refusal site: the site-keyed
+  eligibility row fails closed for a decoded verdict, while the cause-keyed row is decidable
+  from the wire fields. Accepted-blocking eligibility is therefore consumed only from the
+  proof of the verdict the same front-door invocation produced, never from a decoded verdict.
+  *Enforced:* `Verdict.WithRefusal`, `Verdict.Refusal`,
+  `TestRefusalRejectsUnvalidatedProof`, `TestRefusalRejectsFieldsDivergingFromProof`, and
+  `TestGateVerdictAcceptedBlockingEligibility`, which pins that the site-keyed row is eligible
+  in process and ineligible after a JSON round trip. *Source:* [lock-budgeted passthrough](lock-budgeted-passthrough.md).
 
 ## Orchestration / control-plane (OC)
 

docs/lock-budgeted-passthrough.md

@@ -102,6 +102,17 @@ it is acquired.
 | 6 | Eligibility is selected from a closed registry keyed by typed class, reason, and cause or refusal site. No renderer text or SQL substring participates in the decision. |
 | 7 | A dry-run reports the original refusal and a per-statement `blocking_passthrough_eligible` boolean, but executes nothing even when the flag is present. |
 
+The full typed refusal proof is available while its verdict remains in process, and
+`Refusal()` returns it only when it validates as a classified refusal and still matches the
+verdict's exported fields. JSON carries class, reason, owner, and cause, but not the internal
+refusal site. Calling `Refusal()` on a decoded verdict therefore reconstructs the JSON fields
+but cannot restore the site: the site-keyed row (single-relation `index-statement`) fails
+closed after decoding, while the cause-keyed row (`unsupported-partitioned-parent` with
+`parent-blocking-index-build`) is decidable from the wire fields. Decoding is not the
+fail-closed boundary; the front door is. Eligibility is consumed only from the proof of the
+verdict the same `migrate` or `diff` invocation produced, never from a verdict read back
+from JSON.
+
 “Prints before execution” is an ordering requirement for human output and an information
 requirement for JSON. Human mode prints the refusal analysis, then a separate acceptance line,
 then starts the session. JSON remains one final machine-readable object; its executed verdict
@@ -373,7 +384,7 @@ Sequence implementation as follows:
    cause or refusal site. The `index-statement` site distinguishes the single-relation forms
    from `DROP INDEX a, b` and `REINDEX SCHEMA`, `DATABASE`, and `SYSTEM`, so the registry can
    admit the former and refuse the latter without a database. Its completeness tests make new
-   values ineligible by default and prove that render text is never consulted.
+   values ineligible by default and prove that render text is never consulted. *(done)*
 2. Add the executor path through engine-owned bounded sessions, requiring explicit non-zero
    `statement_timeout` and non-zero `lock_timeout`. At this step add lock-budget invariants to
    [the invariant registry](invariants.md): every passthrough statement runs in an

internal/cli/color_test.go

@@ -182,6 +182,9 @@ func fullVerdict(t *testing.T) verdict.Verdict {
 	}
 	rv := reflect.ValueOf(v)
 	for i := range rv.NumField() {
+		if rv.Type().Field(i).PkgPath != "" {
+			continue
+		}
 		require.False(t, rv.Field(i).IsZero(),
 			"Verdict field %s is zero in the all-fields fixture; set it so the renderer parity lock covers it",
 			rv.Type().Field(i).Name)

pkg/migrate/refusal_registry.go

@@ -102,15 +102,19 @@ func siteRefusals() []siteRefusal {
 // (ALTER TABLE, CREATE INDEX). concurrent distinguishes the already-safe
 // maintenance forms — which pg-sprite need not wrap — from the plain forms
 // it refuses in favor of their concurrent idiom.
-func gateRefusal(kind statement.Kind, concurrent bool) (verdict.Refusal, bool) {
+func gateRefusal(kind statement.Kind, concurrent bool, target statement.IndexTarget) (verdict.Refusal, bool) {
 	switch kind {
 	case statement.KindAlterTable, statement.KindCreateIndex:
 		return verdict.Refusal{}, false
 	case statement.KindDropIndex, statement.KindReindex:
 		if concurrent {
 			return verdict.NoOnlineSafetyProblem(verdict.ReasonIndexStatement, verdict.OwnerDirectOperator), true
 		}
-		return verdict.ByDesign(verdict.ReasonIndexStatement), true
+		site := verdict.RefusalSiteIndexOther
+		if target == statement.IndexTargetSingleRelation {
+			site = verdict.RefusalSiteIndexSingleRelation
+		}
+		return verdict.ByDesign(verdict.ReasonIndexStatement).WithSite(site), true
 	case statement.KindCreateTable:
 		return verdict.NoOnlineSafetyProblem(verdict.ReasonUnsupportedStatement, verdict.OwnerDeclarativeFrontDoor), true
 	case statement.KindDataChange:
@@ -201,6 +205,48 @@ func partitionRefusal(cause preflight.PartitionRefusalCause) (verdict.Refusal, b
 	return plan.PartitionRefusal(cause)
 }
 
+// AcceptedBlockingEligible reports whether a typed refusal is in the closed
+// accepted-blocking registry. Unknown combinations fail closed.
+func AcceptedBlockingEligible(r verdict.Refusal) bool {
+	eligible, decided := acceptedBlockingDecision(r)
+	return decided && eligible
+}
+
+// acceptedBlockingDecision is total over the registered reason and
+// cause/site vocabulary. The bool pair is eligibility and whether the key
+// has an explicit decision; completeness tests reject undecided additions.
+func acceptedBlockingDecision(r verdict.Refusal) (bool, bool) {
+	switch r.Reason() {
+	case verdict.ReasonIndexStatement:
+		switch r.Site() {
+		case verdict.RefusalSiteIndexSingleRelation:
+			return r.Class() == verdict.ClassByDesign, true
+		case verdict.RefusalSiteIndexOther, "":
+			return false, true
+		default:
+			return false, false
+		}
+	case verdict.ReasonUnsupportedPartitionedParent:
+		switch r.Cause() {
+		case verdict.CauseParentBlockingIndexBuild:
+			return r.Class() == verdict.ClassCapabilityBoundary, true
+		case verdict.CauseParentConcurrentIndexBuild, verdict.CauseParentIndexAdoption,
+			verdict.CauseParentNotValidForeignKey:
+			return false, true
+		default:
+			return false, false
+		}
+	case verdict.ReasonUnsupportedStatement, verdict.ReasonTableTooLarge,
+		verdict.ReasonInsufficientPrivileges, verdict.ReasonBudgetExceeded,
+		verdict.ReasonRewriteRequired, verdict.ReasonBackendUnavailable,
+		verdict.ReasonDestructiveChange, verdict.ReasonPlanFingerprintMismatch,
+		verdict.ReasonCreateCollision:
+		return false, true
+	default:
+		return false, false
+	}
+}
+
 // isInSentinelSet reports whether err matches any sentinel in set.
 func isInSentinelSet(err error, set []error) bool {
 	for _, sentinel := range set {

pkg/migrate/refusal_registry_test.go

@@ -1,6 +1,7 @@
 package migrate
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"testing"
@@ -35,7 +36,7 @@ func deriveRefusalKeys() (keys []classifiedKey, admitted []string) {
 		keys = append(keys, classifiedKey{"create-shape:" + string(c), r, ok})
 	}
 	for _, c := range preflight.PartitionRefusalCauses() {
-		r, ok := plan.PartitionRefusal(c)
+		r, ok := partitionRefusal(c)
 		keys = append(keys, classifiedKey{"partition:" + string(c), r, ok})
 	}
 	for _, err := range admissionSentinels() {
@@ -48,13 +49,22 @@ func deriveRefusalKeys() (keys []classifiedKey, admitted []string) {
 	}
 	for _, k := range statement.Kinds() {
 		for _, concurrent := range []bool{false, true} {
-			r, ok := gateRefusal(k, concurrent)
-			key := fmt.Sprintf("gate:%s/concurrent=%t", k, concurrent)
-			if !ok {
-				admitted = append(admitted, key)
-				continue
+			targets := []statement.IndexTarget{statement.IndexTargetNone}
+			if !concurrent && (k == statement.KindDropIndex || k == statement.KindReindex) {
+				targets = statement.IndexTargets()
+			}
+			for _, target := range targets {
+				r, ok := gateRefusal(k, concurrent, target)
+				key := fmt.Sprintf("gate:%s/concurrent=%t", k, concurrent)
+				if len(targets) > 1 {
+					key = fmt.Sprintf("%s/target=%s", key, target)
+				}
+				if !ok {
+					admitted = append(admitted, key)
+					continue
+				}
+				keys = append(keys, classifiedKey{key, r, ok})
 			}
-			keys = append(keys, classifiedKey{key, r, ok})
 		}
 	}
 	for _, s := range siteRefusals() {
@@ -105,6 +115,9 @@ func TestRefusalRegistryIsComplete(t *testing.T) {
 			// and owner rule.
 			_, err := verdict.NewRefusal(k.refusal.Class(), k.refusal.Reason(), k.refusal.Owner())
 			require.NoError(t, err)
+			_, decided := acceptedBlockingDecision(k.refusal)
+			require.True(t, decided, "eligibility registry has no explicit decision for class=%q reason=%q cause=%q site=%q",
+				k.refusal.Class(), k.refusal.Reason(), k.refusal.Cause(), k.refusal.Site())
 			classified[k.refusal.Reason()] = true
 		})
 	}
@@ -126,15 +139,15 @@ func TestRefusalRegistryCorrespondence(t *testing.T) {
 	assert.Equal(t, verdict.ClassByDesign, class(plan.CreateShapeRefusal(executor.CreateShapeIfNotExists)))
 	assert.Equal(t, verdict.ClassCapabilityBoundary, class(plan.CreateShapeRefusal(executor.CreateShapePartitionOf)))
 	assert.Equal(t, verdict.ClassInvariantViolation, class(plan.CreateShapeRefusal(executor.CreateShapeMultipleOperations)))
-	assert.Equal(t, verdict.ClassNoOnlineSafetyProblem, class(gateRefusal(statement.KindDataChange, false)))
+	assert.Equal(t, verdict.ClassNoOnlineSafetyProblem, class(gateRefusal(statement.KindDataChange, false, statement.IndexTargetNone)))
 	// unsupported-partitioned-parent spans three.
 	assert.Equal(t, verdict.ClassCapabilityBoundary, class(plan.PartitionRefusal(preflight.PartitionCauseConcurrentIndexBuild)))
 	assert.Equal(t, verdict.ClassByDesign, class(plan.PartitionRefusal(preflight.PartitionCauseIndexAdoption)))
 	assert.Equal(t, verdict.ClassEnvironmental, class(plan.PartitionRefusal(preflight.PartitionCauseNotValidForeignKey)))
 	// index-statement: the plain form is refused by design, the concurrent
 	// form is the operator's to run.
-	assert.Equal(t, verdict.ClassByDesign, class(gateRefusal(statement.KindDropIndex, false)))
-	assert.Equal(t, verdict.ClassNoOnlineSafetyProblem, class(gateRefusal(statement.KindReindex, true)))
+	assert.Equal(t, verdict.ClassByDesign, class(gateRefusal(statement.KindDropIndex, false, statement.IndexTargetSingleRelation)))
+	assert.Equal(t, verdict.ClassNoOnlineSafetyProblem, class(gateRefusal(statement.KindReindex, true, statement.IndexTargetSingleRelation)))
 	// Owners: each no-online-safety-problem kind names who runs it.
 	for kind, owner := range map[statement.Kind]verdict.Owner{
 		statement.KindDataChange:   verdict.OwnerDataChangeRunner,
@@ -143,11 +156,11 @@ func TestRefusalRegistryCorrespondence(t *testing.T) {
 		statement.KindCreateTable:  verdict.OwnerDeclarativeFrontDoor,
 		statement.KindDropIndex:    verdict.OwnerDirectOperator,
 	} {
-		r, ok := gateRefusal(kind, kind == statement.KindDropIndex)
+		r, ok := gateRefusal(kind, kind == statement.KindDropIndex, statement.IndexTargetSingleRelation)
 		require.True(t, ok, kind)
 		assert.Equal(t, owner, r.Owner(), kind)
 	}
-	r, ok := gateRefusal(statement.KindOther, false)
+	r, ok := gateRefusal(statement.KindOther, false, statement.IndexTargetNone)
 	require.True(t, ok)
 	assert.Equal(t, verdict.ClassCapabilityBoundary, r.Class(), "unnamed grammar is a boundary, not someone else's work")
 	assert.Empty(t, r.Owner())
@@ -168,6 +181,157 @@ func TestRefusalRegistryCorrespondence(t *testing.T) {
 	assert.Equal(t, fromCause, fromSentinel)
 }
 
+// The eligible set is pinned from both sides over the same production walk
+// the completeness harness uses: exactly the single-relation DROP INDEX and
+// REINDEX gate keys and the blocking parent index build are eligible, and
+// every other key the closed sets name — the other partition causes, the
+// concurrent and multi-relation index forms, every create-shape cause,
+// admission sentinel, site refusal, and the route refusal — is ineligible.
+// A registry that flipped a sibling cause or dropped a class check would
+// change this set.
+func TestAcceptedBlockingEligibleRowsArePinned(t *testing.T) {
+	keys, _ := deriveRefusalKeys()
+	require.NotEmpty(t, keys)
+
+	var eligible []string
+	for _, k := range keys {
+		if AcceptedBlockingEligible(k.refusal) {
+			eligible = append(eligible, k.key)
+		}
+	}
+	assert.ElementsMatch(t, []string{
+		"gate:DROP INDEX/concurrent=false/target=single-relation",
+		"gate:REINDEX/concurrent=false/target=single-relation",
+		"partition:" + string(preflight.PartitionCauseBlockingIndexBuild),
+	}, eligible)
+
+	for _, cause := range preflight.PartitionRefusalCauses() {
+		r, ok := partitionRefusal(cause)
+		require.True(t, ok, cause)
+		assert.Equal(t, cause == preflight.PartitionCauseBlockingIndexBuild, AcceptedBlockingEligible(r), cause)
+	}
+	assert.False(t, AcceptedBlockingEligible(rewriteRequiredRefusal()))
+	assert.False(t, AcceptedBlockingEligible(backendUnavailableRefusal()))
+}
+
+// The registry's default arms are fail-closed, not merely undecided: a
+// reason, site, or cause outside the vocabulary it enumerates is ineligible
+// and reported as undecided, so the completeness harness rejects it. The
+// per-class constructors do not validate their reason, which is how a
+// vocabulary addition reaches the registry before it is classified.
+func TestAcceptedBlockingUnknownKeysFailClosed(t *testing.T) {
+	tests := []struct {
+		name    string
+		refusal verdict.Refusal
+	}{
+		{"unknown reason", verdict.ByDesign("brand-new-reason")},
+		{"unknown site under index-statement", verdict.ByDesign(verdict.ReasonIndexStatement).WithSite("brand-new-site")},
+		{"unknown cause under partitioned parent", verdict.CapabilityBoundary(verdict.ReasonUnsupportedPartitionedParent).WithCause("brand-new-cause")},
+		{"no cause under partitioned parent", verdict.CapabilityBoundary(verdict.ReasonUnsupportedPartitionedParent)},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			eligible, decided := acceptedBlockingDecision(tc.refusal)
+			assert.False(t, eligible)
+			assert.False(t, decided)
+			assert.False(t, AcceptedBlockingEligible(tc.refusal))
+		})
+	}
+}
+
+// Each eligible row is keyed on its class as well as its reason and
+// discriminator: the same reason and site, or reason and cause, under a
+// different class is a decided ineligible key, not an eligible one.
+func TestAcceptedBlockingRowsRequireTheirClass(t *testing.T) {
+	tests := []struct {
+		name    string
+		refusal verdict.Refusal
+	}{
+		{"index site outside by-design",
+			verdict.NoOnlineSafetyProblem(verdict.ReasonIndexStatement, verdict.OwnerDirectOperator).
+				WithSite(verdict.RefusalSiteIndexSingleRelation)},
+		{"parent cause outside capability-boundary",
+			verdict.ByDesign(verdict.ReasonUnsupportedPartitionedParent).
+				WithCause(verdict.CauseParentBlockingIndexBuild)},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			eligible, decided := acceptedBlockingDecision(tc.refusal)
+			assert.False(t, eligible)
+			assert.True(t, decided)
+		})
+	}
+}
+
+func TestIndexStatementAcceptedBlockingEligibility(t *testing.T) {
+	tests := []struct {
+		name string
+		sql  string
+		want bool
+	}{
+		{"drop one index", "DROP INDEX app.i", true},
+		{"reindex index", "REINDEX INDEX app.i", true},
+		{"reindex table", "REINDEX TABLE app.t", true},
+		{"drop multiple indexes", "DROP INDEX app.i, app.j", false},
+		{"reindex schema", "REINDEX SCHEMA app", false},
+		{"reindex database", "REINDEX DATABASE app", false},
+		{"reindex system", "REINDEX SYSTEM app", false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			st, err := statement.ParseOne(tc.sql)
+			require.NoError(t, err)
+			r, ok := gateRefusal(st.Kind(), st.Concurrent(), st.IndexTarget())
+			require.True(t, ok)
+			assert.Equal(t, tc.want, AcceptedBlockingEligible(r))
+		})
+	}
+}
+
+func TestGateVerdictAcceptedBlockingEligibility(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		sql  string
+		want bool
+	}{
+		{"single relation", "DROP INDEX app.i", true},
+		{"multiple relations", "DROP INDEX app.i, app.j", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			st, err := statement.ParseOne(tc.sql)
+			require.NoError(t, err)
+			v, refused := Gate(st)
+			require.True(t, refused)
+			r, err := v.Refusal()
+			require.NoError(t, err)
+			assert.Equal(t, tc.want, AcceptedBlockingEligible(r))
+
+			if tc.want {
+				encoded, err := json.Marshal(v)
+				require.NoError(t, err)
+				var decoded verdict.Verdict
+				require.NoError(t, json.Unmarshal(encoded, &decoded))
+				decodedRefusal, err := decoded.Refusal()
+				require.NoError(t, err)
+				assert.False(t, AcceptedBlockingEligible(decodedRefusal))
+			}
+		})
+	}
+}
+
+func TestAcceptedBlockingEligibilityIgnoresRenderedText(t *testing.T) {
+	r, ok := gateRefusal(statement.KindDropIndex, false, statement.IndexTargetSingleRelation)
+	require.True(t, ok)
+	first, err := verdict.Verdict{Detail: "first explanation", SaferIdiom: "first rendering"}.WithRefusal(r).Refusal()
+	require.NoError(t, err)
+	second, err := verdict.Verdict{Detail: "completely different", SaferIdiom: "different rendering"}.WithRefusal(r).Refusal()
+	require.NoError(t, err)
+
+	assert.Equal(t, first, second)
+	assert.Equal(t, AcceptedBlockingEligible(first), AcceptedBlockingEligible(second))
+	assert.True(t, AcceptedBlockingEligible(first))
+}
+
 // Membership and classification are one walk: an error outside the sentinel
 // set, or a sentinel wrapped in a step error (execution started), is not an
 // admission refusal.