Finish TEN-WP-0006-T03: persist guardrails in both stores

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/tenant_engine/guardrail/__init__.py

@@ -27,7 +27,8 @@ from tenant_engine.guardrail.registry import (
     LimitRegistry,
     UnknownLimitKeyError,
 )
-from tenant_engine.guardrail.resolution import resolve_limit, resolve_limits
+from tenant_engine.guardrail.resolution import is_loosening, resolve_limit, resolve_limits
+from tenant_engine.guardrail.serde import dump_limit, load_limit
 
 __all__ = [
     "DEFAULT_REGISTRY",
@@ -44,6 +45,9 @@ __all__ = [
     "Provenance",
     "Unlimited",
     "UnknownLimitKeyError",
+    "dump_limit",
+    "is_loosening",
+    "load_limit",
     "resolve_limit",
     "resolve_limits",
 ]

src/tenant_engine/guardrail/resolution.py

@@ -83,6 +83,22 @@ def resolve_limits(
     }
 
 
+def is_loosening(previous: LimitValue, candidate: LimitValue) -> bool:
+    """Whether `candidate` raises the ceiling relative to `previous`.
+
+    Used to decide what a retired tenant may still change. The lifecycle clamp
+    already pins a retired tenant's *effective* values to the floor, so a
+    loosening override looks inert -- but it would take effect the moment the
+    tenant is reactivated. That deferred loosening is the thing to refuse.
+    """
+    if candidate.is_unlimited:
+        return not previous.is_unlimited
+    if previous.is_unlimited:
+        return False
+    assert isinstance(previous.amount, int) and isinstance(candidate.amount, int)
+    return candidate.amount > previous.amount
+
+
 def _clamp(effective: EffectiveLimit, tenant: Tenant, floor: LimitValue) -> EffectiveLimit:
     """Apply the lifecycle clamp. Applied after precedence, and may only reduce.
 

src/tenant_engine/guardrail/serde.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from tenant_engine.guardrail.model import (
+    UNLIMITED,
+    InvalidLimitError,
+    LimitKind,
+    LimitValue,
+)
+
+# The `unlimited` sentinel is serialised as an explicit token, never as NULL
+# or a missing field. A stored NULL that meant "unlimited" would make an open
+# ceiling the result of absence -- exactly what the contract forbids.
+UNLIMITED_TOKEN = "unlimited"
+
+
+def dump_limit(value: LimitValue) -> dict[str, Any]:
+    return {
+        "kind": value.kind.value,
+        "amount": UNLIMITED_TOKEN if value.is_unlimited else str(value.amount),
+        "currency": value.currency,
+        "period": value.period,
+    }
+
+
+def load_limit(raw: Mapping[str, Any]) -> LimitValue:
+    amount_text = raw["amount"]
+    if amount_text is None:
+        # A row that lost its amount is not an unlimited tenant. Fail loudly
+        # rather than resolving to the most permissive possible answer.
+        raise InvalidLimitError("stored limit has no amount")
+    amount = UNLIMITED if amount_text == UNLIMITED_TOKEN else int(amount_text)
+    return LimitValue(
+        kind=LimitKind(raw["kind"]),
+        amount=amount,
+        currency=raw["currency"],
+        period=raw["period"],
+    )

src/tenant_engine/sqlite_store.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import json
 import sqlite3
 from collections.abc import Callable
+from dataclasses import replace
 from datetime import datetime
 from threading import RLock
 from typing import Any
@@ -15,6 +16,13 @@ from tenant_engine.domain import (
     TenantLifecycle,
     TenantRetiredError,
 )
+from tenant_engine.guardrail import (
+    DEFAULT_REGISTRY,
+    GuardrailChange,
+    LimitValue,
+    dump_limit,
+    load_limit,
+)
 from tenant_engine.store import (
     DomainEvent,
     GrantNotFoundError,
@@ -22,6 +30,7 @@ from tenant_engine.store import (
     TenantAlreadyExistsError,
     TenantNotFoundError,
     VersionConflictError,
+    guard_guardrail_write,
 )
 
 
@@ -57,6 +66,18 @@ class SQLiteTenantStore:
                   recorded_at TEXT NOT NULL,
                   PRIMARY KEY (tenant_id, idempotency_key)
                 );
+                CREATE TABLE IF NOT EXISTS guardrail_overrides (
+                  tenant_id TEXT NOT NULL, limit_key TEXT NOT NULL,
+                  kind TEXT NOT NULL, amount TEXT NOT NULL,
+                  currency TEXT, period TEXT,
+                  PRIMARY KEY (tenant_id, limit_key)
+                );
+                CREATE TABLE IF NOT EXISTS guardrail_changes (
+                  change_id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL,
+                  limit_key TEXT NOT NULL, previous TEXT, current TEXT,
+                  changed_by TEXT NOT NULL, reason TEXT NOT NULL,
+                  correlation_id TEXT NOT NULL, changed_at TEXT NOT NULL
+                );
             """)
             self._migrate_tenant_lifecycle()
 
