feat: classify accepted-blocking refusal eligibility

docs/invariants.md

@@ -418,6 +418,11 @@ 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. A verdict
+  decoded from JSON reconstructs class, reason, owner, and cause, but cannot recover the
+  unexported refusal site; eligibility decisions keyed by site therefore fail closed for
+  decoded verdicts. *Enforced:* `Verdict.WithRefusal`, `Verdict.Refusal`, and the accepted-
+  blocking eligibility gate-path tests. *Source:* [lock-budgeted passthrough](lock-budgeted-passthrough.md).
 
 ## Orchestration / control-plane (OC)
 

docs/lock-budgeted-passthrough.md

@@ -102,6 +102,11 @@ 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. 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, and any
+site-keyed accepted-blocking eligibility decision fails closed.
+
 “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 +378,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,87 @@ func TestRefusalRegistryCorrespondence(t *testing.T) {
 	assert.Equal(t, fromCause, fromSentinel)
 }
 
+func TestAcceptedBlockingEligibleRowsArePinned(t *testing.T) {
+	index, ok := gateRefusal(statement.KindDropIndex, false, statement.IndexTargetSingleRelation)
+	require.True(t, ok)
+	parent, ok := partitionRefusal(preflight.PartitionCauseBlockingIndexBuild)
+	require.True(t, ok)
+
+	assert.True(t, AcceptedBlockingEligible(index))
+	assert.True(t, AcceptedBlockingEligible(parent))
+	assert.False(t, AcceptedBlockingEligible(rewriteRequiredRefusal()))
+	assert.False(t, AcceptedBlockingEligible(backendUnavailableRefusal()))
+}
+
+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.

pkg/migrate/verdicts.go

