feat: budget threshold notification tests (#470) and marketplace catalog TTL cache (#377)

.env.example

@@ -116,6 +116,8 @@ PLUGIN_MARKETPLACE_CATALOG=plugins/marketplace.json
 # policy's timeout and size cap, refuses redirects, and performs no request
 # until those checks pass. Empty disables remote catalog fetching.
 PLUGIN_MARKETPLACE_URL=
+# Seconds a parsed catalog is cached in memory (0 disables caching).
+PLUGIN_MARKETPLACE_CACHE_TTL=60
 
 # Read-only plugin discovery directory used by GET /plugins/discover. It is
 # scanned for manifest.json files only; nothing is installed or executed.

README.md

@@ -67,6 +67,7 @@ fail-closed, so a fresh install cannot reach a marketplace or the network:
 | `PLUGIN_MARKETPLACE_ENABLED` | `1` | Read-only discovery is available, but the shipped catalog is empty. |
 | `PLUGIN_MARKETPLACE_CATALOG` | `plugins/marketplace.json` | Offline catalog file; entries must validate against `plugins/plugin.schema.json`. |
 | `PLUGIN_MARKETPLACE_URL` | empty | Remote marketplace fetching is disabled; when set it must be HTTPS, allowlisted, timeout/size-bounded, and redirect-free. |
+| `PLUGIN_MARKETPLACE_CACHE_TTL` | `60` | Seconds a parsed catalog is cached in memory per source; `0` disables caching. Call `reset_marketplace_cache()` to bust it. |
 | `PLUGIN_DISCOVERY_DIR` | `plugins` | Read-only directory scanned for manifests by `GET /plugins/discover`; nothing is installed or executed. |
 | `PLUGIN_TRUST_POLICY` | `if-present` | Accepts unsigned local plugins, but always refuses forged or untrusted signatures. Use `required` in production. |
 | `PLUGIN_TRUSTED_KEYS` | `{}` | No publisher is trusted by default; keys are validated at startup. |

app/config.py

@@ -181,6 +181,10 @@ class Config:
     # Empty by default, so remote fetching stays opt-in and fail closed.
     PLUGIN_MARKETPLACE_URL = os.getenv("PLUGIN_MARKETPLACE_URL", "")
 
+    # Seconds a successfully parsed catalog is served from memory before the
+    # source is read again (#377). ``0`` disables caching.
+    PLUGIN_MARKETPLACE_CACHE_TTL = float(os.getenv("PLUGIN_MARKETPLACE_CACHE_TTL", "60"))
+
     # Directory scanned by the read-only plugin discovery endpoint (#172).
     # Discovery only reads ``manifest.json`` files from plugin subdirectories
     # and never installs, loads, or mutates anything. Defaults to the repo's

app/services/marketplace.py

@@ -36,6 +36,7 @@
 
 import json
 import threading
+import time
 from collections.abc import Mapping
 from dataclasses import dataclass
 from functools import lru_cache
@@ -236,10 +237,35 @@ def _catalog_source(config: Mapping[str, Any]) -> tuple[str, str]:
 _CATALOG_CACHE_LOCK = threading.Lock()
 
 
+#: Default lifetime, in seconds, of a cached lenient catalog load (#377).
+DEFAULT_CACHE_TTL = 60.0
+
+#: Process-local TTL cache of successful lenient loads, keyed by
+#: ``(source kind, source)``. Values are ``(expires_at, result)`` using
+#: :func:`time.monotonic`. Only ``ok``/``empty`` results are cached; stale and
+#: unavailable fallbacks always retry the source.
+_TTL_CACHE: dict[tuple[str, str], tuple[float, MarketplaceLoadResult]] = {}
+
+
+def cache_ttl(config: Mapping[str, Any]) -> float:
+    """Return the configured cache TTL in seconds (``0`` disables caching)."""
+    raw = config.get("PLUGIN_MARKETPLACE_CACHE_TTL", DEFAULT_CACHE_TTL)
+    try:
+        ttl = float(raw)
+    except (TypeError, ValueError):
+        return DEFAULT_CACHE_TTL
+    return ttl if ttl > 0 else 0.0
+
+
 def reset_marketplace_cache() -> None:
-    """Forget cached catalogs (tests and explicit configuration changes)."""
+    """Forget cached catalogs (tests, operators and configuration changes).
+
+    This is the explicit cache-bust path: it clears both the TTL cache and the
+    last-known-good fallbacks.
+    """
     with _CATALOG_CACHE_LOCK:
         _LAST_GOOD_CATALOGS.clear()
+        _TTL_CACHE.clear()
 
 
 def _remember_catalog(kind: str, value: str, catalog: MarketplaceCatalog) -> None:
@@ -376,8 +402,30 @@ def load_marketplace_catalog(
     if not config.get("PLUGIN_MARKETPLACE_ENABLED", True):
         raise MarketplaceDisabledError("The plugin marketplace is disabled.")
 
-    warnings: list[str] = []
     kind, value = _catalog_source(config)
+    ttl = cache_ttl(config)
+    if ttl:
+        with _CATALOG_CACHE_LOCK:
+            hit = _TTL_CACHE.get((kind, value))
+        if hit is not None and time.monotonic() < hit[0]:
+            return hit[1]
+
+    result = _load_marketplace_uncached(kind, value, config, resolver, transport)
+    if ttl and result.state in ("ok", "empty"):
+        with _CATALOG_CACHE_LOCK:
+            _TTL_CACHE[(kind, value)] = (time.monotonic() + ttl, result)
+    return result
+
+
+def _load_marketplace_uncached(
+    kind: str,
+    value: str,
+    config: Mapping[str, Any],
+    resolver: Any,
+    transport: Any,
+) -> MarketplaceLoadResult:
+    """Read and validate the source (the cache-miss path of the lenient loader)."""
+    warnings: list[str] = []
 
     if kind == "remote":
         try:
@@ -439,6 +487,7 @@ def filter_entries(
 
 
 __all__ = [
+    "DEFAULT_CACHE_TTL",
     "DEFAULT_CATALOG_PATH",
     "MANIFEST_SCHEMA_PATH",
     "MAX_MARKETPLACE_WARNINGS",
@@ -449,6 +498,7 @@ def filter_entries(
     "MarketplaceError",
     "MarketplaceLoadResult",
     "MarketplaceUnavailableError",
+    "cache_ttl",
     "filter_entries",
     "load_marketplace",
     "load_marketplace_catalog",

tests/test_budget_threshold_notifications.py

@@ -0,0 +1,79 @@
+"""Budget threshold notifications honour preferences (Phase 10, #470)."""
+
+from __future__ import annotations
+
+from app.extensions import db
+from app.models import Notification, NotificationPreference, Workspace
+from app.services import budgets, usage_ledger
+
+
+def _token_budget(user, limit, *, threshold=0.5):
+    return budgets.upsert_budget(
+        scope="user", scope_id=user.id, token_limit=limit, warn_threshold=threshold
+    )
+
+
+def _levels(user):
+    rows = Notification.query.filter_by(user_id=user.id, type="budget_alert").all()
+    return sorted(row.payload["level"] for row in rows)
+
+
+def test_notifications_fire_at_warning_and_limit_thresholds(app, make_user):
+    user = make_user()
+    _token_budget(user, 10)
+
+    usage_ledger.record_usage(source="chat", user=user, total_tokens=5)
+    assert _levels(user) == ["warning"]
+
+    usage_ledger.record_usage(source="chat", user=user, total_tokens=5)
+    assert _levels(user) == ["blocked", "warning"]
+
+
+def test_opted_out_user_receives_neither_threshold_notification(app, make_user):
+    user = make_user()
+    db.session.add(NotificationPreference(user_id=user.id, budget_alerts=False))
+    db.session.commit()
+    _token_budget(user, 10)
+
+    usage_ledger.record_usage(source="chat", user=user, total_tokens=5)
+    usage_ledger.record_usage(source="chat", user=user, total_tokens=5)
+
+    assert _levels(user) == []
+    assert Notification.query.filter_by(type="budget_alert").count() == 0
+
+
+def test_workspace_budget_notifies_owner_and_respects_owner_preference(app, make_user):
+    owner = make_user(email="owner@example.com", username="owner")
+    member = make_user(email="member@example.com", username="member")
+    workspace = Workspace(user_id=owner.id, name="Team")
+    db.session.add(workspace)
+    db.session.add(NotificationPreference(user_id=owner.id, budget_alerts=False))
+    db.session.commit()
+    budgets.upsert_budget(
+        scope="workspace",
+        scope_id=workspace.id,
+        token_limit=10,
+        warn_threshold=0.5,
+    )
+
+    usage_ledger.record_usage(source="chat", user=member, workspace=workspace, total_tokens=6)
+
+    assert _levels(owner) == []  # owner opted out
+    assert _levels(member) == ["warning"]
+
+
+def test_workspace_budget_notifies_owner_by_default(app, make_user):
+    owner = make_user(email="owner2@example.com", username="owner2")
+    member = make_user(email="member2@example.com", username="member2")
+    workspace = Workspace(user_id=owner.id, name="Team")
+    db.session.add(workspace)
+    db.session.commit()
+    budgets.upsert_budget(
+        scope="workspace", scope_id=workspace.id, token_limit=10, warn_threshold=0.5
+    )
+
+    usage_ledger.record_usage(source="chat", user=member, workspace=workspace, total_tokens=10)
+
+    # Jumping straight past the limit reports the more severe level once.
+    assert _levels(owner) == ["blocked"]
+    assert _levels(member) == ["blocked"]

tests/test_marketplace_cache_ttl.py

@@ -0,0 +1,133 @@
+"""Marketplace catalog TTL cache (Phase 9, #377)."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from app.services import marketplace
+from app.services.marketplace import (
+    cache_ttl,
+    load_marketplace_catalog,
+    reset_marketplace_cache,
+)
+
+HASH = "a" * 64
+
+
+def _entry(plugin_id: str) -> dict:
+    return {
+        "id": plugin_id,
+        "version": "1.0.0",
+        "source": f"https://plugins.example.com/{plugin_id}.zip",
+        "integrity": f"sha256:{HASH}",
+        "manifest": {
+            "id": plugin_id,
+            "name": f"Plugin {plugin_id}",
+            "version": "1.0.0",
+            "description": "A plugin",
+            "author": "Acme",
+            "entry_point": "plugins.test:TestPlugin",
+            "capabilities": ["PROJECT_READ"],
+        },
+    }
+
+
+def _write(path, *ids: str) -> None:
+    path.write_text(
+        json.dumps({"version": 1, "plugins": [_entry(i) for i in ids]}), encoding="utf-8"
+    )
+
+
+@pytest.fixture()
+def clock(monkeypatch):
+    now = [1000.0]
+    monkeypatch.setattr(marketplace.time, "monotonic", lambda: now[0])
+    return now
+
+
+def _config(path, ttl):
+    return {
+        "PLUGIN_MARKETPLACE_ENABLED": True,
+        "PLUGIN_MARKETPLACE_CATALOG": str(path),
+        "PLUGIN_MARKETPLACE_CACHE_TTL": ttl,
+    }
+
+
+def _ids(result):
+    return [entry.plugin_id for entry in result.catalog.list_entries()]
+
+
+def test_miss_reads_source_then_hit_avoids_rereading(tmp_path, clock):
+    path = tmp_path / "catalog.json"
+    _write(path, "alpha")
+    config = _config(path, 60)
+
+    assert _ids(load_marketplace_catalog(config)) == ["alpha"]
+
+    # The source changes (and even disappears) but the cache is still fresh.
+    _write(path, "alpha", "beta")
+    clock[0] += 59
+    assert _ids(load_marketplace_catalog(config)) == ["alpha"]
+    path.unlink()
+    assert _ids(load_marketplace_catalog(config)) == ["alpha"]
+
+
+def test_entry_expires_after_ttl(tmp_path, clock):
+    path = tmp_path / "catalog.json"
+    _write(path, "alpha")
+    config = _config(path, 60)
+    load_marketplace_catalog(config)
+
+    _write(path, "alpha", "beta")
+    clock[0] += 60
+    assert _ids(load_marketplace_catalog(config)) == ["alpha", "beta"]
+
+
+def test_cache_is_keyed_by_source(tmp_path, clock):
+    first, second = tmp_path / "one.json", tmp_path / "two.json"
+    _write(first, "alpha")
+    _write(second, "beta")
+
+    assert _ids(load_marketplace_catalog(_config(first, 60))) == ["alpha"]
+    assert _ids(load_marketplace_catalog(_config(second, 60))) == ["beta"]
+
+
+def test_zero_ttl_disables_caching(tmp_path, clock):
+    path = tmp_path / "catalog.json"
+    _write(path, "alpha")
+    config = _config(path, 0)
+    load_marketplace_catalog(config)
+
+    _write(path, "alpha", "beta")
+    assert _ids(load_marketplace_catalog(config)) == ["alpha", "beta"]
+
+
+def test_reset_busts_the_cache(tmp_path, clock):
+    path = tmp_path / "catalog.json"
+    _write(path, "alpha")
+    config = _config(path, 60)
+    load_marketplace_catalog(config)
+
+    _write(path, "alpha", "beta")
+    reset_marketplace_cache()
+    assert _ids(load_marketplace_catalog(config)) == ["alpha", "beta"]
+
+
+def test_unavailable_source_is_not_cached(tmp_path, clock):
+    path = tmp_path / "missing.json"
+    config = _config(path, 60)
+    assert load_marketplace_catalog(config).state == "unavailable"
+
+    _write(path, "alpha")
+    assert _ids(load_marketplace_catalog(config)) == ["alpha"]
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [(None, 60.0), ("30", 30.0), (0, 0.0), (-5, 0.0), ("bad", 60.0)],
+)
+def test_cache_ttl_parsing(raw, expected):
+    config = {} if raw is None else {"PLUGIN_MARKETPLACE_CACHE_TTL": raw}
+    assert cache_ttl(config) == expected

tests/test_marketplace_fallbacks.py

@@ -53,12 +53,15 @@ def _config(path) -> dict:
         "PLUGIN_MARKETPLACE_ENABLED": True,
         "PLUGIN_MARKETPLACE_CATALOG": str(path),
         "PLUGIN_MARKETPLACE_URL": "",
+        # These tests exercise the last-good fallback, which needs a re-read.
+        "PLUGIN_MARKETPLACE_CACHE_TTL": 0,
     }
 
 
 def _login(app, client, make_user, login):
     make_user(username="member", email="member@example.com")
     app.config["PLUGIN_MARKETPLACE_ENABLED"] = True
+    app.config["PLUGIN_MARKETPLACE_CACHE_TTL"] = 0  # fallback tests re-read the source
     login(email="member@example.com")