feat(connections): add provider SDK contract

backend/app/api/connections.py

@@ -12,6 +12,12 @@ from flask import Blueprint, jsonify, request
 
 from app.middleware.rbac import admin_required, auth_required, get_current_user
 from app.services.connection_registry import ConnectionRegistry
+from app.services.connection_provider_sdk import (
+    ConnectionProviderRegistry,
+    ConnectionRef,
+)
+# Register built-in provider adapters before serving SDK schema/operations.
+from app.services import connection_providers  # noqa: F401
 from app.services.container_registry_service import ContainerRegistryService
 from app.services.workspace_service import WorkspaceService
 
@@ -41,6 +47,117 @@ def _workspace_id():
         user, request.headers.get('X-Workspace-Id') or request.args.get('workspace_id'))
 
 
+def _connection_ref(kind, connection_id):
+    user = get_current_user()
+    normalized_id = int(connection_id) if str(connection_id).isdigit() else connection_id
+    return ConnectionRef(
+        kind=kind,
+        connection_id=normalized_id,
+        user_id=user.id if user else None,
+        workspace_id=_workspace_id(),
+    )
+
+
+def _provider_response(result):
+    if result.success:
+        return jsonify(result.to_dict())
+    status = {
+        'validation_error': 400,
+        'invalid_reference': 400,
+        'not_found': 404,
+        'unknown_provider': 404,
+        'unsupported_operation': 405,
+        'rate_limited': 429,
+        'connection_test_failed': 502,
+        'rotation_test_failed': 502,
+    }.get(result.error_code, 502)
+    response = jsonify(result.to_dict())
+    if result.retry_after:
+        response.headers['Retry-After'] = str(result.retry_after)
+    return response, status
+
+
+def _audit_provider_operation(kind, connection_id, operation, result):
+    """One secret-free audit shape for provider mutations and live tests."""
+    from app.services.audit_service import AuditService
+
+    user = get_current_user()
+    AuditService.log(
+        action=f'connection.provider.{operation}',
+        user_id=user.id if user else None,
+        target_type=f'connection:{kind}',
+        target_id=int(connection_id) if str(connection_id).isdigit() else None,
+        details={
+            'connection_id': str(connection_id),
+            'operation': operation,
+            'success': result.success,
+            'error_code': result.error_code,
+            'retryable': result.retryable,
+        },
+    )
+
+
+# ── Provider SDK ─────────────────────────────────────────────────────────────
+
+@connections_bp.route('/providers', methods=['GET'])
+@admin_required
+def list_provider_schemas():
+    return jsonify({'providers': ConnectionProviderRegistry.schemas()})
+
+
+@connections_bp.route('/providers/<kind>/validate', methods=['POST'])
+@admin_required
+def validate_provider_credentials(kind):
+    result = ConnectionProviderRegistry.execute(
+        kind, 'validate', payload=request.get_json() or {},
+        partial=request.args.get('partial', '').lower() == 'true')
+    return _provider_response(result)
+
+
+@connections_bp.route('/providers/<kind>/<connection_id>/health', methods=['GET'])
+@admin_required
+def provider_health(kind, connection_id):
+    result = ConnectionProviderRegistry.execute(
+        kind, 'health', ref=_connection_ref(kind, connection_id))
+    return _provider_response(result)
+
+
+@connections_bp.route('/providers/<kind>/<connection_id>/resources', methods=['GET'])
+@admin_required
+def provider_resources(kind, connection_id):
+    result = ConnectionProviderRegistry.execute(
+        kind, 'list_resources', ref=_connection_ref(kind, connection_id))
+    return _provider_response(result)
+
+
+@connections_bp.route('/providers/<kind>/<connection_id>/test', methods=['POST'])
+@admin_required
+def provider_test(kind, connection_id):
+    result = ConnectionProviderRegistry.execute(
+        kind, 'test', ref=_connection_ref(kind, connection_id))
+    _audit_provider_operation(kind, connection_id, 'test', result)
+    return _provider_response(result)
+
+
+@connections_bp.route('/providers/<kind>/<connection_id>/rotate', methods=['POST'])
+@admin_required
+def provider_rotate(kind, connection_id):
+    result = ConnectionProviderRegistry.execute(
+        kind, 'rotate', ref=_connection_ref(kind, connection_id),
+        payload=request.get_json() or {})
+    _audit_provider_operation(kind, connection_id, 'rotate', result)
+    return _provider_response(result)
+
+
+@connections_bp.route('/providers/<kind>/<connection_id>', methods=['DELETE'])
+@admin_required
+def provider_disconnect(kind, connection_id):
+    result = ConnectionProviderRegistry.execute(
+        kind, 'disconnect', ref=_connection_ref(kind, connection_id))
+    _audit_provider_operation(kind, connection_id, 'disconnect', result)
+    return _provider_response(result)
+
+
 @connections_bp.route('/registries', methods=['GET'])
 @auth_required()
 def list_registries():
