Unify DefaultPolicyEngine decision and explanation rule chain

AGENTS.md

@@ -136,10 +136,11 @@ See [docs/integrations.md](docs/integrations.md) for MCP and HTTP examples.
 
 ## Adding a policy rule
 
-1. Add the rule to `DefaultPolicyEngine.evaluate()` in `policy.py`.
-2. **Placement matters:** rules are evaluated in order. A new rule placed before sensitivity checks silently bypasses them.
-3. If adding a new `SensitivityTag`, you must also add a corresponding policy rule — otherwise the tag is silently ignored.
-4. Cover it with a test in `tests/test_policy.py`.
+1. Implement the condition in the appropriate `default_policy_*_rules.py` helper.
+2. Register it exactly once in the ordered `_DEFAULT_RULES` tuple in `default_policy_rules.py`; both `evaluate()` and `explain()` consume that chain.
+3. **Placement matters:** rules are evaluated in order. A new rule placed before sensitivity checks can change which denial short-circuits first.
+4. If adding a new `SensitivityTag`, you must also add a corresponding policy rule — otherwise the tag is silently ignored.
+5. Cover decision/explanation agreement in `tests/test_policy_rule_chain.py` and behavior in `tests/test_policy.py`.
 
 ## Review checklist (beyond `make ci`)
 

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+- **Default policy decisions and explanations now share one ordered rule chain (#219).** `evaluate()` short-circuits the shared chain while `explain()` collects every failure through the same rules. Rate-limit explanation uses a read-only `peek()` so it can predict a denial without consuming, creating, or pruning limiter state. Agreement and no-mutation regressions cover the security-critical boundary.
+
 ## [0.12.0] - 2026-08-14
 
 ### Changed

docs/architecture.md

@@ -99,7 +99,7 @@ Intent-aware rules fail closed: a request with `intent=None` never matches a rul
 
 #### Denial explanations
 
-`PolicyEngine.explain()` (when available) returns a structured `DenialExplanation` with `denied`, `rule_name`, a `failed_conditions: list[FailedCondition]` describing each missing condition with `required`/`actual`/`suggestion`/`reason_code`, a `remediation` list, a human-readable `narrative`, and a top-level `reason_code` (the code of the first failed condition). Engines collect all failing conditions (no short-circuit) so callers get the full picture. For `DeclarativePolicyEngine`, an explicit deny rule that fully matches is reported as the cause; partial-match deny rules are skipped during explanation so the surfaced advice is actionable rather than self-defeating.
+`PolicyEngine.explain()` (when available) returns a structured `DenialExplanation` with `denied`, `rule_name`, a `failed_conditions: list[FailedCondition]` describing each missing condition with `required`/`actual`/`suggestion`/`reason_code`, a `remediation` list, a human-readable `narrative`, and a top-level `reason_code` (the code of the first failed condition). Engines collect all failing conditions (no short-circuit) so callers get the full picture. `DefaultPolicyEngine.evaluate()` and `.explain()` are driven by one internal ordered rule chain; evaluation short-circuits and records rate usage, while explanation traverses the same rate-limit rule through a read-only `peek()` that never mutates limiter state. For `DeclarativePolicyEngine`, an explicit deny rule that fully matches is reported as the cause; partial-match deny rules are skipped during explanation so the surfaced advice is actionable rather than self-defeating.
 
 #### Reason codes
 

src/weaver_kernel/coding_agent.py

@@ -11,6 +11,7 @@
 from dataclasses import dataclass
 from typing import Any, NoReturn
 
+from .default_policy_rule_types import MIN_JUSTIFICATION
 from .enums import SafetyClass
 from .errors import DriverError, PolicyDenied
 from .models import (
@@ -21,7 +22,6 @@
     PolicyTraceStep,
     Principal,
 )
-from .policy import _MIN_JUSTIFICATION
 from .policy_matching import scope_globs_match
 from .policy_reasons import AllowReason, DenialReason
 
@@ -186,10 +186,10 @@ def _require_justification(cls, capability: Capability, justification: str) -> N
         if capability.safety_class not in (SafetyClass.WRITE, SafetyClass.DESTRUCTIVE):
             return
         stripped_len = len(justification.strip())
-        if stripped_len < _MIN_JUSTIFICATION:
+        if stripped_len < MIN_JUSTIFICATION:
             cls._deny(
                 f"{capability.safety_class.value.upper()} capabilities require a justification "
-                f"of at least {_MIN_JUSTIFICATION} characters after trimming whitespace.",
+                f"of at least {MIN_JUSTIFICATION} characters after trimming whitespace.",
                 reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION),
             )
 

