feat: add user secrets management with encrypted storage

.gitignore

@@ -151,6 +151,8 @@ config.yml
 *.local.yaml
 *.local.yml
 secrets/
+!src/**/secrets/
+!tests/**/secrets/
 *.encrypted
 
 # Data directories

docs/architecture/data-storage-matrix.md

@@ -177,12 +177,14 @@ Map **data requiring properties** ↔ **storage providing properties**.
 | **UserModel** | Med | Low | SC | Relational (JOIN on zone_id, email lookup) | Small | Med | Persistent | System | Core user accounts with soft delete | SQLAlchemy with soft delete | **Keep SQLAlchemy** (relational queries) | ✅ KEEP |
 | **UserOAuthAccountModel** | Med | Low | SC | Relational (FK to user_id, unique constraint on provider+provider_user_id) | Small | Med | Persistent | System | OAuth provider accounts for SSO login | SQLAlchemy | **Keep SQLAlchemy** (FK, unique constraints) | ✅ KEEP |
 | **OAuthCredentialModel** | Med | Low | SC | Relational (FK to user_id, zone_id, encrypted tokens) | Small | Med | Persistent | Zone | OAuth tokens for backend integrations (Google Drive, OneDrive) | SQLAlchemy with encryption | **Keep SQLAlchemy** (FK, encryption) | ✅ KEEP |
+| **UserSecretModel** | Med | Low | SC | Relational (FK to user_id, unique constraint on user_id+zone_id+name, Fernet-encrypted value) | Small | Med | Persistent | Zone | User-managed encrypted secrets (API keys, tokens) referenced via `nexus-secret:NAME` in plugin/agent configs | SQLAlchemy with encryption | **Keep SQLAlchemy** (FK, encryption, unique constraint) | ✅ KEEP |
 | **UserSessionModel** | High | Med | EC | KV (by session_id) | Tiny | High | Session | System | Active user sessions | SQLAlchemy | **CacheStore** (Dragonfly / In-Memory) | ✅ DECIDED: CacheStore |
 
 **Analysis (Step 1+3 DECIDED):**
 - **No merges or abstractions needed** — well-designed, minimal redundancy:
   - **UserOAuthAccountModel** vs **OAuthCredentialModel**: Intentionally separate — *login auth* (ID token only) vs *backend integration* (access/refresh tokens). Different security flows.
-  - User/OAuth models: ✅ KEEP RecordStore — relational queries, FK, encryption
+  - **UserSecretModel**: User-managed encrypted key-value secrets. Distinct from OAuthCredentialModel (system-managed OAuth tokens) — different lifecycle, different crypto (standalone Fernet key vs OAuth encryption key). Scoped by (user_id, zone_id, name). Accessed via `SecretResolver` for `nexus-secret:NAME` pattern injection in plugin/agent configs. Audit trail via SecretsAuditLogModel (Part 20).
+  - User/OAuth/Secrets models: ✅ KEEP RecordStore — relational queries, FK, encryption
 - **UserSessionModel affinity (Step 3)**:
   - Required: KV by session_id, TTL expiry, high read freq, EC sufficient
   - Relational ACID (RecordStore): ✅ works, but ❌ no native TTL, ❌ overkill (no JOINs/FK needed)
@@ -351,7 +353,7 @@ individual rationale, and the Quartet section below for complete data type → p
 ### ✅ Confirmed NO-MERGE (architecture is correct):
 
 5. **ReBAC 4 types** — Zanzibar-correct: SSOT (Tuple, Namespace), Derived (GroupClosure), Audit (Changelog)
-6. **User/Auth 4 types** — Clean separation: identity (User), login auth (OAuthAccount), backend integration (OAuthCredential), sessions (UserSession)
+6. **User/Auth 5 types** — Clean separation: identity (User), login auth (OAuthAccount), backend integration (OAuthCredential), user-managed secrets (UserSecret), sessions (UserSession)
 7. **Events 3 types** — Different lifecycles: ephemeral (FileEvent), persistent config (Subscription), audit (Delivery)
 8. **CompactFileMetadata** — Cache-tier projection of FileMetadata (auto-generated from proto)
 9. **FileMetadataModel (custom KV)** — Arbitrary user-defined pairs, fundamentally different from fixed-schema FileMetadata