@@ -92,8 +209,10 @@ def delete_registry(registry_id):
     registry = ContainerRegistryService.get(registry_id)
     if not registry:
         return jsonify({'error': 'Registry not found'}), 404
-    ContainerRegistryService.delete(registry)
-    return jsonify({'success': True})
+    result = ConnectionProviderRegistry.execute(
+        'registry', 'disconnect', ref=_connection_ref('registry', registry_id))
+    _audit_provider_operation('registry', registry_id, 'disconnect', result)
+    return _provider_response(result)
 
 
 @connections_bp.route('/registries/<int:registry_id>/test', methods=['POST'])
@@ -102,7 +221,7 @@ def test_registry(registry_id):
     registry = ContainerRegistryService.get(registry_id)
     if not registry:
         return jsonify({'error': 'Registry not found'}), 404
-    result = ContainerRegistryService.test_connection(registry)
-    if result.get('success'):
-        return jsonify({'success': True, 'message': f'Logged in to {registry.login_host()}'})
-    return jsonify({'success': False, 'error': result.get('error', 'Login failed')}), 400
+    result = ConnectionProviderRegistry.execute(
+        'registry', 'test', ref=_connection_ref('registry', registry_id))
+    _audit_provider_operation('registry', registry_id, 'test', result)
+    return _provider_response(result)

backend/app/models/container_registry.py

@@ -36,6 +36,9 @@ class ContainerRegistry(db.Model):
     workspace_id = db.Column(db.Integer, db.ForeignKey('workspaces.id'), nullable=True, index=True)
     created_by = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True)
     last_used_at = db.Column(db.DateTime, nullable=True)
+    last_tested_at = db.Column(db.DateTime, nullable=True)
+    last_test_ok = db.Column(db.Boolean, nullable=True)
+    last_test_error = db.Column(db.String(500), nullable=True)
     created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
     updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
 
@@ -65,6 +68,8 @@ class ContainerRegistry(db.Model):
             'workspace_id': self.workspace_id,
             'created_by': self.created_by,
             'last_used_at': self.last_used_at.isoformat() if self.last_used_at else None,
+            'last_tested_at': self.last_tested_at.isoformat() if self.last_tested_at else None,
+            'last_test_ok': self.last_test_ok,
             'created_at': self.created_at.isoformat() if self.created_at else None,
             'updated_at': self.updated_at.isoformat() if self.updated_at else None,
         }

backend/app/services/connection_provider_sdk.py