src/weaver_kernel/default_policy_access_rules.py

@@ -0,0 +1,216 @@
+"""Access, sensitivity, and memory checks for the default policy chain."""
+
+from __future__ import annotations
+
+from .default_policy_rule_types import MIN_JUSTIFICATION, RuleContext, RuleFailure
+from .enums import SafetyClass, SensitivityTag
+from .models import FailedCondition, PolicyTraceStep
+from .policy_reasons import DenialReason
+
+
+def _justification_failure(ctx: RuleContext, label: str) -> RuleFailure | None:
+    stripped_len = len(ctx.justification.strip())
+    if stripped_len >= MIN_JUSTIFICATION:
+        return None
+    detail = (
+        f"{label} capabilities require a justification of at least "
+        f"{MIN_JUSTIFICATION} characters. Got {len(ctx.justification)} characters "
+        f"({stripped_len} after trimming whitespace)."
+    )
+    return RuleFailure(
+        detail=detail,
+        condition=FailedCondition(
+            condition="min_justification",
+            required=MIN_JUSTIFICATION,
+            actual=stripped_len,
+            suggestion=(
+                f"Provide justification with at least {MIN_JUSTIFICATION} "
+                f"characters (currently {stripped_len})"
+            ),
+            reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION),
+        ),
+        reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION),
+    )
+
+
+def check_safety_class(ctx: RuleContext) -> list[RuleFailure]:
+    """Apply WRITE/DESTRUCTIVE role and justification requirements."""
+    failures: list[RuleFailure] = []
+    roles = set(ctx.principal.roles)
+    pid = ctx.principal.principal_id
+
+    if ctx.capability.safety_class == SafetyClass.WRITE:
+        if not (roles & {"writer", "admin"}):
+            failures.append(
+                RuleFailure(
+                    detail=(
+                        "WRITE capabilities require the 'writer' or 'admin' role. "
+                        f"Principal '{pid}' has roles: {sorted(roles)}."
+                    ),
+                    condition=FailedCondition(
+                        condition="roles",
+                        required=["writer", "admin"],
+                        actual=sorted(roles),
+                        suggestion=f"Add 'writer' or 'admin' role to principal '{pid}'",
+                        reason_code=str(DenialReason.MISSING_ROLE),
+                    ),
+                    reason_code=str(DenialReason.MISSING_ROLE),
+                )
+            )
+        failure = _justification_failure(ctx, "WRITE")
+        if failure is not None:
+            failures.append(failure)
+
+    elif ctx.capability.safety_class == SafetyClass.DESTRUCTIVE:
+        if "admin" not in roles:
+            failures.append(
+                RuleFailure(
+                    detail=(
+                        "DESTRUCTIVE capabilities require the 'admin' role. "
+                        f"Principal '{pid}' has roles: {sorted(roles)}."
+                    ),
+                    condition=FailedCondition(
+                        condition="roles",
+                        required=["admin"],
+                        actual=sorted(roles),
+                        suggestion=f"Add 'admin' role to principal '{pid}'",
+                        reason_code=str(DenialReason.MISSING_ROLE),
+                    ),
+                    reason_code=str(DenialReason.MISSING_ROLE),
+                )
+            )
+        failure = _justification_failure(ctx, "DESTRUCTIVE")
+        if failure is not None:
+            failures.append(failure)
+
+    return failures
+
+
+def check_tenant_sensitivity(ctx: RuleContext) -> list[RuleFailure]:
+    """Apply PII/PCI tenant requirements and allowed-field narrowing."""
+    if ctx.capability.sensitivity not in (SensitivityTag.PII, SensitivityTag.PCI):
+        return []
+    pid = ctx.principal.principal_id
+    if "tenant" not in ctx.principal.attributes:
+        return [
+            RuleFailure(
+                detail=(
+                    f"Capability '{ctx.capability.capability_id}' has "
+                    f"{ctx.capability.sensitivity.value} sensitivity and requires "
+                    "the principal to have a 'tenant' attribute."
+                ),
+                condition=FailedCondition(
+                    condition="tenant_attribute",
+                    required="present",
+                    actual="absent",
+                    suggestion=f"Add 'tenant' attribute to principal '{pid}'",
+                    reason_code=str(DenialReason.MISSING_TENANT_ATTRIBUTE),
+                ),
+                reason_code=str(DenialReason.MISSING_TENANT_ATTRIBUTE),
+            )
+        ]
+
+    roles = set(ctx.principal.roles)
+    if ctx.capability.allowed_fields and "pii_reader" not in roles:
+        ctx.constraints["allowed_fields"] = ctx.capability.allowed_fields
+        ctx.trace_steps.append(
+            PolicyTraceStep(
+                name="sensitivity:allowed_fields",
+                outcome="constraint_applied",
+                detail=f"applied allowed_fields={ctx.capability.allowed_fields}",
+            )
+        )
+    return []
+
+
+def check_secrets(ctx: RuleContext) -> list[RuleFailure]:
+    """Apply SECRETS role and justification requirements."""
+    if ctx.capability.sensitivity != SensitivityTag.SECRETS:
+        return []
+    failures: list[RuleFailure] = []
+    roles = set(ctx.principal.roles)
+    pid = ctx.principal.principal_id
+    if not (roles & {"admin", "secrets_reader"}):
+        failures.append(
+            RuleFailure(
+                detail=(
+                    "SECRETS capabilities require the 'admin' or 'secrets_reader' role. "
+                    f"Principal '{pid}' has roles: {sorted(roles)}."
+                ),
+                condition=FailedCondition(
+                    condition="roles",
+                    required=["admin", "secrets_reader"],
+                    actual=sorted(roles),
+                    suggestion=f"Add 'admin' or 'secrets_reader' role to principal '{pid}'",
+                    reason_code=str(DenialReason.MISSING_ROLE),
+                ),
+                reason_code=str(DenialReason.MISSING_ROLE),
+            )
+        )
+    failure = _justification_failure(ctx, "SECRETS")
+    if failure is not None:
+        failures.append(failure)
+    return failures
+
+
+def check_memory(ctx: RuleContext) -> list[RuleFailure]:
+    """Apply MEMORY write and sensitive-read role requirements."""
+    if ctx.capability.sensitivity != SensitivityTag.MEMORY:
+        return []
+    roles = set(ctx.principal.roles)
+    pid = ctx.principal.principal_id
+    memory_scope = str(ctx.request.scope.get("memory_scope", "")) if ctx.request.scope else ""
+    is_write = ctx.capability.safety_class in (SafetyClass.WRITE, SafetyClass.DESTRUCTIVE)
+
+    if is_write and not (roles & {"memory_writer", "admin"}):
+        return [
+            RuleFailure(
+                detail=(
+                    "MEMORY write capabilities require the 'memory_writer' or 'admin' role. "
+                    f"Principal '{pid}' has roles: {sorted(roles)}."
+                ),
+                condition=FailedCondition(
+                    condition="roles",
+                    required=["memory_writer", "admin"],
+                    actual=sorted(roles),
+                    suggestion=f"Add 'memory_writer' or 'admin' role to principal '{pid}'",
+                    reason_code=str(DenialReason.MEMORY_WRITE_REQUIRES_WRITER),
+                ),
+                reason_code=str(DenialReason.MEMORY_WRITE_REQUIRES_WRITER),
+            )
+        ]
+
+    if (
+        not is_write
+        and memory_scope == "sensitive"
+        and not (roles & {"memory_reader_sensitive", "admin"})
+    ):
+        return [
+            RuleFailure(
+                detail=(
+                    "MEMORY read with scope='sensitive' requires the "
+                    f"'memory_reader_sensitive' or 'admin' role. Principal '{pid}' "
+                    f"has roles: {sorted(roles)}."
+                ),
+                condition=FailedCondition(
+                    condition="roles",
+                    required=["memory_reader_sensitive", "admin"],
+                    actual=sorted(roles),
+                    suggestion=(
+                        f"Add 'memory_reader_sensitive' or 'admin' role to principal '{pid}' "
+                        "(or narrow the request scope away from 'sensitive')"
+                    ),
+                    reason_code=str(DenialReason.MEMORY_SENSITIVE_READ_DENIED),
+                ),
+                reason_code=str(DenialReason.MEMORY_SENSITIVE_READ_DENIED),
+            )
+        ]
+    return []
+
+
+__all__ = [
+    "check_memory",
+    "check_safety_class",
+    "check_secrets",
+    "check_tenant_sensitivity",
+]

