fix(permits,state): enforce expiry on the second reservation path; make budget release once-only without a row lock; refuse SQLite in production

app/core/durable_state.py

@@ -181,6 +181,14 @@ def _resolve_backend(self) -> str:
                 return "redis"
             return self._missing_explicit_backend("redis", "REDIS_URL")
         if self._state_backend == "sqlite":
+            if self._production_like:
+                raise DurableStateConfigError(
+                    "STATE_BACKEND=sqlite is not allowed in production-like "
+                    "environments: the trust plane's money and permit paths "
+                    "serialize concurrent writers with SELECT ... FOR UPDATE, "
+                    "which SQLAlchemy silently drops on SQLite. Use "
+                    "STATE_BACKEND=postgres with DATABASE_URL."
+                )
             if self._sqlite_url:
                 return "sqlite"
             return self._missing_explicit_backend("sqlite", "SQLITE_URL")
@@ -197,12 +205,16 @@ def _resolve_backend(self) -> str:
             return "postgres"
         if self._redis_url:
             return "redis"
-        if self._sqlite_url:
+        if self._sqlite_url and not self._production_like:
             return "sqlite"
         if self._production_like:
+            # SQLITE_URL is deliberately not a production fallback here: see the
+            # explicit STATE_BACKEND=sqlite branch above for why row locks make
+            # SQLite unsafe for the money and permit paths.
             raise DurableStateConfigError(
-                "Production-like environments require DATABASE_URL, REDIS_URL, "
-                "or SQLITE_URL for durable state"
+                "Production-like environments require DATABASE_URL or REDIS_URL "
+                "for durable state (SQLite is not a supported production "
+                "backend)"
             )
         return "memory"
 

app/services/permits.py

@@ -742,6 +742,10 @@ async def _once() -> None:
                     now = utc_now()
                     # Atomic guarded reserve (see authorize_and_reserve): the cap
                     # is enforced by the WHERE clause, not a read-then-write.
