refactor(grounding): declare role and provenance checks with exact firing attribution

CONTRIBUTING.md

@@ -311,3 +311,20 @@ is broken: runner geography, holidays and upstream changes need checking.
 The 14-day window is a coarse freshness alarm, not a market-calendar guarantee.
 Reports are retained for 30 days. No authenticated source or broker order path
 is exercised.
+
+
+### Grounding checks
+
+`agent/src/agent/grounding/registry.py` declares each check's stable name,
+issue codes, severity and execution phase. The gate walks identity, answer,
+declaration, per-figure role and price-provenance checks in their original
+order. All production checks currently block; `warn` and `log` findings are
+recorded without changing the verdict. Severity is separate from the repair,
+recovery and redaction decisions in `release.py`.
+
+Add a predicate and registration, then a passing/failing answer pair in
+`tests/test_grounding_registry.py`. The pair catalog is checked against the
+runtime registry. Use sparse evidence fixtures. Validation artifacts retain
+`fired_checks` and `check_findings` (name, severity, findings) from actual
+predicate execution; shared issue codes do not imply that other roles fired.
+Run the grounding suites before changing a predicate or its ordering.

agent/src/agent/grounding/evidence.py

@@ -459,7 +459,6 @@ def _is_metadata_count_leaf(path: str) -> bool:
 # prefix (spec §5). An unregistered leaf is still evidence but cannot ground a
 # price: inferring price-ness from leaf names admitted ``sma_cross`` as a price.
 _REGISTERED_PRICE_INDICATORS: dict[str, tuple[str, ...]] = {
-    "get_fund_nav": ("unit_nav", "accumulated_nav", "adjusted_nav"),
     "technical_indicators": (
         "latest_close",
         "indicators.sma_",
@@ -471,6 +470,8 @@ _REGISTERED_PRICE_INDICATORS: dict[str, tuple[str, ...]] = {
 }
 
 
+_FUND_NAV_PRICE_FIELDS = frozenset({"unit_nav", "accumulated_nav", "adjusted_nav"})
+
 def _is_registered_price_indicator(tool: str, path: str) -> bool:
     """Whether a tool registered this leaf as a price-denominated level.
 
@@ -481,15 +482,15 @@ def _is_registered_price_indicator(tool: str, path: str) -> bool:
     Returns:
         True when the tool has a registration whose prefix the path matches.
     """
-    prefixes = _REGISTERED_PRICE_INDICATORS.get(str(tool or ""))
-    if not prefixes:
-        return False
     leaf = str(path or "").strip()
     if tool == "get_fund_nav":
         # Only this tool's dated, exactly resolved fund rows expose these
         # levels. A generic nav field or daily growth is not a price.
         match = re.fullmatch(r"funds\.[0-9]{6}\.OF\.rows\[[0-9]+\]\.([a-z_]+)", leaf)
-        return bool(match and match.group(1) in prefixes)
+        return bool(match and match.group(1) in _FUND_NAV_PRICE_FIELDS)
+    prefixes = _REGISTERED_PRICE_INDICATORS.get(str(tool or ""))
+    if not prefixes:
+        return False
     for prefix in prefixes:
         if leaf == prefix:
             return True

agent/src/agent/grounding/figures_checks.py

@@ -24,7 +24,7 @@ def _active_block(ledger: Any) -> FiguresBlock:
 
 
 @grounding_check(
-    name="figures-block-malformed",
+    name="figures-block-malformed", phase="declarations", order=0,
     code="figures_block_malformed",
     description="A figures block line could not be read as `value | role | note | ref`.",
 )
@@ -46,7 +46,7 @@ def _figures_block_malformed(ledger: Any, content: str) -> list[dict[str, Any]]:
 
 
 @grounding_check(
-    name="unsourced-symbol-figures",
+    name="unsourced-symbol-figures", phase="answer", order=0,
     code="unsourced_symbol_figures",
     description=(
         "A line pairs a symbol no tool call in this session passed in or "

agent/src/agent/grounding/identity_checks.py

@@ -23,7 +23,7 @@ _PRIVATE_ASSERTION_RE = re.compile(
 
 
 @grounding_check(
-    name="listed-identity-relabelled-private",
+    name="listed-identity-relabelled-private", phase="identity", order=1,
     code="listed_identity_relabelled_private",
     description=(
         "A locked listed identity (security or fund) was relabelled as "
@@ -57,7 +57,7 @@ def _listed_identity_relabelled_private(ledger: Any, content: str) -> list[dict[
 
 
 @grounding_check(
-    name="identity-not-locked",
+    name="identity-not-locked", phase="identity", order=0,
     code="identity_not_locked",
     description=(
         "A run that named an instrument concluded while its identity stayed unresolved, conflicting, or invalidated."

agent/src/agent/grounding/ledger.py

@@ -398,11 +398,12 @@ class GroundingLedger(
         self._ingest_run_dir_ohlc_csvs()
         block = parse_figures_block(content)
         figures = scan_figures(content, block)
+        self._active_check_findings: list[dict[str, Any]] = []
         self._active_block = block
         self._active_figures = figures
         issues: list[dict[str, Any]] = []
         issues.extend(self._validate_identity(content))
-        issues.extend(GROUNDING_CHECKS.run("unsourced-symbol-figures", self, content))
+        issues.extend(GROUNDING_CHECKS.run_phase("answer", self, content))
         issues.extend(self._validate_figures(content, block, figures))
         issues = self._dedupe_issues(issues)
         result = ValidationResult(
@@ -420,11 +421,10 @@ class GroundingLedger(
                 "content_sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(),
                 "valid": result.valid,
                 "issues": issues,
-                # Names let a run card say which declared checks fired without
-                # re-running the gate; inline rules carry codes only until migrated.
-                "fired_checks": GROUNDING_CHECKS.names_for_codes(
-                    {str(issue.get("code") or "") for issue in issues}
-                ),
+                # Attribute the predicates that actually returned a finding;
+                # role checks share issue codes and must not implicate each other.
+                "fired_checks": list(dict.fromkeys(row["name"] for row in self._active_check_findings)),
+                "check_findings": self._active_check_findings,
                 "figures_block": block.raw,
             }
         )

agent/src/agent/grounding/policies.py

@@ -22,7 +22,7 @@ from src.agent.grounding.identity import (
 )
 from src.agent.grounding import identity_checks  # noqa: F401  (registers declared checks)
 from src.agent.grounding import figures_checks  # noqa: F401  (registers declared checks)
-from src.agent.grounding.registry import GROUNDING_CHECKS
+from src.agent.grounding.registry import GROUNDING_CHECKS, grounding_check
 from src.agent.grounding.evidence import (
     EvidenceRecord,
     _is_metadata_count_leaf,
@@ -433,10 +433,7 @@ class _PolicyMixin:
 
     def _validate_identity(self, content: str) -> list[dict[str, Any]]:
         """Validate aggregate state and listed/private contradictions."""
-        issues: list[dict[str, Any]] = []
-        issues.extend(GROUNDING_CHECKS.run("identity-not-locked", self, content))
-        issues.extend(GROUNDING_CHECKS.run("listed-identity-relabelled-private", self, content))
-        return issues
+        return GROUNDING_CHECKS.run_phase("identity", self, content)
 
     def _validate_figures(
         self,
@@ -455,7 +452,7 @@ class _PolicyMixin:
             One issue per figure that its role does not survive, plus one per
             malformed declaration line and the provenance findings.
         """
-        issues: list[dict[str, Any]] = GROUNDING_CHECKS.run("figures-block-malformed", self, content)
+        issues: list[dict[str, Any]] = GROUNDING_CHECKS.run_phase("declarations", self, content)
         records = self._comparable_price_records()
         # Symbol resolution is broader than price comparison. A session may hold
         # quotes for the primary listing and only non-price evidence for another
@@ -542,68 +539,24 @@ class _PolicyMixin:
             ):
                 continue
             if declaration is not None:
-                written = self._written_symbol(
-                    content, figure, line_symbols, symbol_records
+                found = GROUNDING_CHECKS.run(
+                    "figure-symbol", self, content, figure, declaration, symbol, line_symbols, symbol_records,
                 )
-                if written and symbol and written != symbol:
-                    # A declaration names where a number came from; it cannot
-                    # move a figure the sentence attaches to another instrument.
-                    issues.append(
-                        self._figure_issue(
-                            "numeric_claim_conflict",
-                            figure,
-                            declaration.role,
-                            symbol,
-                            "symbol_mismatch",
-                            f"is declared for {symbol} but the answer writes it about {written}",
-                        )
-                    )
+                if found:
+                    issues.extend(found)
                     continue
             if declaration is None:
-                found = self._check_observed(figure, None, symbol, records)
-                if block.present and found:
-                    # Declared, but as a fraction where the answer writes a
-                    # percent (or the reverse): still undeclared, since the
-                    # two are different assertions — but say so, or the model
-                    # reads "not declared" as a lie and repeats the draft.
-                    other_unit = _declared_in_other_unit(block, figure)
-                    issues.append(
-                        self._figure_issue(
-                            "figure_undeclared",
-                            figure,
-                            None,
-                            symbol,
-                            "undeclared",
-                            "is not declared in the figures block and is not an observed "
-                            "value; declare it as observed / derived / proposed / cited / "
-                            "count, or remove it",
-                            **(
-                                {
-                                    "declared_as": other_unit.value_text,
-                                    "declared_sign_differs": (other_unit.value < 0) != (figure.value < 0),
-                                }
-                                if other_unit
-                                else {}
-                            ),
-                        )
-                    )
-                    continue
-                checked_price = True
+                found = GROUNDING_CHECKS.run("undeclared-figure", self, figure, symbol, records, block)
                 issues.extend(found)
+                # An undeclared figure rejected at the declaration boundary did
+                # not trigger provenance in the original walk.
+                if not any(issue["code"] == "figure_undeclared" for issue in found):
+                    checked_price = True
                 continue
             role = declaration.role
-            if figure.column and role != "observed":
-                issues.append(
-                    self._figure_issue(
-                        "numeric_claim_conflict",
-                        figure,
-                        role,
-                        symbol,
-                        "cited_as_observed" if role == "cited" else "role_in_price_column",
-                        f"is declared {role}, but it sits in the {figure.column} column, "
-                        "which holds observed prints only",
-                    )
-                )
+            found = GROUNDING_CHECKS.run("figure-price-column", self, figure, role, symbol)
+            if found:
+                issues.extend(found)
                 continue
             if role == "count":
                 posing = (
@@ -614,7 +567,7 @@ class _PolicyMixin:
                 if not _is_plain_count(figure) or posing:
                     checked_price = True
                     issues.extend(
-                        self._count_as_observed(
+                        GROUNDING_CHECKS.run("figure-count", self,
                             figure,
                             declaration,
                             symbol,
@@ -635,26 +588,96 @@ class _PolicyMixin:
                     else ""
                 )
                 issues.extend(
-                    self._check_cited(figure, declaration, symbol, declared_observed, line)
+                    GROUNDING_CHECKS.run("figure-cited", self, figure, declaration, symbol, declared_observed, line)
                 )
                 continue
             checked_price = True
             if role == "derived":
                 issues.extend(
-                    self._check_derived(figure, declaration, symbol, records, scalar_parameters)
+                    GROUNDING_CHECKS.run("figure-derived", self, figure, declaration, symbol, records, scalar_parameters)
                 )
                 continue
             if role == "proposed":
                 issues.extend(
-                    self._check_proposed(figure, declaration, symbol, records, scalar_parameters)
+                    GROUNDING_CHECKS.run("figure-proposed", self, figure, declaration, symbol, records, scalar_parameters)
                 )
                 continue
-            issues.extend(self._check_observed(figure, declaration, symbol, records))
+            issues.extend(GROUNDING_CHECKS.run("figure-observed", self, figure, declaration, symbol, records))
         market_records = self._price_records()
         if checked_price and market_records:
             issues.extend(self._validate_price_provenance(content, market_records))
         return issues
 
+    @grounding_check(name="figure-symbol", code="numeric_claim_conflict",
+                     description="A declaration cannot move a figure to another instrument.")
+    def _check_figure_symbol(self, content, figure, declaration, symbol, line_symbols, symbol_records):
+        written = self._written_symbol(
+            content, figure, line_symbols, symbol_records
+        )
+        if written and symbol and written != symbol:
+            # A declaration names where a number came from; it cannot
+            # move a figure the sentence attaches to another instrument.
+            return [
+                self._figure_issue(
+                    "numeric_claim_conflict",
+                    figure,
+                    declaration.role,
+                    symbol,
+                    "symbol_mismatch",
+                    f"is declared for {symbol} but the answer writes it about {written}",
+                )
+            ]
+        return []
+
+    @grounding_check(name="undeclared-figure", code="figure_undeclared",
+                     description="An undeclared measurement must be observed in this session.", additional_codes=('numeric_claim_conflict', 'numeric_claim_unavailable'))
+    def _check_undeclared_figure(self, figure, symbol, records, block):
+        found = self._check_observed(figure, None, symbol, records)
+        if block.present and found:
+            # Declared, but as a fraction where the answer writes a
+            # percent (or the reverse): still undeclared, since the
+            # two are different assertions — but say so, or the model
+            # reads "not declared" as a lie and repeats the draft.
+            other_unit = _declared_in_other_unit(block, figure)
+            return [
+                self._figure_issue(
+                    "figure_undeclared",
+                    figure,
+                    None,
+                    symbol,
+                    "undeclared",
+                    "is not declared in the figures block and is not an observed "
+                    "value; declare it as observed / derived / proposed / cited / "
+                    "count, or remove it",
+                    **(
+                        {
+                            "declared_as": other_unit.value_text,
+                            "declared_sign_differs": (other_unit.value < 0) != (figure.value < 0),
+                        }
+                        if other_unit
+                        else {}
+                    ),
+                )
+            ]
+        return found
+
+    @grounding_check(name="figure-price-column", code="numeric_claim_conflict",
+                     description="An OHLC column can contain only observed prints.")
+    def _check_figure_column(self, figure, role, symbol):
+        if figure.column and role != "observed":
+            return [
+                self._figure_issue(
+                    "numeric_claim_conflict",
+                    figure,
+                    role,
+                    symbol,
+                    "cited_as_observed" if role == "cited" else "role_in_price_column",
+                    f"is declared {role}, but it sits in the {figure.column} column, "
+                    "which holds observed prints only",
+                )
+            ]
+        return []
+
     @staticmethod
     def _figure_issue(
         code: str,
@@ -1400,6 +1423,9 @@ class _PolicyMixin:
                     return _nearest(figure.value, values)
         return _nearest(figure.value, fallback)
 
+    @grounding_check(name="figure-count", code="numeric_claim_conflict",
+                     additional_codes=("numeric_claim_unavailable",),
+                     description="Validate the count figure contract against evidence.")
     def _count_as_observed(
         self,
         figure: Figure,
@@ -1538,6 +1564,9 @@ class _PolicyMixin:
             band = min(band, _written_half_unit(written) * unit * (1 + 1e-9))
         return abs(candidate - target) <= max(band, 1e-9)
 
+    @grounding_check(name="figure-observed", code="numeric_claim_conflict",
+                     additional_codes=("numeric_claim_unavailable",),
+                     description="Validate the observed figure contract against evidence.")
     def _check_observed(
         self,
         figure: Figure,
@@ -1864,6 +1893,9 @@ class _PolicyMixin:
             if not sign or target * sign >= 0
         )
 
+    @grounding_check(name="figure-derived", code="numeric_claim_conflict",
+                     additional_codes=("numeric_claim_unavailable",),
+                     description="Validate the derived figure contract against evidence.")
     def _check_derived(
         self,
         figure: Figure,
@@ -1922,6 +1954,9 @@ class _PolicyMixin:
             )
         ]
 
+    @grounding_check(name="figure-proposed", code="numeric_claim_conflict",
+                     additional_codes=("numeric_claim_unavailable",),
+                     description="Validate the proposed figure contract against evidence.")
     def _check_proposed(
         self,
         figure: Figure,
@@ -1998,6 +2033,9 @@ class _PolicyMixin:
             )
         ]
 
+    @grounding_check(name="figure-cited", code="numeric_claim_conflict",
+                     additional_codes=("numeric_claim_unavailable",),
+                     description="Validate the cited figure contract against evidence.")
     def _check_cited(
         self,
         figure: Figure,
@@ -2065,13 +2103,8 @@ class _PolicyMixin:
         }
         return next(iter(matches)) if len(matches) == 1 else None
 
-    def _validate_price_provenance(
-        self,
-        content: str,
-        records: Sequence[EvidenceRecord],
-    ) -> list[dict[str, Any]]:
-        """Require canonical symbol, actual source, and quote currency in output."""
-        issues: list[dict[str, Any]] = []
+    def _provenance_scope(self, content, records):
+        """Resolve the original symbol-scoped provenance pool without changing it."""
         folded = content.casefold()
         symbols = sorted({record.symbol for record in records if record.symbol})
         # ``_scan_symbols`` canonicalizes, so an answer that writes Shanghai as
@@ -2082,6 +2115,20 @@ class _PolicyMixin:
             for symbol in symbols
             if symbol in written or symbol.casefold() in folded
         ]
+        target_symbols = set(mentioned or (symbols if len(symbols) == 1 else []))
+        target_records = [
+            record
+            for record in records
+            if not target_symbols or record.symbol in target_symbols
+        ]
+
+        return folded, symbols, mentioned, target_records
+
+    @grounding_check(name="canonical-symbol-surfaced", code="canonical_symbol_not_surfaced", phase="provenance", order=0,
+                     description="Require canonical symbol surfaced for a checked price.")
+    def _check_canonical_symbol_surfaced(self, content, records):
+        issues = []
+        folded, symbols, mentioned, target_records = self._provenance_scope(content, records)
         if not mentioned:
             issues.append(
                 {
@@ -2098,13 +2145,13 @@ class _PolicyMixin:
                     ),
                 }
             )
-        target_symbols = set(mentioned or (symbols if len(symbols) == 1 else []))
-        target_records = [
-            record
-            for record in records
-            if not target_symbols or record.symbol in target_symbols
-        ]
+        return issues
 
+    @grounding_check(name="data-source-surfaced", code="data_source_not_surfaced", phase="provenance", order=1,
+                     description="Require data source surfaced for a checked price.")
+    def _check_data_source_surfaced(self, content, records):
+        issues = []
+        folded, symbols, mentioned, target_records = self._provenance_scope(content, records)
         sources = sorted(
             {
                 record.source
@@ -2138,6 +2185,13 @@ class _PolicyMixin:
                 }
             )
 
+        return issues
+
+    @grounding_check(name="quote-currency-surfaced", code="currency_not_surfaced", phase="provenance", order=2,
+                     description="Require quote currency surfaced for a checked price.")
+    def _check_quote_currency_surfaced(self, content, records):
+        issues = []
+        folded, symbols, mentioned, target_records = self._provenance_scope(content, records)
         currencies = sorted(
             {record.currency for record in target_records if record.currency}
         )
@@ -2165,6 +2219,10 @@ class _PolicyMixin:
             )
         return issues
 
+    def _validate_price_provenance(self, content, records):
+        """Walk declared provenance checks only after a price was checked."""
+        return GROUNDING_CHECKS.run_phase("provenance", self, content, records)
+
     @staticmethod
     def _currency_is_surfaced(currency: str, content: str) -> bool:
         """Return whether a quote currency or an unambiguous alias is visible."""

agent/src/agent/grounding/registry.py

@@ -9,21 +9,22 @@ a fixture pair, not touching call flow.
 
 Migration is golden-pass: existing suites must pass unchanged while checks
 move over one at a time. ``GROUNDING_CHECKS`` walks in registration order;
-the gate in ``ledger.py`` keeps built-ins first and walks the registry where
-each migrated check used to run inline, so issue order is preserved.
+the gate walks ordered phases and per-role predicates at their original
+positions. All current checks block; severity describes the gate verdict,
+not the independent repair/redaction decision owned by release.py.
 """
 
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any, Callable, Literal
 
 if TYPE_CHECKING:
     from src.agent.grounding.ledger import GroundingLedger
 
 # A predicate gets the ledger (identity/figure state lives there) and the
 # candidate answer, and returns issue dicts in the existing shape.
-GroundingPredicate = Callable[["GroundingLedger", str], list[dict[str, Any]]]
+GroundingPredicate = Callable[..., list[dict[str, Any]]]
 
 
 @dataclass(frozen=True)
@@ -36,13 +37,21 @@ class GroundingCheck:
             codes, so a registered check never invents a new one without a
             matching decision entry.
         description: What the check rejects, for the run card and docs.
-        predicate: The check itself.
+        predicate: The check itself, with phase-specific arguments.
+        severity: Block the draft, or record a warning/log finding only.
+        phase: Ordered execution phase; manual predicates are dispatched by role.
+        order: Order within a phase, preserving short-circuit semantics.
+        additional_codes: Other issue codes this predicate can emit.
     """
 
     name: str
     code: str
     description: str
     predicate: GroundingPredicate
+    severity: Literal["block", "warn", "log"] = "block"
+    phase: str = "manual"
+    order: int = 0
+    additional_codes: tuple[str, ...] = ()
 
 
 class GroundingRegistry:
@@ -52,6 +61,8 @@ class GroundingRegistry:
         self._checks: dict[str, GroundingCheck] = {}
 
     def register(self, check: GroundingCheck) -> GroundingCheck:
+        if check.severity not in {"block", "warn", "log"}:
+            raise ValueError(f"invalid grounding severity: {check.severity}")
         if check.name in self._checks:
             raise ValueError(f"duplicate grounding check: {check.name}")
         self._checks[check.name] = check
@@ -60,24 +71,41 @@ class GroundingRegistry:
     def get(self, name: str) -> GroundingCheck:
         return self._checks[name]
 
-    def run(self, name: str, ledger: "GroundingLedger", content: str) -> list[dict[str, Any]]:
-        return self._checks[name].predicate(ledger, content)
+    def run(self, name: str, ledger: "GroundingLedger", *args: Any, **kwargs: Any) -> list[dict[str, Any]]:
+        """Run a declared predicate and record its actual findings, not code guesses."""
+        check = self._checks[name]
+        issues = check.predicate(ledger, *args, **kwargs)
+        if issues:
+            undeclared = {str(issue.get("code")) for issue in issues} - {check.code, *check.additional_codes}
+            if undeclared:
+                raise ValueError(f"{name} emitted undeclared grounding codes: {sorted(undeclared)}")
+            findings = getattr(ledger, "_active_check_findings", None)
+            if findings is not None:
+                findings.append({"name": name, "severity": check.severity, "issues": issues})
+        return issues if check.severity == "block" else []
 
-    def walk(self, ledger: "GroundingLedger", content: str) -> list[tuple[GroundingCheck, list[dict[str, Any]]]]:
-        """Every check with its issues, in registration order."""
-        return [(check, check.predicate(ledger, content)) for check in self._checks.values()]
+    def walk(self, ledger: "GroundingLedger", *args: Any, phase: str = "manual", **kwargs: Any):
+        """Run one ordered phase; role predicates receive their own typed arguments."""
+        checks = sorted((c for c in self._checks.values() if c.phase == phase), key=lambda c: c.order)
+        return [(check, self.run(check.name, ledger, *args, **kwargs)) for check in checks]
+
+    def run_phase(self, phase: str, ledger: "GroundingLedger", *args: Any, **kwargs: Any):
+        """Flatten findings without changing predicate or issue order."""
+        return [issue for _, issues in self.walk(ledger, *args, phase=phase, **kwargs) for issue in issues]
 
     def names_for_codes(self, codes: set[str]) -> list[str]:
         """Names of registered checks emitting any of ``codes``, in registration order.
 
-        Codes from rules that still run inline resolve to no name; they get one
-        when the check is migrated.
+        Catalog lookup only. Never use this to attribute actual execution: several
+        predicates can emit the same code.
         """
-        return [check.name for check in self._checks.values() if check.code in codes]
+        return [check.name for check in self._checks.values() if codes.intersection((check.code, *check.additional_codes))]
 
-    def describe(self) -> list[dict[str, str]]:
+    def describe(self) -> list[dict[str, Any]]:
         """The check catalog for run cards and docs, in registration order."""
-        return [{"name": c.name, "code": c.code, "description": c.description} for c in self._checks.values()]
+        return [{"name": c.name, "code": c.code, "additional_codes": list(c.additional_codes),
+                 "description": c.description, "severity": c.severity, "phase": c.phase}
+                for c in self._checks.values()]
 
     def __len__(self) -> int:
         return len(self._checks)
@@ -86,11 +114,18 @@ class GroundingRegistry:
 GROUNDING_CHECKS = GroundingRegistry()
 
 
-def grounding_check(*, name: str, code: str, description: str):
+def grounding_check(
+    *, name: str, code: str, description: str,
+    severity: Literal["block", "warn", "log"] = "block", phase: str = "manual",
+    order: int = 0, additional_codes: tuple[str, ...] = (),
+):
     """Decorator: register a predicate as a declared grounding check."""
 
     def wrap(predicate: GroundingPredicate) -> GroundingPredicate:
-        GROUNDING_CHECKS.register(GroundingCheck(name=name, code=code, description=description, predicate=predicate))
+        GROUNDING_CHECKS.register(GroundingCheck(
+            name=name, code=code, description=description, predicate=predicate,
+            severity=severity, phase=phase, order=order, additional_codes=additional_codes,
+        ))
         return predicate
 
     return wrap

agent/tests/test_grounding_registry.py

@@ -7,6 +7,9 @@ from pathlib import Path
 
 import pytest
 
+from tests.test_grounding_figures_contract import HDR, HDR_ROW, _block, _ledger
+from tests.test_grounding_role_hardening import MARKET_A, MARKET_C, _ledger as _multi_ledger
+
 from src.agent.grounding.ledger import GroundingLedger
 from src.agent.grounding.registry import GROUNDING_CHECKS, GroundingCheck, GroundingRegistry
 
@@ -22,7 +25,7 @@ def test_registry_rejects_duplicate_names() -> None:
 def test_registry_walk_order_and_describe_are_stable() -> None:
     registry = GroundingRegistry()
     for name in ("b-check", "a-check"):
-        registry.register(GroundingCheck(name=name, code=f"{name}_code", description="d", predicate=lambda l, c: []))
+        registry.register(GroundingCheck(name=name, code=f"{name}_code", description="d", predicate=lambda ledger, content: []))
     assert [c.name for c, _ in registry.walk(None, "")] == ["b-check", "a-check"]
     assert [d["name"] for d in registry.describe()] == ["b-check", "a-check"]
 
@@ -103,11 +106,11 @@ def test_fired_checks_name_the_declared_check_that_fired(tmp_path: Path) -> None
     assert _recorded_validations(tmp_path)[-1]["fired_checks"] == ["listed-identity-relabelled-private"]
 
 
-def test_fired_checks_stay_empty_when_only_inline_rules_fire(tmp_path: Path) -> None:
+def test_fired_checks_name_the_previously_inline_undeclared_rule(tmp_path: Path) -> None:
     ledger = _ledger_with_locked_listed_identity(tmp_path)
     result = ledger.validate_final_answer("GLD 现价 999.99 美元，建议买入。")
     assert result.issues
-    assert _recorded_validations(tmp_path)[-1]["fired_checks"] == []
+    assert _recorded_validations(tmp_path)[-1]["fired_checks"] == ["undeclared-figure"]
 
 
 def test_fired_checks_empty_for_a_clean_answer(tmp_path: Path) -> None:
@@ -135,6 +138,9 @@ def test_declared_checks_cover_identity_and_figure_slices_in_order() -> None:
         "identity-not-locked",
         "figures-block-malformed",
         "unsourced-symbol-figures",
+        "figure-symbol", "undeclared-figure", "figure-price-column",
+        "figure-count", "figure-observed", "figure-derived", "figure-proposed", "figure-cited",
+        "canonical-symbol-surfaced", "data-source-surfaced", "quote-currency-surfaced",
     ]
 
 
@@ -190,3 +196,76 @@ def test_fired_checks_name_the_identity_and_figure_slices(tmp_path: Path) -> Non
         (tmp_path / "figs" / "artifacts" / "grounding_evidence.json").read_text(encoding="utf-8")
     )
     assert artifact["validations"][-1]["fired_checks"] == ["figures-block-malformed"]
+
+
+# One positive/negative answer pair for every catalog entry. Numeric fixtures
+# use only the sparse two-bar ledger: dense frames can accidentally anchor bugs.
+
+PAIRS = {
+    "listed-identity-relabelled-private": ("listed", "GLD is a listed fund.", "GLD is a private company."),
+    "identity-not-locked": ("identity", "GLD is a listed fund.", "机器人ETF 可以买入。"),
+    "figures-block-malformed": ("market", HDR, HDR + "\n```figures\nbad row\n```"),
+    "unsourced-symbol-figures": ("market", HDR, HDR + " 000858.SZ costs 15.20 CNY."),
+    "figure-symbol": ("multi", "510300.SH akshare CNY close 3.912." + _block("3.912 | observed | 510300.SH close | c3"),
+                      "159516.SZ akshare CNY close 3.912." + _block("3.912 | observed | 510300.SH close | c3")),
+    "undeclared-figure": ("market", HDR + _block(HDR_ROW), HDR + " Value 0.777." + _block(HDR_ROW)),
+    "figure-price-column": ("market", HDR + "\n| Close |\n|---|\n|0.666|" + _block(HDR_ROW),
+                            HDR + "\n| Close |\n|---|\n|0.666|" + _block("0.666 | proposed | entry")),
+    "figure-count": ("market", HDR + " Count 3.0." + _block(HDR_ROW, "3.0 | count | names"),
+                     HDR + " Count 3.0 CNY." + _block(HDR_ROW, "3.0 | count | names")),
+    "figure-observed": ("market", HDR + _block(HDR_ROW), HDR + " Price 0.777." + _block(HDR_ROW, "0.777 | observed | close | c1")),
+    "figure-derived": ("market", HDR + " Result 0.099." + _block(HDR_ROW, "0.099 | derived | 0.666 - 0.567 | c1"),
+                       HDR + " Result 0.123." + _block(HDR_ROW, "0.123 | derived | 0.666 - 0.567 | c1")),
+    "figure-proposed": ("market", HDR + " Entry 0.950." + _block(HDR_ROW, "0.950 | proposed | entry"),
+                        HDR + " Entry 9.950." + _block(HDR_ROW, "9.950 | proposed | entry")),
+    "figure-cited": ("market", HDR + " Study score 1.8." + _block(HDR_ROW, "1.8 | cited | Study"),
+                     HDR + " Score 1.8." + _block(HDR_ROW, "1.8 | cited | Study")),
+    "canonical-symbol-surfaced": ("market", HDR, "akshare CNY close 0.666."),
+    "data-source-surfaced": ("market", HDR, "159516.SZ CNY close 0.666."),
+    "quote-currency-surfaced": ("market", HDR, "159516.SZ akshare close 0.666."),
+}
+
+
+def test_every_declared_check_has_a_fixture_pair():
+    assert set(PAIRS) == {row["name"] for row in GROUNDING_CHECKS.describe()}
+    assert {row["severity"] for row in GROUNDING_CHECKS.describe()} == {"block"}
+
+
+@pytest.mark.parametrize("name", PAIRS)
+def test_declared_check_fixture_pair(tmp_path, name):
+    kind, good, bad = PAIRS[name]
+    for failing, answer in [(False, good), (True, bad)]:
+        path = tmp_path / str(failing)
+        if kind == "identity" and failing:
+            ledger = _ledger_with_invalidated_identity(path)
+        elif kind in {"identity", "listed"}:
+            ledger = _ledger_with_locked_listed_identity(path)
+        elif kind == "multi":
+            ledger = _multi_ledger(path, MARKET_A, MARKET_C)
+        else:
+            ledger = _ledger(path)
+        verdict = ledger.validate_final_answer(answer)
+        recorded = _recorded_validations(path)[-1]
+        assert (name in recorded["fired_checks"]) is failing, recorded
+        if not failing:
+            assert verdict.valid, verdict.issues
+        assert all(row["severity"] == "block" for row in recorded["check_findings"])
+
+
+def test_shared_issue_codes_do_not_implicate_other_roles(tmp_path):
+    ledger = _ledger(tmp_path)
+    ledger.validate_final_answer(PAIRS["figure-derived"][2])
+    assert _recorded_validations(tmp_path)[-1]["fired_checks"] == ["figure-derived"]
+    ledger.validate_final_answer(HDR)
+    assert _recorded_validations(tmp_path)[-1]["check_findings"] == []
+
+
+def test_warning_and_log_findings_are_audited_without_blocking():
+    from types import SimpleNamespace
+    ledger = SimpleNamespace(_active_check_findings=[])
+    registry = GroundingRegistry()
+    for severity in ("warn", "log", "block"):
+        registry.register(GroundingCheck(name=severity, code="fixture", description="d", severity=severity,
+                                        predicate=lambda ledger, content: [{"code": "fixture"}]))
+    assert registry.run_phase("manual", ledger, "answer") == [{"code": "fixture"}]
+    assert [row["severity"] for row in ledger._active_check_findings] == ["warn", "log", "block"]