@@ -0,0 +1,225 @@
+"""Contract and dispatcher for external-account providers.
+
+Provider implementations may keep their existing persistence model, but every
+operation crossing the Connections boundary uses the same result envelope.
+This keeps provider-specific HTTP clients and credential shapes out of the API
+layer while giving the UI stable capability, health, retry, and rate-limit
+metadata.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Dict, Iterable, Optional, Type
+
+
+OPERATIONS = (
+    'validate',
+    'test',
+    'list_resources',
+    'health',
+    'rotate',
+    'disconnect',
+)
+
+
+class ProviderOperationError(Exception):
+    """Expected provider failure with transport-neutral recovery metadata."""
+
+    def __init__(self, message, *, code='provider_error', retryable=False,
+                 retry_after=None, rate_limit=None, details=None):
+        super().__init__(message)
+        self.code = code
+        self.retryable = bool(retryable)
+        self.retry_after = retry_after
+        self.rate_limit = rate_limit or {}
+        self.details = details or {}
+
+
+@dataclass(frozen=True)
+class ConnectionRef:
+    kind: str
+    connection_id: Any
+    provider: Optional[str] = None
+    user_id: Optional[int] = None
+    workspace_id: Optional[int] = None
+
+
+@dataclass
+class ProviderResult:
+    operation: str
+    success: bool
+    data: Dict[str, Any] = field(default_factory=dict)
+    error: Optional[str] = None
+    error_code: Optional[str] = None
+    retryable: bool = False
+    retry_after: Optional[int] = None
+    rate_limit: Dict[str, Any] = field(default_factory=dict)
+    details: Dict[str, Any] = field(default_factory=dict)
+    observed_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
+
+    def to_dict(self):
+        payload = asdict(self)
+        if not self.error:
+            payload.pop('error')
+            payload.pop('error_code')
+        if not self.retry_after:
+            payload.pop('retry_after')
+        if not self.rate_limit:
+            payload.pop('rate_limit')
+        if not self.details:
+            payload.pop('details')
+        return payload
+
+
+class ConnectionProvider:
+    """Base provider contract.
+
+    A provider opts into an operation by overriding its method. Unsupported
+    operations remain visible in the schema with ``supported: false`` instead
+    of failing later with an ad-hoc response.
+    """
+
+    kind = None
+    display_name = None
+    credential_schema = {}
+    permission_scopes = ()
+    max_attempts = 1
+    retry_operations = ('test', 'list_resources', 'health')
+
+    @classmethod
+    def capabilities(cls):
+        return {
+            operation: getattr(cls, operation) is not getattr(ConnectionProvider, operation)
+            for operation in OPERATIONS
+        }
+
+    @classmethod
+    def schema(cls):
+        return {
+            'kind': cls.kind,
+            'display_name': cls.display_name or cls.kind,
+            'credentials': cls.credential_schema,
+            'permission_scopes': list(cls.permission_scopes),
+            'capabilities': cls.capabilities(),
+            'retry_policy': {
+                'max_attempts': max(1, int(cls.max_attempts)),
+                'operations': list(cls.retry_operations),
+                'honors_retry_after': True,
+            },
+        }
+
+    def validate(self, payload, *, partial=False):
+        raise NotImplementedError
+
+    def test(self, ref):
+        raise NotImplementedError
+
+    def list_resources(self, ref):
+        raise NotImplementedError
+
+    def health(self, ref):
+        raise NotImplementedError
+
+    def rotate(self, ref, payload):
+        raise NotImplementedError
+
+    def disconnect(self, ref):
+        raise NotImplementedError
+
+
+class ConnectionProviderRegistry:
+    """Registers provider adapters and normalizes every operation result."""
+
+    _providers: Dict[str, Type[ConnectionProvider]] = {}
+
+    @classmethod
+    def register(cls, provider_cls):
+        kind = getattr(provider_cls, 'kind', None)
+        if not isinstance(kind, str) or not kind.strip():
+            raise TypeError('Connection providers require a non-empty kind')
+        if kind in cls._providers and cls._providers[kind] is not provider_cls:
+            raise ValueError(f"Connection provider '{kind}' is already registered")
+        cls._providers[kind] = provider_cls
+        return provider_cls
+
+    @classmethod
+    def get(cls, kind):
+        provider_cls = cls._providers.get(kind)
+        if not provider_cls:
+            raise ProviderOperationError(
+                f"Unknown connection provider kind: {kind}", code='unknown_provider')
+        return provider_cls()
+
+    @classmethod
+    def schemas(cls):
+        return [cls._providers[kind].schema() for kind in sorted(cls._providers)]
+
+    @classmethod
+    def execute(cls, kind, operation, *, ref=None, payload=None, partial=False):
+        if operation not in OPERATIONS:
+            return ProviderResult(
+                operation=operation, success=False,
+                error=f'Unknown provider operation: {operation}',
+                error_code='unknown_operation',
+            )
+        try:
+            provider = cls.get(kind)
+            if not provider.capabilities()[operation]:
+                raise ProviderOperationError(
+                    f"{provider.display_name or kind} does not support {operation}",
+                    code='unsupported_operation')
+
+            attempts = 0
+            while True:
+                attempts += 1
+                try:
+                    if operation == 'validate':
+                        data = provider.validate(payload or {}, partial=partial)
+                    elif operation == 'rotate':
+                        data = provider.rotate(ref, payload or {})
+                    else:
+                        data = getattr(provider, operation)(ref)
+                    break
+                except ProviderOperationError as exc:
+                    can_retry = (
+                        exc.retryable
+                        and exc.retry_after is None
+                        and operation in provider.retry_operations
+                        and attempts < max(1, int(provider.max_attempts))
+                    )
+                    if not can_retry:
+                        exc.details.setdefault('attempts', attempts)
+                        raise
+
+            if isinstance(data, ProviderResult):
+                return data
+            result = ProviderResult(operation=operation, success=True, data=data or {})
+            if attempts > 1:
+                result.details['attempts'] = attempts
+            return result
+        except ProviderOperationError as exc:
+            return ProviderResult(
+                operation=operation, success=False, error=str(exc),
+                error_code=exc.code, retryable=exc.retryable,
+                retry_after=exc.retry_after, rate_limit=exc.rate_limit,
+                details=exc.details,
+            )
+        except NotImplementedError:
+            return ProviderResult(
+                operation=operation, success=False,
+                error=f'{kind} does not support {operation}',
+                error_code='unsupported_operation',
+            )
+        except Exception as exc:
+            return ProviderResult(
+                operation=operation, success=False, error=str(exc),
+                error_code='provider_error',
+            )
+
+
+def register_providers(providers: Iterable[Type[ConnectionProvider]]):
+    """Small helper for extension/provider packages registering in one call."""
+    for provider in providers:
+        ConnectionProviderRegistry.register(provider)

backend/app/services/connection_providers/__init__.py

@@ -0,0 +1,6 @@
+"""Built-in adapters for the connection provider SDK."""
+
+from app.services.connection_providers.container_registry import (  # noqa: F401
+    ContainerRegistryProvider,
+)
+

backend/app/services/connection_providers/container_registry.py

@@ -0,0 +1,103 @@
+"""Connection-provider adapter for private container registries."""
+
+from datetime import datetime
+
+from app.models.container_registry import ContainerRegistry
+from app.services.connection_provider_sdk import (
+    ConnectionProvider,
+    ConnectionProviderRegistry,
+    ProviderOperationError,
+)
+from app.services.container_registry_service import ContainerRegistryService
+
+
+@ConnectionProviderRegistry.register
+class ContainerRegistryProvider(ConnectionProvider):
+    kind = 'registry'
+    display_name = 'Container registry'
+    permission_scopes = ('image:pull',)
+    credential_schema = {
+        'type': 'object',
+        'required': ['secret'],
+        'properties': {
+            'secret': {'type': 'string', 'secret': True, 'minLength': 1},
+        },
+        'additionalProperties': False,
+    }
+
+    @staticmethod
+    def _get(ref):
+        if ref is None:
+            raise ProviderOperationError('Connection reference is required', code='invalid_reference')
+        registry = ContainerRegistryService.get(ref.connection_id)
+        if not registry:
+            raise ProviderOperationError('Registry not found', code='not_found')
+        if ref.workspace_id is not None and registry.workspace_id not in (None, ref.workspace_id):
+            raise ProviderOperationError('Registry not found', code='not_found')
+        return registry
+
+    def validate(self, payload, *, partial=False):
+        errors = {}
+        secret = payload.get('secret')
+        if not partial and not isinstance(secret, str):
+            errors['secret'] = 'secret is required'
+        elif secret is not None and (not isinstance(secret, str) or not secret.strip()):
+            errors['secret'] = 'secret must be a non-empty string'
+        extra = sorted(set(payload) - {'secret'})
+        if extra:
+            errors['_schema'] = f"unknown credential fields: {', '.join(extra)}"
+        if errors:
+            raise ProviderOperationError(
+                'Credential validation failed', code='validation_error',
+                details={'field_errors': errors})
+        return {'valid': True}
+
+    def test(self, ref):
+        registry = self._get(ref)
+        result = ContainerRegistryService.test_connection(registry)
+        if not result.get('success'):
+            raise ProviderOperationError(
+                result.get('error') or 'Registry login failed',
+                code=result.get('error_code') or 'connection_test_failed',
+                retryable=bool(result.get('retryable')),
+                retry_after=result.get('retry_after'),
+                rate_limit=result.get('rate_limit'),
+            )
+        return {'message': f'Logged in to {registry.login_host()}', 'health': 'healthy'}
+
+    def health(self, ref):
+        registry = self._get(ref)
+        if registry.last_test_ok is True:
+            status = 'healthy'
+        elif registry.last_test_ok is False:
+            status = 'unhealthy'
+        else:
+            status = 'unknown'
+        return {
+            'status': status,
+            'checked_at': registry.last_tested_at.isoformat() if registry.last_tested_at else None,
+            'message': registry.last_test_error,
+            'configured': bool(registry.secret_encrypted and registry.login_username()),
+        }
+
+    def rotate(self, ref, payload):
+        self.validate(payload)
+        registry = self._get(ref)
+        result = ContainerRegistryService.rotate_secret(registry, payload['secret'])
+        if not result.get('success'):
+            raise ProviderOperationError(
+                result.get('error') or 'New credential failed validation',
+                code=result.get('error_code') or 'rotation_test_failed',
+                retryable=bool(result.get('retryable')),
+                retry_after=result.get('retry_after'),
+                rate_limit=result.get('rate_limit'),
+            )
+        return {
+            'rotated_at': datetime.utcnow().isoformat(),
+            'health': 'healthy',
+        }
+
+    def disconnect(self, ref):
+        registry = self._get(ref)
+        ContainerRegistryService.delete(registry)
+        return {'disconnected': True, 'connection_id': ref.connection_id}

backend/app/services/connection_registry.py

@@ -1,6 +1,6 @@
 """Unified read facade over every external-account store ServerKit holds.
 