+                    # The expiry term matters for the same reason it does there:
+                    # an expired permit keeps status="active" in storage, so
+                    # without it this second reservation path would still spend
+                    # against a permit that has crossed expires_at.
                     reserved = await session.execute(
                         sa_update(PermitModel)
                         .where(
@@ -753,6 +757,10 @@ async def _once() -> None:
                                 ColumnElement[bool],
                                 PermitModel.status == "active",
                             ),
+                            cast(
+                                ColumnElement[bool],
+                                PermitModel.expires_at > now,
+                            ),
                             cast(
                                 ColumnElement[bool],
                                 PermitModel.spent_credits + amount
@@ -769,6 +777,10 @@ async def _once() -> None:
                         exists = await session.get(PermitModel, permit_id)
                         if exists is None:
                             raise PermitError("permit_not_found")
+                        if exists.status != "active":
+                            raise PermitError("permit_not_active")
+                        if exists.expires_at <= now:
+                            raise PermitError("permit_expired")
                         raise PermitError("permit_budget_exceeded")
 
                     # Budget percentage alerts, recomputed from the committed
@@ -869,6 +881,35 @@ async def _once() -> bool:
                     if attempt.budget_released_at is not None:
                         return False
                     now = utc_now()
+                    # Claim the release atomically before touching the permit.
+                    # The read above is guarded by a row lock, but SQLAlchemy
+                    # silently drops FOR UPDATE on engines that do not support
+                    # it (SQLite), so on those two concurrent callers would both
+                    # observe budget_released_at IS NULL and both decrement.
+                    # This guarded UPDATE is the once-only gate on every engine:
+                    # exactly one caller can flip NULL -> now, and only that
+                    # caller proceeds to release the budget.
+                    claimed = await session.execute(
+                        sa_update(McpDispatchAttemptModel)
+                        .where(
+                            cast(
+                                ColumnElement[bool],
+                                McpDispatchAttemptModel.attempt_id == attempt_id,
+                            ),
+                            cast(
+                                ColumnElement[bool],
+                                cast(
+                                    Any, McpDispatchAttemptModel.budget_released_at
+                                ).is_(None),
+                            ),
+                        )
+                        .values(budget_released_at=now, updated_at=now)
+                        .execution_options(synchronize_session=False)
+                    )
+                    if (cast(Any, claimed).rowcount or 0) != 1:
+                        # Another caller claimed it first; its transaction owns
+                        # the single decrement.
+                        return False
                     # Atomic clamped decrement so a concurrent reservation on the
                     # same permit is not clobbered by a read-modify-write here.
                     released = await session.execute(
@@ -899,6 +940,9 @@ async def _once() -> bool:
                     )
                     if (cast(Any, released).rowcount or 0) == 0:
                         raise PermitError("permit_not_found")
+                    # budget_released_at was already set by the guarded claim
+                    # above; refresh the identity-mapped instance so callers
+                    # holding it observe the committed value.
                     attempt.budget_released_at = now
                     attempt.updated_at = now
                     session.add(attempt)

tests/test_db_urls_durable_state.py

@@ -208,6 +208,14 @@ def test_sqlite_without_url_falls_back_outside_production(monkeypatch):
 
 
 def test_sqlite_without_url_refuses_production_like(monkeypatch):
+    """Production still refuses this config; the reason is now stricter.
+
+    This previously failed only because SQLITE_URL was unset, which implied a
+    configured SQLITE_URL would have been accepted. SQLite is now refused in
+    production-like environments outright, so the refusal no longer depends on
+    the URL being missing -- see
+    ``test_sqlite_with_url_refuses_production_like`` for the configured case.
+    """
     reset_runtime_degradation()
     reset_durable_state_for_tests()
     monkeypatch.setenv("ENVIRONMENT", "production")
@@ -218,7 +226,7 @@ def test_sqlite_without_url_refuses_production_like(monkeypatch):
 
     get_settings.cache_clear()
     store = DurableStateStore()
-    with pytest.raises(DurableStateConfigError, match="SQLITE_URL"):
+    with pytest.raises(DurableStateConfigError, match="not allowed in production"):
         store._resolve_backend()
     get_settings.cache_clear()
     reset_durable_state_for_tests()
@@ -663,3 +671,55 @@ def test_close_does_not_block_on_a_connection_inherited_across_fork(tmp_path):
         f"parent could not close its own connection afterwards: {proc.stdout!r}"
     )
     assert proc.returncode == 0, f"unexpected exit code {proc.returncode}"
+
+
+def test_sqlite_with_url_refuses_production_like(monkeypatch):
+    """SQLite is not a supported production backend, even fully configured.
+
+    The trust plane's money and permit paths serialize concurrent writers with
+    ``SELECT ... FOR UPDATE``. SQLAlchemy silently drops that on SQLite, so the
+    once-only and read-modify-write guards those paths rely on stop holding.
+    A configured SQLITE_URL previously satisfied the explicit backend with no
+    production objection at all.
+    """
+    reset_runtime_degradation()
+    reset_durable_state_for_tests()
+    monkeypatch.setenv("ENVIRONMENT", "production")
+    monkeypatch.setenv("STATE_BACKEND", "sqlite")
+    monkeypatch.setenv("SQLITE_URL", "/var/lib/agent-middleware/state.db")
+    monkeypatch.setenv("DATABASE_URL", "")
+    from app.core.config import get_settings
+
+    get_settings.cache_clear()
+    store = DurableStateStore()
+    with pytest.raises(DurableStateConfigError, match="not allowed in production"):
+        store._resolve_backend()
+    get_settings.cache_clear()
+    reset_durable_state_for_tests()
+    reset_runtime_degradation()
+
+
+def test_auto_backend_refuses_sqlite_in_production_like(monkeypatch):
+    """Auto-resolution must not quietly land on SQLite in production either.
+
+    A SQLite ``DATABASE_URL`` is the ordinary local-development setting, and
+    auto-resolution maps it to the sqlite backend. In a production-like
+    environment that turned a misconfigured DSN into an unsafe backend without
+    a word; it must fail closed instead.
+    """
+    reset_runtime_degradation()
+    reset_durable_state_for_tests()
+    monkeypatch.setenv("ENVIRONMENT", "production")
+    monkeypatch.setenv("STATE_BACKEND", "auto")
+    monkeypatch.setenv("SQLITE_URL", "")
+    monkeypatch.setenv("REDIS_URL", "")
+    monkeypatch.setenv("DATABASE_URL", "sqlite+aiosqlite:///./prod.db")
+    from app.core.config import get_settings
+
+    get_settings.cache_clear()
+    store = DurableStateStore()
+    with pytest.raises(DurableStateConfigError, match="SQLite is not a supported"):
+        store._resolve_backend()
+    get_settings.cache_clear()
+    reset_durable_state_for_tests()
+    reset_runtime_degradation()

tests/test_mcp_dispatch_reconciliation.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import json
 from dataclasses import dataclass
 from datetime import timedelta
@@ -1010,3 +1011,50 @@ async def test_lost_commit_ack_recovery_no_double_charge(
     permit_after_retry = await get_permit_service().get_permit(permit["permit_id"])
     assert permit_after_retry is not None
     assert permit_after_retry.spent_credits == spent_after_first
+
+
+@pytest.mark.anyio
+async def test_concurrent_budget_release_decrements_exactly_once(
+    client: AsyncClient,
+    clean_database,
+) -> None:
+    """The release must be once-only without relying on a row lock.
+
+    ``release_dispatch_budget_once`` read ``budget_released_at`` under
+    ``with_for_update=True`` and then decremented the permit. SQLAlchemy
+    silently drops FOR UPDATE on engines that do not support it, so on those
+    two concurrent callers could both observe NULL and both decrement -- the
+    permit would be credited back twice for one dispatch. The claim is now a
+    guarded UPDATE, which is once-only on every engine.
+    """
+    seed = await _seed_attempt(
+        client,
+        suffix="concurrent-release",
+        state="returned_error",
+        result_payload={"error": "confirmed"},
+        error_code="upstream_returned_error",
+    )
+    permits = get_permit_service()
+    before = await permits.get_permit(seed.permit_id)
+    assert before is not None
+    spent_before = before.spent_credits
+
+    results = await asyncio.gather(
+        permits.release_dispatch_budget_once(seed.attempt_id),
+        permits.release_dispatch_budget_once(seed.attempt_id),
+        return_exceptions=True,
+    )
+    ok = [r for r in results if r is True]
+    errors = [r for r in results if isinstance(r, BaseException)]
+    assert not errors, f"release raised: {errors}"
+    assert len(ok) == 1, f"expected exactly one release to win, got {results}"
+
+    after = await permits.get_permit(seed.permit_id)
+    assert after is not None
+    # Exactly one decrement, of exactly the authorized amount.
+    assert after.spent_credits == spent_before - CREDITS
+
+    # A third call is still a no-op.
+    assert await permits.release_dispatch_budget_once(seed.attempt_id) is False
+    final = await permits.get_permit(seed.permit_id)
+    assert final is not None and final.spent_credits == after.spent_credits

tests/test_onboarding_contract.py

@@ -132,6 +132,10 @@ def test_remediation_lookup_has_a_safe_default() -> None:
 _DOC_SWEEP_EXCLUDES = (
     "site/",
     "node_modules/",
+    # Gitignored agent scratch: `.claude/worktrees/` holds full checkouts, so a
+    # sweep that walks it re-finds every excluded file under a second path and
+    # fails on a copy of a file this list already forgave.
+    ".claude/",
     "tests/test_wedge_honesty.py",
     "tests/test_onboarding_contract.py",
     "docs/tech-debt-remediation-plan.md",

tests/test_permits.py

@@ -663,3 +663,50 @@ async def test_permit_service_validation_denial_reasons(client, clean_database):
     )
     assert validation.allowed is False
     assert validation.reason == "permit_signature_invalid"
+
+
+async def test_reserve_budget_refuses_an_expired_permit(
+    client,
+    clean_database,
+    enforce_naive_utc_datetime_columns,
+):
+    """reserve_budget is a second reservation path and must enforce expiry.
+
+    An expired permit keeps ``status="active"`` in storage, so a guarded update
+    without an expiry term happily spends against it. ``authorize_and_reserve``
+    grew that term; this path is the one that did not have it.
+    """
+    provisioned = await provision_agent_wallet(client)
+    service = get_permit_service()
+    permit = await service.create_permit(
+        PermitCreateRequest(
+            issuer_wallet_id=provisioned["agent_wallet_id"],
+            subject_wallet_id=provisioned["agent_wallet_id"],
+            subject_key_id=provisioned["key_id"],
+            allowed_tools=["expiry-reserve-tool"],
+            scopes=["tool:expiry-reserve-tool:invoke"],
+            max_credits=Decimal("10"),
+            expires_at=(
+                datetime.now(timezone.utc) + timedelta(minutes=30)
+            ).replace(tzinfo=None),
+        )
+    )
+
+    # Cross expires_at without touching status, exactly as the clock would.
+    factory = get_session_factory()
+    async with factory() as session:
+        async with session.begin():
+            model = await session.get(PermitModel, permit.permit_id)
+            assert model is not None
+            model.expires_at = utc_now() - timedelta(seconds=1)
+            session.add(model)
+
+    with pytest.raises(PermitError) as excinfo:
+        await service.reserve_budget(permit.permit_id, Decimal("1"))
+    assert str(excinfo.value) == "permit_expired"
+
+    # And no budget moved.
+    async with factory() as session:
+        model = await session.get(PermitModel, permit.permit_id)
+        assert model is not None
+        assert model.spent_credits == Decimal("0")