fix: reject unknown fields on SemanticRule, remove dead CheckResult.severity

- Add extra="forbid" to SemanticRule so old YAML with filter_column fails
  loudly instead of silently losing enforcement
- Remove CheckResult.severity field (was set to "block" by every checker
  but never read by the Validator — enforcement comes from the rule config)
- Document session cost timing design choice in run_query

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: flyersworder

src/agentic_data_contracts/core/schema.py

@@ -44,6 +44,8 @@ class ResultCheck(BaseModel):
 
 
 class SemanticRule(BaseModel):
+    model_config = {"extra": "forbid"}
+
     name: str
     description: str
     enforcement: Enforcement

src/agentic_data_contracts/tools/factory.py

@@ -269,7 +269,10 @@ async def run_query(args: dict[str, Any]) -> dict[str, Any]:
             )
             return _text_response(msg)
 
-        # Record estimated cost from EXPLAIN
+        # Record estimated cost from EXPLAIN — charged before execution because
+        # the cost budget tracks database resource consumption, not successful
+        # operations. Even if result checks later block the output, the database
+        # work was performed.
         if vresult.estimated_cost_usd is not None:
             session.record_cost(vresult.estimated_cost_usd)
 

src/agentic_data_contracts/validation/checkers.py

@@ -13,7 +13,6 @@
 @dataclass
 class CheckResult:
     passed: bool
-    severity: str  # "block" | "warn" | "log"
     message: str
 
 
@@ -45,10 +44,9 @@ def check_ast(self, ast: exp.Expression, contract: DataContract) -> CheckResult:
         if disallowed:
             return CheckResult(
                 passed=False,
-                severity="block",
                 message=f"Tables not in allowlist: {', '.join(sorted(disallowed))}",
             )
-        return CheckResult(passed=True, severity="block", message="")
+        return CheckResult(passed=True, message="")
 
 
 class OperationBlocklistChecker:
@@ -68,7 +66,6 @@ def check_ast(self, ast: exp.Expression, contract: DataContract) -> CheckResult:
             if isinstance(ast, expr_type) and op_name in forbidden:
                 return CheckResult(
                     passed=False,
-                    severity="block",
                     message=f"Forbidden operation: {op_name}",
                 )
 
@@ -82,11 +79,10 @@ def check_ast(self, ast: exp.Expression, contract: DataContract) -> CheckResult:
         ):
             return CheckResult(
                 passed=False,
-                severity="block",
                 message="Forbidden operation: TRUNCATE",
             )
 
-        return CheckResult(passed=True, severity="block", message="")
+        return CheckResult(passed=True, message="")
 
 
 class NoSelectStarChecker:
@@ -96,10 +92,9 @@ def check_ast(self, ast: exp.Expression) -> CheckResult:
         if any(ast.find_all(exp.Star)):
             return CheckResult(
                 passed=False,
-                severity="block",
                 message="SELECT * is not allowed — specify explicit columns",
             )
-        return CheckResult(passed=True, severity="block", message="")
+        return CheckResult(passed=True, message="")
 
 
 class RequiredFilterChecker:
@@ -117,10 +112,9 @@ def check_ast(self, ast: exp.Expression) -> CheckResult:
         if self.column.lower() not in where_columns:
             return CheckResult(
                 passed=False,
-                severity="block",
                 message=f"Missing required filter: {self.column}",
             )
-        return CheckResult(passed=True, severity="block", message="")
+        return CheckResult(passed=True, message="")
 
 
 class BlockedColumnsChecker:
@@ -138,7 +132,6 @@ def check_ast(self, ast: exp.Expression) -> CheckResult:
         if any(ast.find_all(exp.Star)):
             return CheckResult(
                 passed=False,
-                severity="block",
                 message=(
                     "SELECT * may expose blocked columns: "
                     f"{', '.join(sorted(self.blocked))}"
@@ -155,10 +148,9 @@ def check_ast(self, ast: exp.Expression) -> CheckResult:
         if found:
             return CheckResult(
                 passed=False,
-                severity="block",
                 message=f"Blocked columns in SELECT: {', '.join(sorted(found))}",
             )
-        return CheckResult(passed=True, severity="block", message="")
+        return CheckResult(passed=True, message="")
 
 
 class RequireLimitChecker:
@@ -168,10 +160,9 @@ def check_ast(self, ast: exp.Expression) -> CheckResult:
         if not list(ast.find_all(exp.Limit)):
             return CheckResult(
                 passed=False,
-                severity="block",
                 message="Query must include a LIMIT clause",
             )
-        return CheckResult(passed=True, severity="block", message="")
+        return CheckResult(passed=True, message="")
 
 
 class MaxJoinsChecker:
@@ -185,12 +176,11 @@ def check_ast(self, ast: exp.Expression) -> CheckResult:
         if join_count > self.max_joins:
             return CheckResult(
                 passed=False,
-                severity="block",
                 message=(
                     f"Query has {join_count} JOINs, exceeds maximum of {self.max_joins}"
                 ),
             )
-        return CheckResult(passed=True, severity="block", message="")
+        return CheckResult(passed=True, message="")
 
 
 class ResultCheckRunner:
@@ -219,7 +209,6 @@ def check_results(self, columns: list[str], rows: list[tuple]) -> CheckResult:
         if self.min_rows is not None and row_count < self.min_rows:
             return CheckResult(
                 passed=False,
-                severity="block",
                 message=(
                     f"Rule '{self.rule_name}': query returned {row_count} rows, "
                     f"minimum is {self.min_rows}"
@@ -228,7 +217,6 @@ def check_results(self, columns: list[str], rows: list[tuple]) -> CheckResult:
         if self.max_rows is not None and row_count > self.max_rows:
             return CheckResult(
                 passed=False,
-                severity="block",
                 message=(
                     f"Rule '{self.rule_name}': query returned {row_count} rows, "
                     f"maximum is {self.max_rows}"
@@ -239,15 +227,14 @@ def check_results(self, columns: list[str], rows: list[tuple]) -> CheckResult:
             col_lower = {c.lower(): i for i, c in enumerate(columns)}
             idx = col_lower.get(self.column.lower())
             if idx is None:
-                return CheckResult(passed=True, severity="block", message="")
+                return CheckResult(passed=True, message="")
 
             values = [row[idx] for row in rows]
 
             if self.not_null and any(v is None for v in values):
                 null_count = sum(1 for v in values if v is None)
                 return CheckResult(
                     passed=False,
-                    severity="block",
                     message=(
                         f"Rule '{self.rule_name}': column '{self.column}' "
                         f"contains {null_count} null values"
@@ -265,7 +252,6 @@ def check_results(self, columns: list[str], rows: list[tuple]) -> CheckResult:
                     if actual_min < self.min_value:
                         return CheckResult(
                             passed=False,
-                            severity="block",
                             message=(
                                 f"Rule '{self.rule_name}': column '{self.column}' "
                                 f"min value {actual_min} "
@@ -277,11 +263,10 @@ def check_results(self, columns: list[str], rows: list[tuple]) -> CheckResult:
                     if actual_max > self.max_value:
                         return CheckResult(
                             passed=False,
-                            severity="block",
                             message=(
                                 f"Rule '{self.rule_name}': column '{self.column}' "
                                 f"max value {actual_max} exceeds limit {self.max_value}"
                             ),
                         )
 
-        return CheckResult(passed=True, severity="block", message="")
+        return CheckResult(passed=True, message="")

tests/test_core/test_schema.py

@@ -118,6 +118,19 @@ def test_rule_rejects_both_checks() -> None:
         )
 
 
+def test_old_filter_column_rejected() -> None:
+    """Old YAML with filter_column should fail loudly, not silently lose enforcement."""
+    with pytest.raises(ValueError, match="extra"):
+        SemanticRule.model_validate(
+            {
+                "name": "tenant_filter",
+                "description": "Must filter by tenant_id",
+                "enforcement": "block",
+                "filter_column": "tenant_id",
+            }
+        )
+
+
 def test_advisory_rule_no_checks() -> None:
     """Rules with neither check are advisory — shown in prompt only."""
     rule = SemanticRule(