-ServerKit's connections live in five places, each with its own model/service and
+ServerKit's connections live in several places, each with its own model/service and
 write path (kept that way deliberately — they have genuinely different auth and
 cardinality):
 
@@ -10,11 +10,11 @@ cardinality):
   - registrars            (RegistrarConnection)   — GoDaddy / Namecheap
   - object storage        (storage.json)          — S3-compatible / Backblaze B2
 
-This registry doesn't replace any of them; it presents them as ONE normalized,
+This registry presents them as ONE normalized,
 secret-free list (`GET /api/v1/connections`) so any surface can answer "what
 external accounts are connected, and are their secrets encrypted?" in one call.
-All five now encrypt secrets identically via app.utils.crypto, so `encrypted`
-should be True for every real connection.
+New write paths migrate behind ``connection_provider_sdk`` adapters; legacy
+stores stay readable here and report ``sdk_managed: false`` until migrated.
 """
 
 import logging
@@ -22,8 +22,9 @@ import logging
 logger = logging.getLogger(__name__)
 
 
-def _entry(kind, provider, label, *, id=None, scope=None, encrypted=True, created_at=None):
-    return {
+def _entry(kind, provider, label, *, id=None, scope=None, encrypted=True,
+           created_at=None, health=None, capabilities=None, sdk_managed=False):
+    entry = {
         'kind': kind,
         'provider': provider,
         'id': id,
@@ -31,7 +32,13 @@ def _entry(kind, provider, label, *, id=None, scope=None, encrypted=True, create
         'scope': scope,
         'encrypted': bool(encrypted),
         'created_at': created_at.isoformat() if hasattr(created_at, 'isoformat') else created_at,
+        'sdk_managed': bool(sdk_managed),
     }
+    if health is not None:
+        entry['health'] = health
+    if capabilities is not None:
+        entry['capabilities'] = capabilities
+    return entry
 
 
 class ConnectionRegistry:
@@ -118,9 +125,21 @@ class ConnectionRegistry:
     @staticmethod
     def _registries():
         from app.models.container_registry import ContainerRegistry
+        # Importing the built-in adapter registers it with the SDK.
+        from app.services.connection_providers import ContainerRegistryProvider
+
+        capabilities = ContainerRegistryProvider.capabilities()
         return [
             _entry('registry', c.provider, c.name, id=c.id, scope=c.login_host(),
-                   encrypted=bool(c.secret_encrypted), created_at=c.created_at)
+                   encrypted=bool(c.secret_encrypted), created_at=c.created_at,
+                   sdk_managed=True, capabilities=capabilities,
+                   health={
+                       'status': ('healthy' if c.last_test_ok is True else
+                                  'unhealthy' if c.last_test_ok is False else 'unknown'),
+                       'checked_at': (c.last_tested_at.isoformat()
+                                      if c.last_tested_at else None),
+                       'message': c.last_test_error,
+                   })
             for c in ContainerRegistry.query.all()
         ]
 

backend/app/services/container_registry_service.py

@@ -86,27 +86,30 @@ class ContainerRegistryService:
 
     # ── secret resolution ──
     @staticmethod
-    def _password(registry):
+    def _password(registry, secret_override=None):
         """Plaintext password for ``docker login``. For ECR, exchange the stored
         AWS keys for a short-lived token lazily; otherwise decrypt the stored
         secret."""
         if registry.provider == 'ecr':
-            token = ContainerRegistryService.ecr_password(registry)
+            token = ContainerRegistryService.ecr_password(
+                registry, secret_override=secret_override)
             if token:
                 return token
+        if secret_override is not None:
+            return secret_override
         if not registry.secret_encrypted:
             return None
         return registry.secret
 
     # ── docker login / logout ──
     @staticmethod
-    def login(registry):
+    def login(registry, *, secret_override=None, record_use=True):
         """``docker login`` the registry, piping the secret via stdin. Returns
         ``{'success': bool, 'error'?: str}``. Never logs the secret."""
         username = registry.login_username()
         if not username:
             return {'success': False, 'error': 'Registry has no username configured'}
-        password = ContainerRegistryService._password(registry)
+        password = ContainerRegistryService._password(registry, secret_override)
         if not password:
             return {'success': False, 'error': 'No stored secret for this registry'}
 
@@ -116,11 +119,12 @@ class ContainerRegistryService:
         result = run_checked(cmd, input=password, timeout=None)
 
         if result['success']:
-            try:
-                registry.last_used_at = datetime.utcnow()
-                db.session.commit()
-            except Exception:
-                db.session.rollback()
+            if record_use:
+                try:
+                    registry.last_used_at = datetime.utcnow()
+                    db.session.commit()
+                except Exception:
+                    db.session.rollback()
             return {'success': True}
         return {'success': False, 'error': result['error'] or 'docker login failed'}
 
@@ -138,12 +142,40 @@ class ContainerRegistryService:
     def test_connection(registry):
         """``docker login`` then immediately ``docker logout`` — report
         success/failure without leaving a session behind."""
-        result = ContainerRegistryService.login(registry)
+        result = ContainerRegistryService.login(registry, record_use=False)
         ContainerRegistryService.logout(registry.login_host())
+        ContainerRegistryService._record_test(registry, result)
         return result
 
     @staticmethod
-    def ecr_password(registry):
+    def _record_test(registry, result):
+        """Persist secret-free health evidence for all connection surfaces."""
+        registry.last_tested_at = datetime.utcnow()
+        registry.last_test_ok = bool(result.get('success'))
+        registry.last_test_error = None if result.get('success') else str(
+            result.get('error') or 'Connection test failed')[:500]
+        db.session.commit()
+
+    @staticmethod
+    def rotate_secret(registry, new_secret):
+        """Test a replacement secret before storing it.
+
+        The old ciphertext stays untouched when validation fails, so a typo
+        cannot destroy the last known credential.
+        """
+        result = ContainerRegistryService.login(
+            registry, secret_override=new_secret, record_use=False)
+        ContainerRegistryService.logout(registry.login_host())
+        ContainerRegistryService._record_test(registry, result)
+        if not result.get('success'):
+            return result
+        registry.secret = new_secret
+        registry.last_test_error = None
+        db.session.commit()
+        return {'success': True}
+
+    @staticmethod
+    def ecr_password(registry, secret_override=None):
         """Best-effort AWS ECR token via ``aws ecr get-login-password``.
 
         Optional/lazy: needs the AWS CLI on PATH. The stored secret may hold