@@ -511,10 +513,10 @@ Data classes (`FileMetadata`, `PaginatedResult`) live in `contracts/metadata.py`
 | ReBACNamespaceModel | SQLAlchemy | Part 5 | Permission config, KV by namespace_id |
 | SystemSettingsModel | SQLAlchemy | Part 13 | System config, KV by key |
 
-**RecordStore** (Relational — PostgreSQL/SQLite) — 49 types:
+**RecordStore** (Relational — PostgreSQL/SQLite) — 50 types:
 | Category | Data Types | From Part | Rationale |
 |----------|-----------|-----------|-----------|
-| **Users & Auth** | UserModel, UserOAuthAccountModel, OAuthCredentialModel | Part 6 | FK, unique constraints, encryption |
+| **Users & Auth** | UserModel, UserOAuthAccountModel, OAuthCredentialModel, UserSecretModel | Part 6 | FK, unique constraints, encryption |
 | **ReBAC** | ReBACTupleModel, ReBACGroupClosureModel, ReBACChangelogModel | Part 5 | Composite indexes (SSOT), materialized view, append-only BRIN |
 | **Memory System** | MemoryModel, **MemoryConfig**, TrajectoryModel, TrajectoryFeedbackModel, PlaybookModel | Part 4 | Vector search (pgvector), relational FK; MemoryConfig co-exists with MemoryModel |
 | **Versioning** | VersionHistoryModel, WorkspaceSnapshotModel | Part 3 | Parent FK, BRIN time-series |

src/nexus/bricks/auth/secrets/__init__.py

@@ -0,0 +1,7 @@
+"""User secrets management — encrypted key-value storage per user/zone."""
+
+from nexus.bricks.auth.secrets.crypto import SecretsCrypto
+from nexus.bricks.auth.secrets.resolver import SecretResolver
+from nexus.bricks.auth.secrets.service import UserSecretsService
+
+__all__ = ["SecretsCrypto", "SecretResolver", "UserSecretsService"]

src/nexus/bricks/auth/secrets/crypto.py

@@ -0,0 +1,110 @@
+"""Fernet encryption for user secrets.
+
+Standalone crypto service for the secrets subsystem — does NOT share
+keys with OAuthCrypto.  Each subsystem gets its own encryption key
+stored under a distinct SystemSettings key.
+"""
+
+import logging
+from typing import TYPE_CHECKING
+
+from cryptography.fernet import Fernet, InvalidToken
+
+if TYPE_CHECKING:
+    from nexus.storage.record_store import RecordStoreABC
+
+logger = logging.getLogger(__name__)
+
+SECRETS_ENCRYPTION_KEY_NAME = "user_secrets_encryption_key"
+
+
+class SecretsCrypto:
+    """Fernet encryption service for user secrets.
+
+    Key management mirrors OAuthCrypto's pattern but uses a separate
+    ``SystemSettingsModel`` row (``user_secrets_encryption_key``) so
+    rotating one subsystem's key never affects the other.
+    """
+
+    def __init__(
+        self,
+        encryption_key: str | None = None,
+        *,
+        record_store: "RecordStoreABC | None" = None,
+    ) -> None:
+        self._session_factory = record_store.session_factory if record_store else None
+
+        if encryption_key is not None:
+            self._init_fernet(encryption_key)
+            return
+
+        if record_store:
+            db_key = self._load_or_create_key_from_db()
+            if db_key:
+                self._init_fernet(db_key)
+                return
+
+        logger.warning(
+            "Generating random secrets encryption key. This key will NOT persist "
+            "across restarts! Pass encryption_key or record_store for production use."
+        )
+        self._init_fernet(Fernet.generate_key().decode("utf-8"))
+
+    def _init_fernet(self, encryption_key: str) -> None:
+        try:
+            self._fernet = Fernet(encryption_key.encode("utf-8"))
+        except Exception as e:
+            raise ValueError(f"Invalid encryption key: {e}") from e
+
+    def _load_or_create_key_from_db(self) -> str | None:
+        try:
+            from sqlalchemy import select
+
+            from nexus.storage.models import SystemSettingsModel
+
+            if self._session_factory is None:
+                return None
+
+            with self._session_factory() as session:
+                stmt = select(SystemSettingsModel).where(
+                    SystemSettingsModel.key == SECRETS_ENCRYPTION_KEY_NAME
+                )
+                setting = session.execute(stmt).scalar_one_or_none()
+
+                if setting:
+                    return str(setting.value)
+
+                new_key = Fernet.generate_key().decode("utf-8")
+                new_setting = SystemSettingsModel(
+                    key=SECRETS_ENCRYPTION_KEY_NAME,
+                    value=new_key,
+                    description="Fernet encryption key for user secrets",
+                    is_sensitive=1,
+                )
+                session.add(new_setting)
+                session.commit()
+                logger.info("Generated and stored new secrets encryption key in database")
+                return new_key
+
+        except Exception:
+            logger.warning("Failed to load/store secrets encryption key", exc_info=True)
+            return None
+
+    @staticmethod
+    def generate_key() -> str:
+        return Fernet.generate_key().decode("utf-8")
+
+    def encrypt(self, plaintext: str) -> str:
+        if not plaintext:
+            raise ValueError("Plaintext cannot be empty")
+        return self._fernet.encrypt(plaintext.encode("utf-8")).decode("utf-8")
+
+    def decrypt(self, ciphertext: str) -> str:
+        if not ciphertext:
+            raise ValueError("Ciphertext cannot be empty")
+        try:
+            return self._fernet.decrypt(ciphertext.encode("utf-8")).decode("utf-8")
+        except InvalidToken as e:
+            raise InvalidToken(
+                "Failed to decrypt secret. Value may be corrupted or key may have changed."
+            ) from e