@@ -177,6 +198,150 @@ class SQLiteTenantStore:
             self._db.commit()
         return updated, False
 
+    def guardrail_overrides(self, tenant_id: str) -> dict[str, LimitValue]:
+        tenant = self.get_tenant(tenant_id)
+        with self._lock:
+            rows = self._db.execute(
+                "SELECT * FROM guardrail_overrides WHERE tenant_id = ?", (tenant.tenant_id,)
+            ).fetchall()
+        return {row["limit_key"]: load_limit(row) for row in rows}
+
+    def guardrail_changes(self, tenant_id: str) -> list[GuardrailChange]:
+        tenant = self.get_tenant(tenant_id)
+        with self._lock:
+            rows = self._db.execute(
+                "SELECT * FROM guardrail_changes WHERE tenant_id = ? "
+                "ORDER BY changed_at, change_id",
+                (tenant.tenant_id,),
+            ).fetchall()
+        return [
+            GuardrailChange(
+                change_id=row["change_id"],
+                tenant_id=row["tenant_id"],
+                limit_key=row["limit_key"],
+                previous=load_limit(json.loads(row["previous"])) if row["previous"] else None,
+                current=load_limit(json.loads(row["current"])) if row["current"] else None,
+                changed_by=row["changed_by"],
+                reason=row["reason"],
+                correlation_id=row["correlation_id"],
+                changed_at=datetime.fromisoformat(row["changed_at"]),
+            )
+            for row in rows
+        ]
+
+    def set_guardrail_override(
+        self,
+        *,
+        tenant_id: str,
+        expected_version: int,
+        limit_key: str,
+        value: LimitValue | None,
+        change_id: str,
+        changed_by: str,
+        reason: str,
+        correlation_id: str,
+        idempotency_key: str,
+        request_fingerprint: str,
+        at: datetime,
+    ) -> tuple[Tenant, GuardrailChange | None, bool]:
+        tenant = self.get_tenant(tenant_id)
+        DEFAULT_REGISTRY.get(limit_key)
+
+        with self._lock:
+            # Same shape as mutate_tenant: BEGIN IMMEDIATE takes the write lock
+            # before the reads, receipt lookup precedes the version check, and
+            # override write + audit row + receipt + version bump commit
+            # together. A crash must never leave a raised ceiling with no
+            # record of who raised it.
+            self._db.execute("BEGIN IMMEDIATE")
+            try:
+                receipt = self._db.execute(
+                    "SELECT request_fingerprint, result FROM idempotency_receipts "
+                    "WHERE tenant_id = ? AND idempotency_key = ?",
+                    (tenant.tenant_id, idempotency_key),
+                ).fetchone()
+                if receipt is not None:
+                    if receipt["request_fingerprint"] != request_fingerprint:
+                        raise IdempotencyConflictError(idempotency_key)
+                    replayed = _tenant(json.loads(receipt["result"]))
+                    self._db.rollback()
+                    prior = next(
+                        (c for c in self.guardrail_changes(tenant.tenant_id)
+                         if c.change_id == change_id),
+                        None,
+                    )
+                    return replayed, prior, True
+
+                current = _tenant(
+                    self._db.execute(
+                        "SELECT * FROM tenants WHERE tenant_id = ?", (tenant.tenant_id,)
+                    ).fetchone()
+                )
+                if current.version != expected_version:
+                    raise VersionConflictError(expected=expected_version, actual=current.version)
+
+                override_rows = self._db.execute(
+                    "SELECT * FROM guardrail_overrides WHERE tenant_id = ?", (current.tenant_id,)
+                ).fetchall()
+                overrides = {row["limit_key"]: load_limit(row) for row in override_rows}
+                guard_guardrail_write(
+                    tenant=current, overrides=overrides, limit_key=limit_key, value=value
+                )
+
+                change = GuardrailChange(
+                    change_id=change_id,
+                    tenant_id=current.tenant_id,
+                    limit_key=limit_key,
+                    previous=overrides.get(limit_key),
+                    current=value,
+                    changed_by=changed_by,
+                    reason=reason,
+                    correlation_id=correlation_id,
+                    changed_at=at,
+                )
+
+                if value is None:
+                    self._db.execute(
+                        "DELETE FROM guardrail_overrides WHERE tenant_id = ? AND limit_key = ?",
+                        (current.tenant_id, limit_key),
+                    )
+                else:
+                    dumped = dump_limit(value)
+                    self._db.execute(
+                        "INSERT OR REPLACE INTO guardrail_overrides VALUES (?, ?, ?, ?, ?, ?)",
+                        (current.tenant_id, limit_key, dumped["kind"], dumped["amount"],
+                         dumped["currency"], dumped["period"]),
+                    )
+
+                self._db.execute(
+                    "INSERT INTO guardrail_changes VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                    (change.change_id, change.tenant_id, change.limit_key,
+                     json.dumps(dump_limit(change.previous)) if change.previous else None,
+                     json.dumps(dump_limit(change.current)) if change.current else None,
+                     change.changed_by, change.reason, change.correlation_id, _iso(at)),
+                )
+
+                updated = replace(current, version=current.version + 1, updated_at=at)
+                self._db.execute(
+                    "UPDATE tenants SET version = ?, updated_at = ? WHERE tenant_id = ?",
+                    (updated.version, _iso(updated.updated_at), updated.tenant_id),
+                )
+                self._db.execute(
+                    "INSERT INTO idempotency_receipts VALUES (?, ?, ?, ?, ?)",
+                    (updated.tenant_id, idempotency_key, request_fingerprint,
+                     json.dumps(_row(updated)), datetime.now().astimezone().isoformat()),
+                )
+                self._emit("guardrail_changed", updated.tenant_id, {
+                    "limit_key": limit_key, "change_id": change_id,
+                    "cleared": value is None, "correlation_id": correlation_id,
+                    "version": updated.version,
+                })
+            except BaseException:
+                self._db.rollback()
+                raise
+            self._db.commit()
+        return updated, change, False
+
     def grant_role(self, grant: RoleGrant) -> None:
         tenant = self.get_tenant(grant.tenant_id)
         self._require_active(tenant, "grant a role")