@@ -21,7 +21,7 @@ import (
 // statements are never executed. Gate needs no database, so a caller can
 // refuse before dialing; [Run] re-checks it regardless.
 func Gate(st statement.Statement) (verdict.Verdict, bool) {
-	r, refused := gateRefusal(st.Kind(), st.Concurrent())
+	r, refused := gateRefusal(st.Kind(), st.Concurrent(), st.IndexTarget())
 	if !refused {
 		return verdict.Verdict{}, false
 	}
@@ -170,14 +170,20 @@ func admissionRefusalVerdict(st statement.Statement, err error, r verdict.Refusa
 // different strategy, while a blind attempt that ran past its budget is
 // doing rewrite work. A refused forced attempt still records the override.
 func budgetVerdict(st statement.Statement, budgetErr *executor.BudgetError, forced, online bool) verdict.Verdict {
+	cause := verdict.CauseNone
+	switch budgetErr.Cause {
+	case executor.CauseLock:
+		cause = verdict.CauseLockBudget
+	case executor.CauseStatement:
+		cause = verdict.CauseStatementBudget
+	}
 	v := verdict.Verdict{
 		Statement: st.SQL(),
 		Table:     qualified(st),
 		Forced:    forced,
-	}.WithRefusal(budgetExceededRefusal())
+	}.WithRefusal(budgetExceededRefusal().WithCause(cause))
 	switch budgetErr.Cause {
 	case executor.CauseLock:
-		v.Cause = verdict.CauseLockBudget
 		v.Attempts = budgetErr.Attempts
 		if budgetErr.Attempts > 1 {
 			v.Detail = fmt.Sprintf("the lock was not granted within the %s lock budget on any of %d bounded "+
@@ -188,7 +194,6 @@ func budgetVerdict(st statement.Statement, budgetErr *executor.BudgetError, forc
 		v.Detail = fmt.Sprintf("the lock was not granted within the %s lock budget: the table is too "+
 			"contended right now; nothing was executed", budgetErr.Budget)
 	case executor.CauseStatement:
-		v.Cause = verdict.CauseStatementBudget
 		if online {
 			v.Detail = fmt.Sprintf("cancelled after the %s budget: the statement already is the safe online "+
 				"idiom — the work needs more time, not a different strategy; retry with a larger budget for "+

pkg/plan/plan_test.go

@@ -205,6 +205,19 @@ func TestRefuseUnsupportedPartitionedParentClassFollowsCause(t *testing.T) {
 	}
 }
 
+// The refusal proof carries the preflight cause under the verdict package's
+// own typed vocabulary. The two constant sets spell the same wire tokens, and
+// this pins that correspondence for every cause preflight registers: a cause
+// added on one side without the other is a registry hole, not a silent
+// no-cause refusal.
+func TestPartitionRefusalCarriesItsCause(t *testing.T) {
+	for _, cause := range preflight.PartitionRefusalCauses() {
+		r, ok := plan.PartitionRefusal(cause)
+		require.True(t, ok, cause)
+		assert.Equal(t, string(cause), string(r.Cause()), cause)
+	}
+}
+
 func TestRefuseUnsupportedPartitionedParentFailsClosed(t *testing.T) {
 	r := plan.Report{Disposition: router.DispositionExecute, Statements: []plan.Statement{{Disposition: router.DispositionExecute}}}
 	require.Error(t, plan.RefuseUnsupportedPartitionedParent(&r, nil), "positional length mismatch")

pkg/plan/refusal.go

@@ -44,21 +44,30 @@ func CreateShapeRefusal(cause executor.CreateShapeCause) (verdict.Refusal, bool)
 // PartitionRefusal classifies a partitioned-parent refusal by its cause. The
 // closed key set is preflight.PartitionRefusalCauses(); ok is false outside
 // it. The four causes span three classes, which is why the plan carries the
-// cause rather than a bare refused flag.
+// cause rather than a bare refused flag. The refusal keeps its typed cause,
+// so every consumer of the proof — both front doors and the accepted-blocking
+// eligibility registry — reads one narrowing instead of re-deriving it from
+// the preflight error.
 func PartitionRefusal(cause preflight.PartitionRefusalCause) (verdict.Refusal, bool) {
 	switch cause {
-	case preflight.PartitionCauseConcurrentIndexBuild, preflight.PartitionCauseBlockingIndexBuild:
-		// The partition-aware concurrent index flow is a planned capability;
-		// refusing the blocking substitute is the policy half of the same gap.
-		return verdict.CapabilityBoundary(verdict.ReasonUnsupportedPartitionedParent), true
+	case preflight.PartitionCauseConcurrentIndexBuild:
+		// The partition-aware concurrent index flow is a planned capability.
+		return verdict.CapabilityBoundary(verdict.ReasonUnsupportedPartitionedParent).
+			WithCause(verdict.CauseParentConcurrentIndexBuild), true
+	case preflight.PartitionCauseBlockingIndexBuild:
+		// Refusing the blocking substitute is the policy half of the same gap.
+		return verdict.CapabilityBoundary(verdict.ReasonUnsupportedPartitionedParent).
+			WithCause(verdict.CauseParentBlockingIndexBuild), true
 	case preflight.PartitionCauseIndexAdoption:
 		// No supported PostgreSQL version adopts an index as a constraint on
 		// a partitioned parent; waiting for an engine release waits for nothing.
-		return verdict.ByDesign(verdict.ReasonUnsupportedPartitionedParent), true
+		return verdict.ByDesign(verdict.ReasonUnsupportedPartitionedParent).
+			WithCause(verdict.CauseParentIndexAdoption), true
 	case preflight.PartitionCauseNotValidForeignKey:
 		// The same statement runs on a newer server; the unblocking action is
 		// a server upgrade, not an engine release.
-		return verdict.Environmental(verdict.ReasonUnsupportedPartitionedParent), true
+		return verdict.Environmental(verdict.ReasonUnsupportedPartitionedParent).
+			WithCause(verdict.CauseParentNotValidForeignKey), true
 	default:
 		return verdict.Refusal{}, false
 	}