src/nexus/bricks/auth/secrets/resolver.py

@@ -0,0 +1,86 @@
+"""Secret resolver — scans configs for nexus-secret:NAME patterns and injects values.
+
+Used by PluginRegistry and agent config loading to resolve secret references
+into actual decrypted values before execution.
+"""
+
+import logging
+import re
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Pattern: nexus-secret:SECRET_NAME
+SECRET_PATTERN = re.compile(r"nexus-secret:([A-Za-z0-9_.\-]+)")
+
+
+class SecretResolver:
+    """Resolves nexus-secret:NAME references in configuration dicts/strings.
+
+    Args:
+        secrets_service: UserSecretsService instance for value lookups.
+        user_id: The user whose secrets to resolve.
+        zone_id: The zone scope for secret lookups.
+    """
+
+    def __init__(
+        self,
+        secrets_service: Any,
+        user_id: str,
+        zone_id: str | None = None,
+    ) -> None:
+        from nexus.contracts.constants import ROOT_ZONE_ID
+
+        self._service = secrets_service
+        self._user_id = user_id
+        self._zone_id = zone_id or ROOT_ZONE_ID
+
+    def resolve_string(self, value: str) -> str:
+        """Replace all nexus-secret:NAME patterns in a string with decrypted values.
+
+        If a secret is not found, the pattern is left unchanged and a warning is logged.
+        """
+
+        def _replacer(match: re.Match[str]) -> str:
+            secret_name = match.group(1)
+            resolved = self._service.get_secret_value(
+                user_id=self._user_id,
+                name=secret_name,
+                zone_id=self._zone_id,
+            )
+            if resolved is None:
+                logger.warning(
+                    "Secret %r not found for user=%s zone=%s",
+                    secret_name,
+                    self._user_id,
+                    self._zone_id,
+                )
+                return str(match.group(0))  # leave unresolved
+            return str(resolved)
+
+        return SECRET_PATTERN.sub(_replacer, value)
+
+    def resolve_config(self, config: Any) -> Any:
+        """Recursively resolve nexus-secret:NAME patterns in a config structure.
+
+        Handles dicts, lists, and string values. Non-string leaves are returned as-is.
+        """
+        if isinstance(config, str):
+            if SECRET_PATTERN.search(config):
+                return self.resolve_string(config)
+            return config
+        elif isinstance(config, dict):
+            return {k: self.resolve_config(v) for k, v in config.items()}
+        elif isinstance(config, list):
+            return [self.resolve_config(item) for item in config]
+        return config
+
+    def has_secrets(self, config: Any) -> bool:
+        """Check if a config structure contains any nexus-secret:NAME references."""
+        if isinstance(config, str):
+            return bool(SECRET_PATTERN.search(config))
+        elif isinstance(config, dict):
+            return any(self.has_secrets(v) for v in config.values())
+        elif isinstance(config, list):
+            return any(self.has_secrets(item) for item in config)
+        return False