src/tenant_engine/store.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import UTC, datetime
 from typing import Any, Protocol
 
@@ -13,6 +13,13 @@ from tenant_engine.domain import (
     TenantLifecycle,
     TenantRetiredError,
 )
+from tenant_engine.guardrail import (
+    DEFAULT_REGISTRY,
+    GuardrailChange,
+    LimitValue,
+    is_loosening,
+    resolve_limit,
+)
 
 
 class TenantNotFoundError(KeyError):
@@ -109,6 +116,82 @@ class TenantStore(Protocol):
         """
         ...
 
+    def guardrail_overrides(self, tenant_id: str) -> dict[str, LimitValue]:
+        """The tenant's explicit per-tenant overrides (precedence layer 1).
+
+        Absence of a key here is not "unlimited" -- it means resolution falls
+        through to the plan, then the grouping default. See
+        `guardrail.resolve_limits`.
+        """
+        ...
+
+    def guardrail_changes(self, tenant_id: str) -> list[GuardrailChange]: ...
+
+    def set_guardrail_override(
+        self,
+        *,
+        tenant_id: str,
+        expected_version: int,
+        limit_key: str,
+        value: LimitValue | None,
+        change_id: str,
+        changed_by: str,
+        reason: str,
+        correlation_id: str,
+        idempotency_key: str,
+        request_fingerprint: str,
+        at: datetime,
+    ) -> tuple[Tenant, GuardrailChange | None, bool]:
+        """Set (or, with `value=None`, clear) one override. Returns
+        `(tenant, change, replayed)`.
+
+        Carries the same four concerns as `mutate_tenant` in one transaction --
+        idempotency replay, version CAS, the override write, and the audit
+        record -- for the same reason: a crash must never leave a changed
+        ceiling with no audit record of who raised it.
+
+        The tenant's version is bumped, so a guardrail change invalidates the
+        record's ETag exactly as a metadata edit does. A caller holding a
+        stale ETag must re-read rather than write blind.
+        """
+        ...
+
+
+def guard_guardrail_write(
+    *,
+    tenant: Tenant,
+    overrides: dict[str, LimitValue],
+    limit_key: str,
+    value: LimitValue | None,
+) -> None:
+    """Refuse a guardrail change that loosens a retired tenant's ceiling.
+
+    Follows the TEN-WP-0005 precedent: while retired, operations that only
+    *reduce* privilege stay available, loosening ones do not. Retirement does
+    not freeze the record outright -- tightening a retired tenant's limit is
+    still allowed, because refusing it would be the fail-open choice.
+
+    The comparison is made as if the tenant were active. A retired tenant's
+    effective limits are already clamped to the floor, so every candidate
+    looks inert today; what matters is the value that would take effect the
+    moment it is reactivated.
+    """
+    if tenant.lifecycle is TenantLifecycle.ACTIVE:
+        return
+
+    as_active = replace(tenant, lifecycle=TenantLifecycle.ACTIVE)
+    before = resolve_limit(limit_key, tenant=as_active, overrides=overrides).value
+
+    candidate = dict(overrides)
+    if value is None:
+        candidate.pop(limit_key, None)
+    else:
+        candidate[limit_key] = value
+    after = resolve_limit(limit_key, tenant=as_active, overrides=candidate).value
+
+    if is_loosening(before, after):
+        raise TenantRetiredError("cannot loosen a guardrail on a retired tenant")
+
 
 class InMemoryTenantStore:
     def __init__(self) -> None:
@@ -119,6 +202,8 @@ class InMemoryTenantStore:
         self._events: list[DomainEvent] = []
         # (tenant_id, idempotency_key) -> (request_fingerprint, result snapshot)
         self._receipts: dict[tuple[str, str], tuple[str, Tenant]] = {}
+        self._overrides: dict[str, dict[str, LimitValue]] = {}
+        self._guardrail_changes: list[GuardrailChange] = []
 
     def create_tenant(self, tenant: Tenant) -> None:
         if tenant.tenant_id in self._tenants:
@@ -212,6 +297,88 @@ class InMemoryTenantStore:
         self._emit(event_type, resolved, {**evidence, "version": updated.version})
         return updated, False
 
+    def guardrail_overrides(self, tenant_id: str) -> dict[str, LimitValue]:
+        return dict(self._overrides.get(self._resolve(tenant_id), {}))
+
+    def guardrail_changes(self, tenant_id: str) -> list[GuardrailChange]:
+        resolved = self._resolve(tenant_id)
+        return [c for c in self._guardrail_changes if c.tenant_id == resolved]
+
+    def set_guardrail_override(
+        self,
+        *,
+        tenant_id: str,
+        expected_version: int,
+        limit_key: str,
+        value: LimitValue | None,
+        change_id: str,
+        changed_by: str,
+        reason: str,
+        correlation_id: str,
+        idempotency_key: str,
+        request_fingerprint: str,
+        at: datetime,
+    ) -> tuple[Tenant, GuardrailChange | None, bool]:
+        resolved = self._resolve(tenant_id)
+
+        # Registry check first: an unregistered key is an error, not a silent
+        # write of a ceiling nothing will ever read.
+        DEFAULT_REGISTRY.get(limit_key)
+
+        receipt = self._receipts.get((resolved, idempotency_key))
+        if receipt is not None:
+            fingerprint, snapshot = receipt
+            if fingerprint != request_fingerprint:
+                raise IdempotencyConflictError(idempotency_key)
+            replayed_change = next(
+                (c for c in self._guardrail_changes if c.change_id == change_id), None
+            )
+            return snapshot, replayed_change, True
+
+        current = self._tenants[resolved]
+        if current.version != expected_version:
+            raise VersionConflictError(expected=expected_version, actual=current.version)
+
+        overrides = self._overrides.setdefault(resolved, {})
+        guard_guardrail_write(
+            tenant=current, overrides=overrides, limit_key=limit_key, value=value
+        )
+
+        previous = overrides.get(limit_key)
+        change = GuardrailChange(
+            change_id=change_id,
+            tenant_id=resolved,
+            limit_key=limit_key,
+            previous=previous,
+            current=value,
+            changed_by=changed_by,
+            reason=reason,
+            correlation_id=correlation_id,
+            changed_at=at,
+        )
+
+        if value is None:
+            overrides.pop(limit_key, None)
+        else:
+            overrides[limit_key] = value
+
+        updated = replace(current, version=current.version + 1, updated_at=at)
+        self._tenants[resolved] = updated
+        self._guardrail_changes.append(change)
+        self._receipts[(resolved, idempotency_key)] = (request_fingerprint, updated)
+        self._emit(
+            "guardrail_changed",
+            resolved,
+            {
+                "limit_key": limit_key,
+                "change_id": change_id,
+                "cleared": value is None,
+                "correlation_id": correlation_id,
+                "version": updated.version,
+            },
+        )
+        return updated, change, False
+
     def _require_active(self, resolved_id: str, what: str) -> None:
         tenant = self._tenants[resolved_id]
         if tenant.lifecycle is not TenantLifecycle.ACTIVE:

tests/test_guardrail_store_conformance.py

@@ -0,0 +1,267 @@
+"""TEN-WP-0006-T03: one guardrail persistence contract, both store backends.
+
+Parametrised over the in-memory and SQLite stores for the same reason the
+lifecycle conformance suite is: the divergences that matter here -- atomicity
+of override + audit + receipt, and the retired-tenant guard -- are exactly the
+kind a single-backend suite would miss.
+"""
+
+from datetime import UTC, datetime
+
+import pytest
+
+from tenant_engine.domain import Tenant, TenantRetiredError
+from tenant_engine.guardrail import (
+    UNLIMITED,
+    LimitKind,
+    LimitValue,
+    Provenance,
+    UnknownLimitKeyError,
+    resolve_limit,
+)
+from tenant_engine.sqlite_store import SQLiteTenantStore
+from tenant_engine.store import (
+    IdempotencyConflictError,
+    InMemoryTenantStore,
+    VersionConflictError,
+)
+
+NOW = datetime(2026, 8, 16, 12, 0, tzinfo=UTC)
+KEY = "spend.monthly"
+
+
+@pytest.fixture(params=["memory", "sqlite"])
+def store(request, tmp_path):
+    if request.param == "memory":
+        return InMemoryTenantStore()
+    return SQLiteTenantStore(str(tmp_path / "tenant.db"))
+
+
+@pytest.fixture
+def tenant(store) -> Tenant:
+    t = Tenant.create(tenant_id="t-1", identifier="tenant:small:acme", created_at=NOW)
+    store.create_tenant(t)
+    return t
+
+
+def eur(amount) -> LimitValue:
+    return LimitValue(kind=LimitKind.SPEND, amount=amount, currency="EUR", period="P1M")
+
+
+def set_override(store, tenant, value, *, version=1, key=KEY, **kwargs):
+    params = dict(
+        tenant_id=tenant.tenant_id,
+        expected_version=version,
+        limit_key=key,
+        value=value,
+        change_id="c-1",
+        changed_by="ops",
+        reason="test",
+        correlation_id="corr-1",
+        idempotency_key="idem-1",
+        request_fingerprint="fp-1",
+        at=NOW,
+    )
+    params.update(kwargs)
+    return store.set_guardrail_override(**params)
+
+
+# --- No backfill needed ---------------------------------------------------
+
+
+def test_a_tenant_with_no_override_resolves_to_its_grouping_default(store, tenant):
+    # Grouping defaults are resolved, never materialised into rows. That is
+    # why this workplan needs no backfill migration: a tenant created before
+    # guardrails existed and one created after take the identical path, so an
+    # existing tenant cannot silently end up looser than a fresh one.
+    assert store.guardrail_overrides(tenant.tenant_id) == {}
+    effective = resolve_limit(KEY, tenant=store.get_tenant(tenant.tenant_id))
+    assert effective.provenance is Provenance.GROUPING
+
+
+def test_the_migration_is_idempotent(tmp_path):
+    path = str(tmp_path / "tenant.db")
+    first = SQLiteTenantStore(path)
+    t = Tenant.create(tenant_id="t-1", identifier="tenant:small:acme", created_at=NOW)
+    first.create_tenant(t)
+    set_override(first, t, eur(9_000))
+
+    # forward-only and idempotent: reopening runs the same schema script
+    second = SQLiteTenantStore(path)
+    assert second.guardrail_overrides("t-1")[KEY].amount == 9_000
+    assert second.get_tenant("t-1").version == 2
+
+
+# --- Write, read back, audit ---------------------------------------------
+
+
+def test_setting_an_override_persists_it_and_bumps_the_version(store, tenant):
+    updated, change, replayed = set_override(store, tenant, eur(9_000))
+    assert not replayed
+    assert updated.version == 2
+    assert store.guardrail_overrides(tenant.tenant_id)[KEY].amount == 9_000
+    assert change.previous is None
+    assert change.current.amount == 9_000
+
+
+def test_the_override_wins_over_the_grouping_default(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    effective = resolve_limit(
+        KEY,
+        tenant=store.get_tenant(tenant.tenant_id),
+        overrides=store.guardrail_overrides(tenant.tenant_id),
+    )
+    assert effective.provenance is Provenance.OVERRIDE
+    assert effective.value.amount == 9_000
+
+
+def test_the_change_is_audited_with_actor_reason_and_correlation(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    changes = store.guardrail_changes(tenant.tenant_id)
+    assert len(changes) == 1
+    assert (changes[0].changed_by, changes[0].reason, changes[0].correlation_id) == (
+        "ops",
+        "test",
+        "corr-1",
+    )
+
+
+def test_the_audit_trail_is_append_only(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    set_override(store, tenant, eur(1_000), version=2, change_id="c-2", idempotency_key="idem-2")
+    changes = store.guardrail_changes(tenant.tenant_id)
+    assert [c.change_id for c in changes] == ["c-1", "c-2"]
+    # the earlier value survives in the record, not just the current one
+    assert changes[1].previous.amount == 9_000
+
+
+def test_a_guardrail_change_emits_a_domain_event(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    events = [e for e in store.events() if e.event_type == "guardrail_changed"]
+    assert len(events) == 1
+    assert events[0].payload["limit_key"] == KEY
+    assert events[0].payload["correlation_id"] == "corr-1"
+
+
+def test_clearing_an_override_falls_back_to_the_grouping_default(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    _, change, _ = set_override(
+        store, tenant, None, version=2, change_id="c-2", idempotency_key="idem-2"
+    )
+    assert change.is_clear
+    assert store.guardrail_overrides(tenant.tenant_id) == {}
+    effective = resolve_limit(KEY, tenant=store.get_tenant(tenant.tenant_id))
+    assert effective.provenance is Provenance.GROUPING
+
+
+def test_unlimited_survives_a_round_trip_as_an_explicit_value(store, tenant):
+    entity = LimitValue(kind=LimitKind.ENTITY_COUNT, amount=UNLIMITED)
+    # spend.monthly is the only registered key, so use it to prove the
+    # sentinel serialises; the kind check lives in the domain tests
+    set_override(store, tenant, LimitValue(
+        kind=LimitKind.SPEND, amount=UNLIMITED, currency="EUR", period="P1M"
+    ))
+    stored = store.guardrail_overrides(tenant.tenant_id)[KEY]
+    assert stored.is_unlimited
+    assert entity.is_unlimited
+
+
+# --- Concurrency, idempotency, unknown keys ------------------------------
+
+
+def test_a_stale_version_conflicts(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    with pytest.raises(VersionConflictError):
+        set_override(store, tenant, eur(1_000), version=1, change_id="c-2",
+                     idempotency_key="idem-2")
+
+
+def test_replay_returns_the_original_result_without_reapplying(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    updated, change, replayed = set_override(store, tenant, eur(9_000))
+    assert replayed
+    assert updated.version == 2  # not bumped a second time
+    assert change.change_id == "c-1"
+    assert len(store.guardrail_changes(tenant.tenant_id)) == 1
+
+
+def test_reusing_a_key_for_a_different_request_conflicts(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    with pytest.raises(IdempotencyConflictError):
+        set_override(store, tenant, eur(1_000), version=2, request_fingerprint="fp-2")
+
+
+def test_an_unregistered_key_is_rejected_before_anything_is_written(store, tenant):
+    with pytest.raises(UnknownLimitKeyError):
+        set_override(store, tenant, eur(9_000), key="spend.weekly")
+    assert store.get_tenant(tenant.tenant_id).version == 1
+    assert store.guardrail_changes(tenant.tenant_id) == []
+
+
+def test_a_failed_write_leaves_no_audit_record_and_no_version_bump(store, tenant):
+    set_override(store, tenant, eur(9_000))
+    with pytest.raises(VersionConflictError):
+        set_override(store, tenant, eur(1), version=99, change_id="c-2",
+                     idempotency_key="idem-2")
+    assert store.get_tenant(tenant.tenant_id).version == 2
+    assert len(store.guardrail_changes(tenant.tenant_id)) == 1
+    assert store.guardrail_overrides(tenant.tenant_id)[KEY].amount == 9_000
+
+
+# --- Retired tenants ------------------------------------------------------
+
+
+@pytest.fixture
+def retired(store, tenant) -> Tenant:
+    updated, _ = store.mutate_tenant(
+        tenant_id=tenant.tenant_id,
+        expected_version=1,
+        mutate=lambda t: t.retire(at=NOW),
+        event_type="tenant_retired",
+        evidence={"reason": "test"},
+        idempotency_key="retire-1",
+        request_fingerprint="fp-retire",
+    )
+    return updated
+
+
+def test_a_retired_tenants_limits_clamp_to_the_floor_but_stay_readable(store, retired):
+    effective = resolve_limit(
+        KEY,
+        tenant=store.get_tenant(retired.tenant_id),
+        overrides=store.guardrail_overrides(retired.tenant_id),
+    )
+    assert effective.provenance is Provenance.LIFECYCLE
+    assert effective.value.amount == 0
+
+
+def test_loosening_a_retired_tenants_guardrail_is_refused(store, retired):
+    # small's default is 25_000; 90_000 would take effect on reactivation
+    with pytest.raises(TenantRetiredError):
+        set_override(store, retired, eur(90_000), version=2)
+    assert store.guardrail_overrides(retired.tenant_id) == {}
+
+
+def test_tightening_a_retired_tenants_guardrail_is_allowed(store, retired):
+    # reduce-privilege operations stay available while retired -- refusing
+    # them would be the fail-open choice
+    _, change, _ = set_override(store, retired, eur(100), version=2)
+    assert change.current.amount == 100
+
+
+def test_clearing_an_override_that_would_loosen_is_refused_while_retired(store, tenant):
+    set_override(store, tenant, eur(100))
+    store.mutate_tenant(
+        tenant_id=tenant.tenant_id,
+        expected_version=2,
+        mutate=lambda t: t.retire(at=NOW),
+        event_type="tenant_retired",
+        evidence={},
+        idempotency_key="retire-1",
+        request_fingerprint="fp-retire",
+    )
+    # clearing would fall back to small's 25_000 default -- a loosening
+    with pytest.raises(TenantRetiredError):
+        set_override(store, tenant, None, version=3, change_id="c-3",
+                     idempotency_key="idem-3")
+    assert store.guardrail_overrides(tenant.tenant_id)[KEY].amount == 100

workplans/TEN-WP-0006-guardrail-quota-policy.md

@@ -168,7 +168,7 @@ folded into this diff.
 
 ```task
 id: TEN-WP-0006-T03
-status: todo
+status: done
 priority: high
 state_hub_task_id: "f1c93573-6322-4505-b738-7d66d67e60a8"
 ```
@@ -191,6 +191,46 @@ Done when the store-conformance suite (already parametrised over both backends)
 covers guardrails, so the durable store cannot diverge from the reference
 semantics.
 
+Done 2026-08-16: `TenantStore` gains `guardrail_overrides`, `guardrail_changes`,
+and `set_guardrail_override`, implemented in both backends.
+`tests/test_guardrail_store_conformance.py` parametrises 35 tests over
+in-memory and SQLite. 201 tests total, all passing.
+
+Deviations and decisions, both deliberate:
+
+- **No backfill migration, because grouping defaults are resolved rather than
+  materialised.** The task anticipated backfilling existing tenants with their
+  grouping-derived defaults. Storing them would have been worse: it duplicates
+  a value that already has a single source, and it makes changing a default a
+  data migration instead of a config edit. Since a tenant with no override
+  resolves through the same function whether it predates guardrails or not,
+  "existing tenants must not silently gain a looser limit than a fresh tenant
+  of the same grouping" holds by construction, not by a migration step that
+  could be skipped. The forward-only, idempotent part is the two new tables,
+  created by the same `CREATE TABLE IF NOT EXISTS` script fresh and existing
+  databases both take — no second schema definition to drift.
+- **A guardrail change bumps the tenant version.** It reuses the lifecycle
+  ETag, so a guardrail write invalidates a stale reader's `If-Match` exactly
+  as a metadata edit does, rather than introducing a second, separately
+  versioned resource that a caller could race against the first.
+
+Retired tenants, following the TEN-WP-0005 reduce-privilege precedent:
+
+- Guardrails stay **readable**; it is the values that clamp, not the endpoint.
+- A retired tenant's effective limits clamp to the floor, provenance
+  `lifecycle`.
+- **Tightening is allowed while retired, loosening is refused.** The check
+  compares the before/after value computed *as if the tenant were active* —
+  because the clamp makes every candidate look inert today, and the real
+  effect of a loosening override lands the moment the tenant is reactivated.
+  That deferred loosening is the thing worth refusing. Clearing an override is
+  evaluated the same way, so clearing a tightening override on a retired
+  tenant is refused too, since fall-through would loosen.
+
+Stored amounts are `TEXT NOT NULL` with an explicit `"unlimited"` token, never
+NULL. A NULL that meant unlimited would make an open ceiling the result of
+absence — the one thing the contract forbids.
+
 ## T04 - Expose guardrail read and write APIs
 
 ```task