src/weaver_kernel/default_policy_limit_rules.py

@@ -0,0 +1,98 @@
+"""Constraint and rate-limit checks for the default policy chain."""
+
+from __future__ import annotations
+
+from .default_policy_rule_types import (
+    MAX_ROWS_SERVICE,
+    MAX_ROWS_USER,
+    RuleContext,
+    RuleFailure,
+)
+from .models import FailedCondition, PolicyTraceStep
+from .policy_reasons import DenialReason
+from .rate_limit import SERVICE_RATE_MULTIPLIER
+
+
+def apply_row_cap(ctx: RuleContext) -> list[RuleFailure]:
+    """Validate/cap ``max_rows`` and record the applied constraint."""
+    roles = set(ctx.principal.roles)
+    max_rows = MAX_ROWS_SERVICE if "service" in roles else MAX_ROWS_USER
+    if "max_rows" in ctx.constraints:
+        try:
+            requested = int(ctx.constraints["max_rows"])
+        except (TypeError, ValueError) as exc:
+            return [
+                RuleFailure(
+                    detail=(
+                        f"Invalid 'max_rows' constraint: {ctx.constraints['max_rows']!r} "
+                        "is not a valid integer."
+                    ),
+                    condition=FailedCondition(
+                        condition="max_rows",
+                        required="integer",
+                        actual=ctx.constraints["max_rows"],
+                        suggestion="Provide 'max_rows' as a valid integer",
+                        reason_code=str(DenialReason.INVALID_CONSTRAINT),
+                    ),
+                    reason_code=str(DenialReason.INVALID_CONSTRAINT),
+                    cause=exc,
+                )
+            ]
+        ctx.constraints["max_rows"] = min(max(requested, 0), max_rows)
+    else:
+        ctx.constraints["max_rows"] = max_rows
+
+    ctx.trace_steps.append(
+        PolicyTraceStep(
+            name="row_cap",
+            outcome="constraint_applied",
+            detail="max_rows capped",
+        )
+    )
+    return []
+
+
+def check_rate_limit(ctx: RuleContext) -> list[RuleFailure]:
+    """Check the current sliding window and record usage only for decisions."""
+    safety_class = ctx.capability.safety_class
+    if safety_class not in ctx.rate_limits:
+        return []
+
+    roles = set(ctx.principal.roles)
+    limit, window = ctx.rate_limits[safety_class]
+    if "service" in roles:
+        limit *= SERVICE_RATE_MULTIPLIER
+    pid = ctx.principal.principal_id
+    cid = ctx.capability.capability_id
+    rate_key = f"{pid}:{cid}"
+    allowed = (
+        ctx.limiter.peek(rate_key, limit, window)
+        if ctx.read_only
+        else ctx.limiter.check(rate_key, limit, window)
+    )
+    if not allowed:
+        return [
+            RuleFailure(
+                detail=(
+                    f"Rate limit exceeded: {limit} {safety_class.value} "
+                    f"invocations per {window}s for principal '{pid}'"
+                ),
+                condition=FailedCondition(
+                    condition="rate_limit",
+                    required=f"fewer than {limit} invocations per {window}s",
+                    actual="limit exceeded",
+                    suggestion=(
+                        f"Wait for the {window}s rate-limit window before retrying "
+                        f"capability '{cid}'"
+                    ),
+                    reason_code=str(DenialReason.RATE_LIMITED),
+                ),
+                reason_code=str(DenialReason.RATE_LIMITED),
+            )
+        ]
+    if not ctx.read_only:
+        ctx.limiter.record(rate_key)
+    return []
+
+
+__all__ = ["apply_row_cap", "check_rate_limit"]