@@ -161,7 +193,7 @@ class ContainerRegistryService:
                     region = parts[idx + 1]
 
             env = None
-            secret = registry.secret
+            secret = secret_override if secret_override is not None else registry.secret
             if secret and ':' in secret:
                 access_key, secret_key = secret.split(':', 1)
                 env = dict(os.environ)

backend/migrations/versions/094_connection_provider_health.py

@@ -0,0 +1,48 @@
+"""Add normalized health observations to container registry connections.
+
+Revision ID: 094_connection_provider_health
+Revises: 093_container_scale_policies
+Create Date: 2026-08-28
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = '094_connection_provider_health'
+down_revision = '093_container_scale_policies'
+branch_labels = None
+depends_on = None
+
+
+def _columns(table):
+    inspector = sa.inspect(op.get_bind())
+    if table not in inspector.get_table_names():
+        return set()
+    return {column['name'] for column in inspector.get_columns(table)}
+
+
+def upgrade():
+    columns = _columns('container_registries')
+    if not columns:
+        return
+    with op.batch_alter_table('container_registries') as batch:
+        if 'last_tested_at' not in columns:
+            batch.add_column(sa.Column('last_tested_at', sa.DateTime(), nullable=True))
+        if 'last_test_ok' not in columns:
+            batch.add_column(sa.Column('last_test_ok', sa.Boolean(), nullable=True))
+        if 'last_test_error' not in columns:
+            batch.add_column(sa.Column('last_test_error', sa.String(length=500), nullable=True))
+
+
+def downgrade():
+    columns = _columns('container_registries')
+    if not columns:
+        return
+    with op.batch_alter_table('container_registries') as batch:
+        if 'last_test_error' in columns:
+            batch.drop_column('last_test_error')
+        if 'last_test_ok' in columns:
+            batch.drop_column('last_test_ok')
+        if 'last_tested_at' in columns:
+            batch.drop_column('last_tested_at')
+

backend/tests/test_connection_provider_sdk.py

@@ -0,0 +1,86 @@
+"""Provider SDK contract tests independent of any credential model."""
+
+import pytest
+
+from app.services.connection_provider_sdk import (
+    ConnectionProvider,
+    ConnectionProviderRegistry,
+    ConnectionRef,
+    ProviderOperationError,
+)
+
+
+@pytest.fixture
+def isolated_registry(monkeypatch):
+    monkeypatch.setattr(ConnectionProviderRegistry, '_providers', {})
+    return ConnectionProviderRegistry
+
+
+def test_registration_requires_a_kind(isolated_registry):
+    class MissingKind(ConnectionProvider):
+        pass
+
+    with pytest.raises(TypeError):
+        isolated_registry.register(MissingKind)
+
+
+def test_capabilities_and_schema_are_derived_from_overrides(isolated_registry):
+    @isolated_registry.register
+    class Minimal(ConnectionProvider):
+        kind = 'minimal'
+        display_name = 'Minimal provider'
+        credential_schema = {'type': 'object'}
+
+        def health(self, ref):
+            return {'status': 'healthy'}
+
+    schema = isolated_registry.schemas()[0]
+    assert schema['capabilities']['health'] is True
+    assert schema['capabilities']['rotate'] is False
+    assert schema['retry_policy']['honors_retry_after'] is True
+
+    unsupported = isolated_registry.execute(
+        'minimal', 'rotate', ref=ConnectionRef('minimal', 'one'), payload={})
+    assert unsupported.success is False
+    assert unsupported.error_code == 'unsupported_operation'
+
+
+def test_retryable_reads_use_bounded_provider_policy(isolated_registry):
+    @isolated_registry.register
+    class Flaky(ConnectionProvider):
+        kind = 'flaky'
+        max_attempts = 3
+
+        def __init__(self):
+            self.calls = 0
+
+        def health(self, ref):
+            self.calls += 1
+            if self.calls < 3:
+                raise ProviderOperationError('temporary', retryable=True)
+            return {'status': 'healthy'}
+
+    result = isolated_registry.execute(
+        'flaky', 'health', ref=ConnectionRef('flaky', 'one'))
+    assert result.success is True
+    assert result.details['attempts'] == 3
+
+
+def test_retry_after_is_returned_without_blocking(isolated_registry):
+    @isolated_registry.register
+    class Limited(ConnectionProvider):
+        kind = 'limited'
+        max_attempts = 3
+
+        def test(self, ref):
+            raise ProviderOperationError(
+                'rate limited', code='rate_limited', retryable=True,
+                retry_after=60, rate_limit={'remaining': 0})
+
+    result = isolated_registry.execute(
+        'limited', 'test', ref=ConnectionRef('limited', 'one'))
+    assert result.success is False
+    assert result.error_code == 'rate_limited'
+    assert result.retry_after == 60
+    assert result.rate_limit == {'remaining': 0}
+    assert result.details['attempts'] == 1

backend/tests/test_container_registry.py

@@ -131,6 +131,29 @@ def test_test_connection_logs_in_and_out(app, registry, fake_subprocess):
     result = ContainerRegistryService.test_connection(registry)
     assert result['success'] is True
     assert _docker_subcommands(fake_subprocess) == ['login', 'logout']
+    assert registry.last_test_ok is True
+    assert registry.last_tested_at is not None
+
+
+def test_rotate_secret_tests_before_persisting(app, registry, fake_subprocess):
+    fake_subprocess.script(['docker'], stdout='ok')
+
+    result = ContainerRegistryService.rotate_secret(registry, 'replacement-token')
+
+    assert result['success'] is True
+    assert fake_subprocess.kwargs_for(['docker', 'login'])['input'] == 'replacement-token'
+    assert ContainerRegistryService._password(registry) == 'replacement-token'
+
+
+def test_failed_rotation_preserves_previous_secret(app, registry, fake_subprocess):
+    fake_subprocess.script(['docker', 'login'], returncode=1, stderr='denied')
+    fake_subprocess.script(['docker', 'logout'])
+
+    result = ContainerRegistryService.rotate_secret(registry, 'bad-token')
+
+    assert result['success'] is False
+    assert ContainerRegistryService._password(registry) == 'ghp_supersecrettoken'
+    assert registry.last_test_ok is False
 
 
 # ── proving test: deploy path authenticates before pull ──────────────────────
@@ -270,4 +293,50 @@ def test_registry_appears_in_unified_connections(client, auth_headers, app):
     reg_entries = [c for c in conns if c['kind'] == 'registry']
     assert len(reg_entries) == 1
     assert reg_entries[0]['encrypted'] is True
+    assert reg_entries[0]['sdk_managed'] is True
+    assert reg_entries[0]['capabilities']['rotate'] is True
+    assert reg_entries[0]['capabilities']['list_resources'] is False
+    assert reg_entries[0]['health']['status'] == 'unknown'
     assert 'secret' not in reg_entries[0]
+
+
+def test_provider_sdk_schema_and_operations(
+        client, auth_headers, app, registry, fake_subprocess):
+    schema = client.get('/api/v1/connections/providers', headers=auth_headers)
+    assert schema.status_code == 200
+    provider = next(p for p in schema.get_json()['providers'] if p['kind'] == 'registry')
+    assert provider['credentials']['properties']['secret']['secret'] is True
+    assert provider['capabilities'] == {
+        'validate': True,
+        'test': True,
+        'list_resources': False,
+        'health': True,
+        'rotate': True,
+        'disconnect': True,
+    }
+
+    invalid = client.post(
+        '/api/v1/connections/providers/registry/validate',
+        headers=auth_headers, json={})
+    assert invalid.status_code == 400
+    assert invalid.get_json()['error_code'] == 'validation_error'
+    assert invalid.get_json()['details']['field_errors']['secret'] == 'secret is required'
+
+    fake_subprocess.script(['docker'], stdout='ok')
+    tested = client.post(
+        f'/api/v1/connections/providers/registry/{registry.id}/test',
+        headers=auth_headers)
+    assert tested.status_code == 200, tested.get_json()
+    assert tested.get_json()['data']['health'] == 'healthy'
+
+    health = client.get(
+        f'/api/v1/connections/providers/registry/{registry.id}/health',
+        headers=auth_headers)
+    assert health.status_code == 200
+    assert health.get_json()['data']['status'] == 'healthy'
+
+    resources = client.get(
+        f'/api/v1/connections/providers/registry/{registry.id}/resources',
+        headers=auth_headers)
+    assert resources.status_code == 405
+    assert resources.get_json()['error_code'] == 'unsupported_operation'

docs/CONNECTION_PROVIDER_SDK.md

@@ -0,0 +1,48 @@
+# Connection provider SDK
+
+ServerKit's Connections page reads credentials from several persistence models.
+Those models do not need to be collapsed into one table, but their behavior must
+cross one service contract before new providers are added.
+
+The contract lives in `backend/app/services/connection_provider_sdk.py` and has
+six named operations:
+
+- `validate(payload, partial=False)` validates credential shape without storing it.
+- `test(ref)` performs a live, read-only authentication check.
+- `list_resources(ref)` returns provider-owned resources when the API supports it.
+- `health(ref)` returns the last normalized health observation.
+- `rotate(ref, payload)` tests replacement credentials before committing them.
+- `disconnect(ref)` removes the connection and its stored secret material.
+
+Every operation returns a `ProviderResult`. Failures use
+`ProviderOperationError` so HTTP status, retryability, `Retry-After`, rate-limit
+state, and field errors do not depend on the provider's native response shape.
+Safe read operations can opt into bounded retries with `max_attempts`; a response
+carrying `retry_after` is returned immediately rather than sleeping in a request.
+
+## Adding an adapter
+
+1. Subclass `ConnectionProvider` under `app/services/connection_providers/`.
+2. Set a stable `kind`, JSON-style `credential_schema`, and least-privilege
+   `permission_scopes`.
+3. Override only operations the provider can actually support. Capability
+   discovery is derived from these overrides; unsupported operations remain
+   explicit in the schema.
+4. Decorate the class with `@ConnectionProviderRegistry.register` and import it
+   from `connection_providers/__init__.py`.
+5. Keep raw credentials inside the adapter/service boundary. Exceptions, audit
+   details, health messages, and `ProviderResult.data` must be secret-free.
+6. Add contract tests for validation, failed rotation rollback, health, rate
+   limits, permissions, and disconnect.
+
+The admin API is rooted at `/api/v1/connections/providers`. Mutating operations
+and live tests produce `connection.provider.<operation>` audit rows with one
+secret-free details shape. Existing provider-specific routes can remain during
+migration, but should delegate to the SDK as the container-registry routes do.
+
+## Migration rule
+
+`GET /api/v1/connections` marks entries with `sdk_managed`. A provider family is
+not fully migrated until this flag is true and its advertised capability map is
+backed by tests. Do not claim `list_resources` or rotation support by returning
+an empty success response; leave the capability false until it is real.