fix(watchsync): harden connection form contracts

cmd/silo/main.go

@@ -3163,11 +3163,12 @@ func reloadWatchSyncPluginProviders(
 				continue
 			}
 			provider, err := watchsync.NewPluginProvider(watchsync.PluginProviderOptions{
-				InstallationID: installation.ID,
+				InstallationID:         installation.ID,
-				ProviderKey:    fmt.Sprintf("plugin:%d:%s", installation.ID, capability.ID),
+				ProviderKey:            fmt.Sprintf("plugin:%d:%s", installation.ID, capability.ID),
-				CapabilityID:   capability.ID,
+				CapabilityID:           capability.ID,
-				DisplayName:    descriptor.GetDisplayName(),
+				DisplayName:            descriptor.GetDisplayName(),
-				Descriptor:     descriptor.GetWatchSyncProvider(),
+				Descriptor:             descriptor.GetWatchSyncProvider(),
+				ConnectionConfigSchema: descriptor.GetConfigSchema(),
 				ResolveClient: func(callCtx context.Context, installationID int, capabilityID string) (watchsync.WatchSyncPluginClient, error) {
 					return service.WatchSyncProviderClient(callCtx, installationID, capabilityID)
 				},

cmd/silo/main_test.go

@@ -198,6 +198,66 @@ func (failingWatchSyncCapabilityStore) ListCapabilities(context.Context, int) ([
 	return nil, errors.New("database unavailable")
 }
 
+type staticWatchSyncCapabilityStore struct {
+	capabilities []*plugins.Capability
+}
+
+func (s staticWatchSyncCapabilityStore) ListEnabled(context.Context) ([]*plugins.Installation, error) {
+	return []*plugins.Installation{{ID: 4, Enabled: true, Kind: plugins.KindPlugin}}, nil
+}
+
+func (s staticWatchSyncCapabilityStore) ListCapabilities(context.Context, int) ([]*plugins.Capability, error) {
+	return s.capabilities, nil
+}
+
+func TestReloadWatchSyncPluginProvidersPreservesConnectionForm(t *testing.T) {
+	manifest := &pluginv1.PluginManifest{Capabilities: []*pluginv1.CapabilityDescriptor{{
+		Type: "watch_sync_provider.v1", Id: "floppy", DisplayName: "Floppy",
+		ConfigSchema: []*pluginv1.ConfigSchema{{
+			Key: "floppy", Title: "Your Floppy server", Required: true,
+			JsonSchema: `{"type":"object","properties":{"base_url":{"type":"string"}},"required":["base_url"]}`,
+			AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+				Key: "base_url", Label: "Server URL", Required: true,
+				Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT,
+			}}},
+		}},
+		WatchSyncProvider: &pluginv1.WatchSyncProviderDescriptor{
+			AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY},
+		},
+	}}}
+	records, err := plugins.CapabilityRecordsFromManifest(manifest)
+	if err != nil {
+		t.Fatal(err)
+	}
+	capabilities := make([]*plugins.Capability, 0, len(records))
+	for i := range records {
+		record := records[i]
+		capabilities = append(capabilities, &record)
+	}
+
+	registry := watchsync.NewRegistry()
+	if err := reloadWatchSyncPluginProviders(
+		context.Background(), registry, staticWatchSyncCapabilityStore{capabilities: capabilities}, &plugins.Service{}, nil,
+	); err != nil {
+		t.Fatal(err)
+	}
+	provider, ok := registry.Get("plugin:4:floppy")
+	if !ok {
+		t.Fatal("Floppy provider was not registered")
+	}
+	configurable, ok := provider.(interface {
+		ConnectionConfigSchema() []plugins.ConfigSchemaView
+	})
+	if !ok {
+		t.Fatal("Floppy provider does not expose connection configuration")
+	}
+	schemas := configurable.ConnectionConfigSchema()
+	if len(schemas) != 1 || schemas[0].AdminForm == nil || len(schemas[0].AdminForm.Fields) != 1 ||
+		schemas[0].AdminForm.Fields[0].Control != "TEXT" {
+		t.Fatalf("connection config schema = %#v", schemas)
+	}
+}
+
 func TestReloadWatchSyncPluginProvidersDropsStaleProvidersOnCapabilityReadFailure(t *testing.T) {
 	registry := watchsync.NewRegistry()
 	provider, err := watchsync.NewPluginProvider(watchsync.PluginProviderOptions{

docs/superpowers/specs/2026-07-25-plugin-watch-sync-provider-design.md

@@ -25,6 +25,8 @@ rate-limit handling, list reconciliation, and scheduled reconciliation.
   expiry, and opaque secret attributes.
 - Supply declared installation configuration to provider calls without mixing
   it with profile credentials.
+- Allow API-key providers to declare setup fields that are supplied per profile
+  connection rather than forcing one installation-wide endpoint.
 - Keep provider plugins stateless and prevent credentials from entering generic
   plugin configuration.
 - Avoid behavior changes for built-in Trakt, Simkl, and MDBList providers.
@@ -117,9 +119,14 @@ The plugin never needs a Silo API key or a callback into catalog HTTP routes.
 ## Authentication and credentials
 
 The existing watch-provider API-key connection route passes the entered token
-to `ExchangeAPIKey`. Device-code providers use the existing start and poll
+to `ExchangeAPIKey`. A plugin capability may also declare `config_schema`
-routes. The plugin validates or exchanges the credential and returns normalized
+entries; the host renders those fields beside the API key, validates submitted
-credentials plus provider account identity.
+objects against the declared JSON schemas, classifies secret fields, and
+overlays the flattened values onto `provider_config` for that exchange only.
+The plugin validates or exchanges the credential and returns normalized
+credentials plus provider account identity. Provider-specific connection
+context needed by later calls is returned in encrypted opaque credential
+attributes. Device-code providers use the existing start and poll routes.
 
 Credential refresh remains host-initiated and serialized by the existing
 watchsync service. The host encrypts an authoritative credential bundle
@@ -202,6 +209,15 @@ are sent in `secret_values`. Nested manifest fields use the stable
 `<config-key>.<field>` key form, such as `floppy.base_url`. Profile credentials
 remain separate and encrypted in watch-provider connection storage.
 
+Capability-level `config_schema` entries are connection setup fields, not
+installation defaults. For API-key authorization, the host validates and
+flattens them using the same `<config-key>.<field>` convention and lets the
+connection values override a same-named global value. This supports migrations
+from a legacy installation-wide endpoint while ensuring every new connection
+chooses its own endpoint. The submitted objects are not retained as generic
+plugin configuration; the plugin returns any context required after exchange
+inside its opaque encrypted credentials.
+
 ## Apply result handling
 
 Each event has a stable ID. The adapter requires a matching result and maps it

go.mod

@@ -109,7 +109,7 @@ require (
 )
 
 require (
-	github.com/Silo-Server/silo-plugin-sdk v0.13.0
+	github.com/Silo-Server/silo-plugin-sdk v0.13.2
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.30 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.30 // indirect

go.sum

@@ -4,8 +4,10 @@ github.com/PuerkitoBio/goquery v1.8.0 h1:PJTF7AmFCFKk1N6V6jmKfrNH9tV5pNE6lZMkG0g
 github.com/PuerkitoBio/goquery v1.8.0/go.mod h1:ypIiRMtY7COPGk+I/YbZLbxsxn9g5ejnI2HSMtkjZvI=
 github.com/SherClockHolmes/webpush-go v1.4.0 h1:ocnzNKWN23T9nvHi6IfyrQjkIc0oJWv1B1pULsf9i3s=
 github.com/SherClockHolmes/webpush-go v1.4.0/go.mod h1:XSq8pKX11vNV8MJEMwjrlTkxhAj1zKfxmyhdV7Pd6UA=
-github.com/Silo-Server/silo-plugin-sdk v0.13.0 h1:BXf4cuNbOIIsrVJrRdlQnbc1m04X6lcPSZI+AFLMeK0=
+github.com/Silo-Server/silo-plugin-sdk v0.13.1 h1:3vMaV+aPT/vu47CPk/f3nODsjXCp1OseI0QIoLnOvOM=
-github.com/Silo-Server/silo-plugin-sdk v0.13.0/go.mod h1:etqmxLTwjxpFH9goAjBDfNDoqHMv2/sqUXu8yx3hNfA=
+github.com/Silo-Server/silo-plugin-sdk v0.13.1/go.mod h1:etqmxLTwjxpFH9goAjBDfNDoqHMv2/sqUXu8yx3hNfA=
+github.com/Silo-Server/silo-plugin-sdk v0.13.2 h1:w7U0mmljVPauKfzRLNKusuiYFpuoEhuCSX4Hp3s9eRw=
+github.com/Silo-Server/silo-plugin-sdk v0.13.2/go.mod h1:etqmxLTwjxpFH9goAjBDfNDoqHMv2/sqUXu8yx3hNfA=
 github.com/abadojack/whatlanggo v1.0.1 h1:19N6YogDnf71CTHm3Mp2qhYfkRdyvbgwWdd2EPxJRG4=
 github.com/abadojack/whatlanggo v1.0.1/go.mod h1:66WiQbSbJBIlOZMsvbKe5m6pzQovxCH9B/K8tQB2uoc=
 github.com/agnivade/levenshtein v1.2.1 h1:EHBY3UOn1gwdy/VbFwgo4cxecRznFk7fKWN1KOX7eoM=

internal/api/handlers/plugins.go

@@ -201,69 +201,10 @@ type pluginPresentationJSON struct {
 	LicenseSPDX         string `json:"license_spdx"`
 }
 
-type pluginConfigSchemaJSON struct {
+type pluginConfigSchemaJSON = plugins.ConfigSchemaView
-	Key         string               `json:"key"`
+type pluginAdminFormJSON = plugins.AdminFormView
-	Title       string               `json:"title"`
+type pluginAdminFormFieldJSON = plugins.AdminFormFieldView
-	Description string               `json:"description"`
+type pluginAdminFormSectionJSON = plugins.AdminFormSectionView
-	JSONSchema  string               `json:"json_schema"`
-	Required    bool                 `json:"required"`
-	AdminForm   *pluginAdminFormJSON `json:"admin_form,omitempty"`
-}
-
-type pluginAdminFormJSON struct {
-	Fields      []pluginAdminFormFieldJSON   `json:"fields"`
-	SubmitLabel string                       `json:"submit_label,omitempty"`
-	Sections    []pluginAdminFormSectionJSON `json:"sections,omitempty"`
-}
-
-type pluginAdminFormFieldJSON struct {
-	Key                 string                         `json:"key"`
-	Label               string                         `json:"label"`
-	Description         string                         `json:"description,omitempty"`
-	Control             string                         `json:"control"`
-	Placeholder         string                         `json:"placeholder,omitempty"`
-	Required            bool                           `json:"required"`
-	Secret              bool                           `json:"secret"`
-	Multiline           bool                           `json:"multiline"`
-	DefaultValue        any                            `json:"default_value,omitempty"`
-	Options             []pluginAdminFormOptionJSON    `json:"options,omitempty"`
-	Rows                int32                          `json:"rows,omitempty"`
-	DynamicOptions      bool                           `json:"dynamic_options,omitempty"`
-	ShowWhen            []pluginAdminFormConditionJSON `json:"show_when,omitempty"`
-	Validation          *pluginAdminFormValidationJSON `json:"validation,omitempty"`
-	ExclusiveGroupField string                         `json:"exclusive_group_field,omitempty"`
-}
-
-type pluginAdminFormOptionJSON struct {
-	Value       string `json:"value"`
-	Label       string `json:"label"`
-	Description string `json:"description,omitempty"`
-}
-
-type pluginAdminFormConditionJSON struct {
-	Field  string   `json:"field"`
-	Equals []string `json:"equals"`
-}
-
-type pluginAdminFormValidationJSON struct {
-	HasMin    bool    `json:"has_min,omitempty"`
-	Min       float64 `json:"min,omitempty"`
-	HasMax    bool    `json:"has_max,omitempty"`
-	Max       float64 `json:"max,omitempty"`
-	Pattern   string  `json:"pattern,omitempty"`
-	MinLength int32   `json:"min_length,omitempty"`
-	MaxLength int32   `json:"max_length,omitempty"`
-}
-
-type pluginAdminFormSectionJSON struct {
-	Key              string                         `json:"key"`
-	Title            string                         `json:"title"`
-	Description      string                         `json:"description,omitempty"`
-	Collapsible      bool                           `json:"collapsible"`
-	CollapsedDefault bool                           `json:"collapsed_default"`
-	FieldKeys        []string                       `json:"field_keys"`
-	ShowWhen         []pluginAdminFormConditionJSON `json:"show_when,omitempty"`
-}
 
 type pluginCapabilityJSON struct {
 	Type          string                   `json:"type"`
@@ -1567,117 +1508,11 @@ func toUserPluginSettingsSummary(
 }
 
 func configSchemasToJSON(schemas []*pluginv1.ConfigSchema) []pluginConfigSchemaJSON {
-	response := make([]pluginConfigSchemaJSON, 0, len(schemas))
+	return plugins.ConfigSchemaViews(schemas)
-	for _, schema := range schemas {
-		if schema == nil {
-			continue
-		}
-		response = append(response, pluginConfigSchemaJSON{
-			Key:         schema.GetKey(),
-			Title:       schema.GetTitle(),
-			Description: schema.GetDescription(),
-			JSONSchema:  schema.GetJsonSchema(),
-			Required:    schema.GetRequired(),
-			AdminForm:   adminFormToJSON(schema.GetAdminForm()),
-		})
-	}
-	return response
 }
 
 func adminFormToJSON(form *pluginv1.AdminFormDescriptor) *pluginAdminFormJSON {
-	if form == nil {
+	return plugins.AdminFormViewFromProto(form)
-		return nil
-	}
-	fields := make([]pluginAdminFormFieldJSON, 0, len(form.GetFields()))
-	for _, field := range form.GetFields() {
-		if field == nil {
-			continue
-		}
-		options := make([]pluginAdminFormOptionJSON, 0, len(field.GetOptions()))
-		for _, option := range field.GetOptions() {
-			if option == nil {
-				continue
-			}
-			options = append(options, pluginAdminFormOptionJSON{
-				Value:       option.GetValue(),
-				Label:       option.GetLabel(),
-				Description: option.GetDescription(),
-			})
-		}
-		var defaultValue any
-		if field.GetDefaultValue() != nil {
-			defaultValue = field.GetDefaultValue().AsInterface()
-		}
-		var validation *pluginAdminFormValidationJSON
-		if v := field.GetValidation(); v != nil {
-			validation = &pluginAdminFormValidationJSON{
-				HasMin:    v.GetHasMin(),
-				Min:       v.GetMin(),
-				HasMax:    v.GetHasMax(),
-				Max:       v.GetMax(),
-				Pattern:   v.GetPattern(),
-				MinLength: v.GetMinLength(),
-				MaxLength: v.GetMaxLength(),
-			}
-		}
-		fields = append(fields, pluginAdminFormFieldJSON{
-			Key:                 field.GetKey(),
-			Label:               field.GetLabel(),
-			Description:         field.GetDescription(),
-			Control:             strings.TrimPrefix(field.GetControl().String(), "ADMIN_FORM_CONTROL_"),
-			Placeholder:         field.GetPlaceholder(),
-			Required:            field.GetRequired(),
-			Secret:              field.GetSecret(),
-			Multiline:           field.GetMultiline(),
-			DefaultValue:        defaultValue,
-			Options:             options,
-			Rows:                field.GetRows(),
-			DynamicOptions:      field.GetDynamicOptions(),
-			ShowWhen:            adminFormConditionsToJSON(field.GetShowWhen()),
-			Validation:          validation,
-			ExclusiveGroupField: field.GetExclusiveGroupField(),
-		})
-	}
-	sections := make([]pluginAdminFormSectionJSON, 0, len(form.GetSections()))
-	for _, section := range form.GetSections() {
-		if section == nil {
-			continue
-		}
-		sections = append(sections, pluginAdminFormSectionJSON{
-			Key:              section.GetKey(),
-			Title:            section.GetTitle(),
-			Description:      section.GetDescription(),
-			Collapsible:      section.GetCollapsible(),
-			CollapsedDefault: section.GetCollapsedDefault(),
-			FieldKeys:        append([]string(nil), section.GetFieldKeys()...),
-			ShowWhen:         adminFormConditionsToJSON(section.GetShowWhen()),
-		})
-	}
-	return &pluginAdminFormJSON{
-		Fields:      fields,
-		SubmitLabel: form.GetSubmitLabel(),
-		Sections:    sections,
-	}
-}
-
-func adminFormConditionsToJSON(conditions []*pluginv1.AdminFormCondition) []pluginAdminFormConditionJSON {
-	if len(conditions) == 0 {
-		return nil
-	}
-	out := make([]pluginAdminFormConditionJSON, 0, len(conditions))
-	for _, condition := range conditions {
-		if condition == nil {
-			continue
-		}
-		out = append(out, pluginAdminFormConditionJSON{
-			Field:  condition.GetField(),
-			Equals: append([]string(nil), condition.GetEquals()...),
-		})
-	}
-	if len(out) == 0 {
-		return nil
-	}
-	return out
 }
 
 func capabilitiesToJSON(descriptors []*pluginv1.CapabilityDescriptor) []pluginCapabilityJSON {

internal/api/handlers/watch_providers.go

@@ -16,7 +16,7 @@ type WatchProviderService interface {
 	ListProviders() []watchsync.ProviderSummary
 	StartDeviceAuth(ctx context.Context, userID int, profileID string, providerKey string) (watchsync.DeviceAuthSession, error)
 	PollDeviceAuth(ctx context.Context, userID int, profileID string, providerKey string, sessionID string) (watchsync.Connection, error)
-	ConnectAPIKey(ctx context.Context, userID int, profileID string, providerKey string, apiKey string) (watchsync.Connection, error)
+	ConnectAPIKeyWithConfig(ctx context.Context, userID int, profileID string, providerKey string, apiKey string, connectionConfig watchsync.ConnectionConfigValues) (watchsync.Connection, error)
 	GetConnectionStatus(ctx context.Context, userID int, profileID string, provider string) (watchsync.ConnectionStatus, error)
 	UpdateConnection(ctx context.Context, userID int, profileID string, provider string, update watchsync.ConnectionUpdate) (watchsync.ConnectionStatus, error)
 	DeleteConnection(ctx context.Context, userID int, profileID string, provider string) error
@@ -147,13 +147,14 @@ func (h *WatchProviderHandler) HandleConnectAPIKey(w http.ResponseWriter, r *htt
 		return
 	}
 	var req struct {
-		APIKey string `json:"api_key"`
+		APIKey           string                           `json:"api_key"`
+		ConnectionConfig watchsync.ConnectionConfigValues `json:"connection_config"`
 	}
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		writeError(w, http.StatusBadRequest, "bad_request", "Invalid request body")
 		return
 	}
-	if _, err := h.service.ConnectAPIKey(r.Context(), userID, profileID, provider, req.APIKey); err != nil {
+	if _, err := h.service.ConnectAPIKeyWithConfig(r.Context(), userID, profileID, provider, req.APIKey, req.ConnectionConfig); err != nil {
 		writeError(w, http.StatusBadRequest, "watch_provider_error", err.Error())
 		return
 	}

internal/api/handlers/watch_providers_test.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"reflect"
+	"strings"
 	"testing"
 	"time"
 
@@ -23,6 +25,25 @@ type stubWatchProviderService struct {
 	runs      []watchsync.SyncRun
 }
 
+type capturingWatchProviderService struct {
+	stubWatchProviderService
+	apiKey string
+	config watchsync.ConnectionConfigValues
+}
+
+func (s *capturingWatchProviderService) ConnectAPIKeyWithConfig(
+	_ context.Context,
+	_ int,
+	_ string,
+	_ string,
+	apiKey string,
+	config watchsync.ConnectionConfigValues,
+) (watchsync.Connection, error) {
+	s.apiKey = apiKey
+	s.config = config
+	return watchsync.Connection{}, nil
+}
+
 func (s stubWatchProviderService) ListProviders() []watchsync.ProviderSummary {
 	return s.providers
 }
@@ -32,7 +53,7 @@ func (s stubWatchProviderService) StartDeviceAuth(context.Context, int, string,
 func (s stubWatchProviderService) PollDeviceAuth(context.Context, int, string, string, string) (watchsync.Connection, error) {
 	return watchsync.Connection{}, nil
 }
-func (s stubWatchProviderService) ConnectAPIKey(context.Context, int, string, string, string) (watchsync.Connection, error) {
+func (s stubWatchProviderService) ConnectAPIKeyWithConfig(context.Context, int, string, string, string, watchsync.ConnectionConfigValues) (watchsync.Connection, error) {
 	return watchsync.Connection{}, nil
 }
 func (s stubWatchProviderService) GetConnectionStatus(context.Context, int, string, string) (watchsync.ConnectionStatus, error) {
@@ -89,7 +110,7 @@ func TestWatchProviderHandlerListsProviders(t *testing.T) {
 	if len(resp.Providers) != 1 {
 		t.Fatalf("providers length = %d, want 1", len(resp.Providers))
 	}
-	if resp.Providers[0] != service.providers[0] {
+	if !reflect.DeepEqual(resp.Providers[0], service.providers[0]) {
 		t.Fatalf("provider = %#v, want %#v", resp.Providers[0], service.providers[0])
 	}
 }
@@ -175,6 +196,31 @@ func TestWatchProviderHandlerStartsDeviceAuthWithFrontendJSONShape(t *testing.T)
 	}
 }
 
+func TestWatchProviderHandlerPassesConnectionConfigWithAPIKey(t *testing.T) {
+	service := &capturingWatchProviderService{}
+	handler := NewWatchProviderHandler(service)
+	req := httptest.NewRequest(http.MethodPost, "/watch-providers/plugin:4:floppy/auth/api-key", strings.NewReader(`{
+		"api_key":"token",
+		"connection_config":{"floppy":{"base_url":"https://floppy.example.com"}}
+	}`))
+	routeCtx := chi.NewRouteContext()
+	routeCtx.URLParams.Add("provider", "plugin:4:floppy")
+	ctx := context.WithValue(req.Context(), chi.RouteCtxKey, routeCtx)
+	ctx = middleware.SetClaims(ctx, &auth.Claims{UserID: 7})
+	ctx = middleware.SetProfileID(ctx, "profile-1")
+	req = req.WithContext(ctx)
+	rec := httptest.NewRecorder()
+
+	handler.HandleConnectAPIKey(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
+	}
+	if service.apiKey != "token" || service.config["floppy"]["base_url"] != "https://floppy.example.com" {
+		t.Fatalf("api key = %q, config = %#v", service.apiKey, service.config)
+	}
+}
+
 func TestWatchProviderHandlerManualSyncReturnsRun(t *testing.T) {
 	startedAt := time.Date(2026, 5, 4, 16, 0, 0, 0, time.UTC)
 	handler := NewWatchProviderHandler(stubWatchProviderService{

internal/pluginhost/handshake.go

@@ -23,7 +23,9 @@ const (
 	DefaultEventTimeout      = 10 * time.Second
 	DefaultAuthTimeout       = 10 * time.Second
 	DefaultRouteTimeout      = 10 * time.Second
-	DefaultWatchSyncTimeout  = 60 * time.Second
+	// Watch-sync providers may resolve cold external metadata before applying
+	// an event. Bound that work without cutting off valid slow provider calls.
+	DefaultWatchSyncTimeout = 2 * time.Minute
 	// DefaultRequestRouterTimeout bounds a single request_router RPC.
 	// Fulfillment hits remote arr instances, so allow generous headroom.
 	DefaultRequestRouterTimeout = 60 * time.Second

internal/plugins/config_schema_view.go

@@ -0,0 +1,150 @@
+package plugins
+
+import (
+	"strings"
+
+	pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1"
+)
+
+// ConfigSchemaView is the transport-safe representation of a manifest config
+// schema used by non-admin setup surfaces.
+type ConfigSchemaView struct {
+	Key         string         `json:"key"`
+	Title       string         `json:"title"`
+	Description string         `json:"description"`
+	JSONSchema  string         `json:"json_schema"`
+	Required    bool           `json:"required"`
+	AdminForm   *AdminFormView `json:"admin_form,omitempty"`
+}
+
+type AdminFormView struct {
+	Fields      []AdminFormFieldView   `json:"fields"`
+	SubmitLabel string                 `json:"submit_label,omitempty"`
+	Sections    []AdminFormSectionView `json:"sections,omitempty"`
+}
+
+type AdminFormFieldView struct {
+	Key                 string                   `json:"key"`
+	Label               string                   `json:"label"`
+	Description         string                   `json:"description,omitempty"`
+	Control             string                   `json:"control"`
+	Placeholder         string                   `json:"placeholder,omitempty"`
+	Required            bool                     `json:"required"`
+	Secret              bool                     `json:"secret"`
+	Multiline           bool                     `json:"multiline"`
+	DefaultValue        any                      `json:"default_value,omitempty"`
+	Options             []AdminFormOptionView    `json:"options,omitempty"`
+	Rows                int32                    `json:"rows,omitempty"`
+	DynamicOptions      bool                     `json:"dynamic_options,omitempty"`
+	ShowWhen            []AdminFormConditionView `json:"show_when,omitempty"`
+	Validation          *AdminFormValidationView `json:"validation,omitempty"`
+	ExclusiveGroupField string                   `json:"exclusive_group_field,omitempty"`
+}
+
+type AdminFormOptionView struct {
+	Value       string `json:"value"`
+	Label       string `json:"label"`
+	Description string `json:"description,omitempty"`
+}
+
+type AdminFormConditionView struct {
+	Field  string   `json:"field"`
+	Equals []string `json:"equals"`
+}
+
+type AdminFormValidationView struct {
+	HasMin    bool    `json:"has_min,omitempty"`
+	Min       float64 `json:"min,omitempty"`
+	HasMax    bool    `json:"has_max,omitempty"`
+	Max       float64 `json:"max,omitempty"`
+	Pattern   string  `json:"pattern,omitempty"`
+	MinLength int32   `json:"min_length,omitempty"`
+	MaxLength int32   `json:"max_length,omitempty"`
+}
+
+type AdminFormSectionView struct {
+	Key              string                   `json:"key"`
+	Title            string                   `json:"title"`
+	Description      string                   `json:"description,omitempty"`
+	Collapsible      bool                     `json:"collapsible"`
+	CollapsedDefault bool                     `json:"collapsed_default"`
+	FieldKeys        []string                 `json:"field_keys"`
+	ShowWhen         []AdminFormConditionView `json:"show_when,omitempty"`
+}
+
+func ConfigSchemaViews(schemas []*pluginv1.ConfigSchema) []ConfigSchemaView {
+	views := make([]ConfigSchemaView, 0, len(schemas))
+	for _, schema := range schemas {
+		if schema == nil {
+			continue
+		}
+		views = append(views, ConfigSchemaView{
+			Key:         schema.GetKey(),
+			Title:       schema.GetTitle(),
+			Description: schema.GetDescription(),
+			JSONSchema:  schema.GetJsonSchema(),
+			Required:    schema.GetRequired(),
+			AdminForm:   AdminFormViewFromProto(schema.GetAdminForm()),
+		})
+	}
+	return views
+}
+
+func AdminFormViewFromProto(form *pluginv1.AdminFormDescriptor) *AdminFormView {
+	if form == nil {
+		return nil
+	}
+	fields := make([]AdminFormFieldView, 0, len(form.GetFields()))
+	for _, field := range form.GetFields() {
+		if field == nil {
+			continue
+		}
+		options := make([]AdminFormOptionView, 0, len(field.GetOptions()))
+		for _, option := range field.GetOptions() {
+			if option != nil {
+				options = append(options, AdminFormOptionView{Value: option.GetValue(), Label: option.GetLabel(), Description: option.GetDescription()})
+			}
+		}
+		var defaultValue any
+		if field.GetDefaultValue() != nil {
+			defaultValue = field.GetDefaultValue().AsInterface()
+		}
+		var validation *AdminFormValidationView
+		if value := field.GetValidation(); value != nil {
+			validation = &AdminFormValidationView{
+				HasMin: value.GetHasMin(), Min: value.GetMin(), HasMax: value.GetHasMax(), Max: value.GetMax(),
+				Pattern: value.GetPattern(), MinLength: value.GetMinLength(), MaxLength: value.GetMaxLength(),
+			}
+		}
+		fields = append(fields, AdminFormFieldView{
+			Key: field.GetKey(), Label: field.GetLabel(), Description: field.GetDescription(),
+			Control:     strings.TrimPrefix(field.GetControl().String(), "ADMIN_FORM_CONTROL_"),
+			Placeholder: field.GetPlaceholder(), Required: field.GetRequired(), Secret: field.GetSecret(),
+			Multiline: field.GetMultiline(), DefaultValue: defaultValue, Options: options, Rows: field.GetRows(),
+			DynamicOptions: field.GetDynamicOptions(), ShowWhen: adminFormConditionViews(field.GetShowWhen()),
+			Validation: validation, ExclusiveGroupField: field.GetExclusiveGroupField(),
+		})
+	}
+	sections := make([]AdminFormSectionView, 0, len(form.GetSections()))
+	for _, section := range form.GetSections() {
+		if section == nil {
+			continue
+		}
+		sections = append(sections, AdminFormSectionView{
+			Key: section.GetKey(), Title: section.GetTitle(), Description: section.GetDescription(),
+			Collapsible: section.GetCollapsible(), CollapsedDefault: section.GetCollapsedDefault(),
+			FieldKeys: append([]string(nil), section.GetFieldKeys()...), ShowWhen: adminFormConditionViews(section.GetShowWhen()),
+		})
+	}
+	return &AdminFormView{Fields: fields, SubmitLabel: form.GetSubmitLabel(), Sections: sections}
+}
+
+func adminFormConditionViews(conditions []*pluginv1.AdminFormCondition) []AdminFormConditionView {
+	views := make([]AdminFormConditionView, 0, len(conditions))
+	for _, condition := range conditions {
+		if condition != nil {
+			views = append(views, AdminFormConditionView{Field: condition.GetField(), Equals: append([]string(nil), condition.GetEquals()...)})
+		}
+	}
+	return views
+}

internal/plugins/config_secrets.go

@@ -16,9 +16,16 @@ func GlobalConfigFieldSets(
 	manifest *pluginv1.PluginManifest,
 	configKey string,
 ) (publicFields, secretFields []string) {
+	return ConfigSchemaFieldSets(globalConfigSchema(manifest, configKey))
+}
+
+// ConfigSchemaFieldSets returns the top-level public and secret fields for one
+// configuration schema. JSON Schema annotations and Admin form controls are
+// both authoritative; undeclared fields are intentionally absent from both
+// sets so callers can keep them on the protected path.
+func ConfigSchemaFieldSets(schema *pluginv1.ConfigSchema) (publicFields, secretFields []string) {
 	declared := make(map[string]struct{})
 	secrets := make(map[string]struct{})
-	schema := globalConfigSchema(manifest, configKey)
 	if schema != nil {
 		if form := schema.GetAdminForm(); form != nil {
 			for _, field := range form.GetFields() {

internal/plugins/watch_sync_config_test.go

@@ -22,6 +22,11 @@ func TestWatchSyncProviderConfigClassifiesManifestFields(t *testing.T) {
 			"undeclared":    map[string]any{"token": "also-secret"},
 			"   ":           "ignored-empty-field",
 		},
+	}, {
+		Key: "legacy",
+		Value: map[string]any{
+			"base_url": "https://legacy-floppy.example",
+		},
 	}, {
 		Key: "   ",
 		Value: map[string]any{
@@ -33,7 +38,8 @@ func TestWatchSyncProviderConfigClassifiesManifestFields(t *testing.T) {
 	}
 	if config.GetValues()["provider.base_url"] != "https://floppy.example" ||
 		config.GetSecretValues()["provider.client_secret"] != "secret" ||
-		config.GetSecretValues()["provider.undeclared"] != `{"token":"also-secret"}` {
+		config.GetSecretValues()["provider.undeclared"] != `{"token":"also-secret"}` ||
+		config.GetSecretValues()["legacy.base_url"] != "https://legacy-floppy.example" {
 		t.Fatalf("config = %#v", config)
 	}
 	if _, exposed := config.GetValues()["provider.undeclared"]; exposed {

internal/watchsync/plugin_provider.go

@@ -3,15 +3,21 @@ package watchsync
 import (
 	"context"
 	"encoding/base64"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/url"
+	"regexp"
+	"strconv"
 	"strings"
 	"time"
 	"unicode"
+	"unicode/utf8"
 
 	pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1"
+	publicconfig "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginsdk/config"
 	"github.com/Silo-Server/silo-server/internal/historyimport"
+	hostplugins "github.com/Silo-Server/silo-server/internal/plugins"
 	"github.com/Silo-Server/silo-server/internal/userstore"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -34,33 +40,37 @@ type PluginCredentialRepository interface {
 }
 
 type PluginProviderOptions struct {
-	InstallationID int
+	InstallationID         int
-	ProviderKey    string
+	ProviderKey            string
-	CapabilityID   string
+	CapabilityID           string
-	DisplayName    string
+	DisplayName            string
-	Descriptor     *pluginv1.WatchSyncProviderDescriptor
+	Descriptor             *pluginv1.WatchSyncProviderDescriptor
-	ResolveClient  WatchSyncPluginClientResolver
+	ConnectionConfigSchema []*pluginv1.ConfigSchema
-	ResolveConfig  WatchSyncPluginConfigResolver
+	ResolveClient          WatchSyncPluginClientResolver
-	Repository     PluginCredentialRepository
+	ResolveConfig          WatchSyncPluginConfigResolver
+	Repository             PluginCredentialRepository
 }
 
 type PluginProvider struct {
-	installationID int
+	installationID         int
-	providerKey    string
+	providerKey            string
-	capabilityID   string
+	capabilityID           string
-	displayName    string
+	displayName            string
-	descriptor     *pluginv1.WatchSyncProviderDescriptor
+	descriptor             *pluginv1.WatchSyncProviderDescriptor
-	authMethod     string
+	connectionConfigSchema []*pluginv1.ConfigSchema
-	supportedMedia map[pluginv1.WatchSyncMediaType]struct{}
+	authMethod             string
-	resolveClient  WatchSyncPluginClientResolver
+	supportedMedia         map[pluginv1.WatchSyncMediaType]struct{}
-	resolveConfig  WatchSyncPluginConfigResolver
+	resolveClient          WatchSyncPluginClientResolver
-	repository     PluginCredentialRepository
+	resolveConfig          WatchSyncPluginConfigResolver
+	repository             PluginCredentialRepository
 }
 
 const (
 	watchSyncUnsupportedMovieMediaMessage   = "watch sync plugin does not support movie media"
 	watchSyncUnsupportedEpisodeMediaMessage = "watch sync plugin does not support episode media"
 	watchSyncUnsupportedMediaMessage        = "watch sync plugin does not support this media type"
+	watchSyncJSONSchemaNumberType           = "number"
+	watchSyncJSONSchemaBooleanType          = "boolean"
 )
 
 func NewPluginProvider(options PluginProviderOptions) (*PluginProvider, error) {
@@ -81,23 +91,39 @@ func NewPluginProvider(options PluginProviderOptions) (*PluginProvider, error) {
 	if err != nil {
 		return nil, fmt.Errorf("watch sync plugin %q %w", options.ProviderKey, err)
 	}
+	if err := validateWatchSyncConnectionConfigSchemas(options.ConnectionConfigSchema); err != nil {
+		return nil, fmt.Errorf("watch sync plugin %q %w", options.ProviderKey, err)
+	}
+	if authMethod != AuthMethodAPIKey && hasWatchSyncConnectionConfigSchema(options.ConnectionConfigSchema) {
+		return nil, fmt.Errorf("watch sync plugin %q connection config requires API-key authentication", options.ProviderKey)
+	}
 	if options.ResolveClient == nil {
 		return nil, fmt.Errorf("watch sync plugin client resolver is required")
 	}
 	return &PluginProvider{
-		installationID: options.InstallationID,
+		installationID:         options.InstallationID,
-		providerKey:    options.ProviderKey,
+		providerKey:            options.ProviderKey,
-		capabilityID:   options.CapabilityID,
+		capabilityID:           options.CapabilityID,
-		displayName:    options.DisplayName,
+		displayName:            options.DisplayName,
-		descriptor:     options.Descriptor,
+		descriptor:             options.Descriptor,
-		authMethod:     authMethod,
+		connectionConfigSchema: append([]*pluginv1.ConfigSchema(nil), options.ConnectionConfigSchema...),
-		supportedMedia: supportedMedia,
+		authMethod:             authMethod,
-		resolveClient:  options.ResolveClient,
+		supportedMedia:         supportedMedia,
-		resolveConfig:  options.ResolveConfig,
+		resolveClient:          options.ResolveClient,
-		repository:     options.Repository,
+		resolveConfig:          options.ResolveConfig,
+		repository:             options.Repository,
 	}, nil
 }
 
+func hasWatchSyncConnectionConfigSchema(schemas []*pluginv1.ConfigSchema) bool {
+	for _, schema := range schemas {
+		if schema != nil {
+			return true
+		}
+	}
+	return false
+}
+
 func (p *PluginProvider) Key() string { return p.providerKey }
 
 func (p *PluginProvider) DisplayName() string {
@@ -118,6 +144,10 @@ func (p *PluginProvider) HistorySource() userstore.WatchHistorySource {
 
 func (p *PluginProvider) AuthMethod() string { return p.authMethod }
 
+func (p *PluginProvider) ConnectionConfigSchema() []hostplugins.ConfigSchemaView {
+	return hostplugins.ConfigSchemaViews(p.connectionConfigSchema)
+}
+
 func (p *PluginProvider) usesHostPluginConfig() {}
 
 func (p *PluginProvider) authoritativeRefreshProvider() {}
@@ -147,6 +177,14 @@ func (p *PluginProvider) Capabilities() Capabilities {
 }
 
 func (p *PluginProvider) ConnectWithAPIKey(ctx context.Context, apiKey string) (TokenSet, ProviderAccount, error) {
+	return p.ConnectWithAPIKeyConfig(ctx, apiKey, nil)
+}
+
+func (p *PluginProvider) ConnectWithAPIKeyConfig(
+	ctx context.Context,
+	apiKey string,
+	connectionConfig ConnectionConfigValues,
+) (TokenSet, ProviderAccount, error) {
 	if p.authMethod != AuthMethodAPIKey {
 		return TokenSet{}, ProviderAccount{}, errors.New("watch sync plugin does not support API-key authentication")
 	}
@@ -154,6 +192,11 @@ func (p *PluginProvider) ConnectWithAPIKey(ctx context.Context, apiKey string) (
 	if err != nil {
 		return TokenSet{}, ProviderAccount{}, err
 	}
+	connectionValues, connectionSecrets, err := p.connectionConfig(connectionConfig)
+	if err != nil {
+		return TokenSet{}, ProviderAccount{}, err
+	}
+	config = mergeWatchSyncProviderConfig(config, connectionValues)
 	client, err := p.resolveClient(ctx, p.installationID, p.capabilityID)
 	if err != nil {
 		return TokenSet{}, ProviderAccount{}, watchSyncUnavailableError()
@@ -166,7 +209,8 @@ func (p *PluginProvider) ConnectWithAPIKey(ctx context.Context, apiKey string) (
 	if err != nil {
 		return TokenSet{}, ProviderAccount{}, watchSyncRPCError()
 	}
-	if err := watchSyncFaultError(p.Key(), response.GetFault(), apiKey); err != nil {
+	faultSecrets := append([]string{apiKey}, connectionSecrets...)
+	if err := watchSyncFaultError(p.Key(), response.GetFault(), faultSecrets...); err != nil {
 		return TokenSet{}, ProviderAccount{}, err
 	}
 	tokens, err := tokenSetFromProto(response.GetCredentials())
@@ -533,6 +577,497 @@ func (p *PluginProvider) providerConfig(ctx context.Context) (*pluginv1.WatchSyn
 	return config, nil
 }
 
+func (p *PluginProvider) connectionConfig(values ConnectionConfigValues) (*pluginv1.WatchSyncProviderConfig, []string, error) {
+	declared := make(map[string]*pluginv1.ConfigSchema, len(p.connectionConfigSchema))
+	for _, schema := range p.connectionConfigSchema {
+		if schema != nil && strings.TrimSpace(schema.GetKey()) != "" {
+			declared[schema.GetKey()] = schema
+		}
+	}
+	secrets := connectionConfigSecrets(p.connectionConfigSchema, values)
+	for key := range values {
+		if _, ok := declared[key]; !ok {
+			return nil, nil, sanitizedConnectionConfigError(
+				fmt.Errorf("watch sync connection config key %q is not declared", key),
+				secrets,
+			)
+		}
+	}
+
+	result := &pluginv1.WatchSyncProviderConfig{
+		Values:       make(map[string]string),
+		SecretValues: make(map[string]string),
+	}
+	flattenedFields := make(map[string]string)
+	for _, schema := range p.connectionConfigSchema {
+		if schema == nil || strings.TrimSpace(schema.GetKey()) == "" {
+			continue
+		}
+		value, exists := values[schema.GetKey()]
+		if !exists {
+			if schema.GetRequired() {
+				return nil, nil, fmt.Errorf("watch sync connection config %q is required", schema.GetKey())
+			}
+			continue
+		}
+		if err := publicconfig.ValidateValue(schema, "watch sync connection config", schema.GetKey(), value); err != nil {
+			return nil, nil, sanitizedConnectionConfigError(err, secrets)
+		}
+		if err := validateConnectionAdminFormValue(schema, value); err != nil {
+			return nil, nil, sanitizedConnectionConfigError(err, secrets)
+		}
+		publicFieldNames, _ := hostplugins.ConfigSchemaFieldSets(schema)
+		publicFields := make(map[string]struct{}, len(publicFieldNames))
+		for _, field := range publicFieldNames {
+			publicFields[field] = struct{}{}
+		}
+		for field, raw := range value {
+			rawField := field
+			field = strings.TrimSpace(field)
+			if field == "" {
+				continue
+			}
+			encoded, err := connectionConfigString(raw)
+			if err != nil {
+				return nil, nil, sanitizedConnectionConfigError(
+					fmt.Errorf("encode watch sync connection config %q.%s: %w", schema.GetKey(), field, err),
+					secrets,
+				)
+			}
+			key := schema.GetKey() + "." + field
+			source := fmt.Sprintf("config %q field %q", schema.GetKey(), rawField)
+			if previous, exists := flattenedFields[key]; exists {
+				return nil, nil, sanitizedConnectionConfigError(
+					fmt.Errorf("watch sync connection %s conflicts with %s after flattening to %q", source, previous, key),
+					secrets,
+				)
+			}
+			flattenedFields[key] = source
+			if _, public := publicFields[field]; public {
+				result.Values[key] = encoded
+			} else {
+				result.SecretValues[key] = encoded
+			}
+		}
+	}
+	return result, secrets, nil
+}
+
+func connectionConfigSecrets(schemas []*pluginv1.ConfigSchema, values ConnectionConfigValues) []string {
+	var secrets []string
+	for _, schema := range schemas {
+		if schema == nil {
+			continue
+		}
+		value, exists := values[schema.GetKey()]
+		if !exists {
+			continue
+		}
+		_, secretFields := hostplugins.ConfigSchemaFieldSets(schema)
+		for _, field := range secretFields {
+			raw, exists := value[field]
+			if !exists {
+				continue
+			}
+			if encoded, err := connectionConfigString(raw); err == nil && strings.TrimSpace(encoded) != "" {
+				secrets = append(secrets, encoded)
+			}
+			secrets = append(secrets, connectionConfigSecretStrings(raw)...)
+		}
+	}
+	return secrets
+}
+
+func sanitizedConnectionConfigError(err error, secrets []string) error {
+	return errors.New(sanitizeWatchSyncMessage(
+		err.Error(),
+		"watch sync connection config is invalid",
+		secrets...,
+	))
+}
+
+func connectionConfigSecretStrings(value any) []string {
+	switch typed := value.(type) {
+	case map[string]any:
+		var values []string
+		for _, child := range typed {
+			values = append(values, connectionConfigSecretStrings(child)...)
+		}
+		return values
+	case []any:
+		var values []string
+		for _, child := range typed {
+			values = append(values, connectionConfigSecretStrings(child)...)
+		}
+		return values
+	default:
+		encoded, err := connectionConfigString(typed)
+		if err != nil || strings.TrimSpace(encoded) == "" {
+			return nil
+		}
+		return []string{encoded}
+	}
+}
+
+func validateWatchSyncConnectionConfigSchemas(schemas []*pluginv1.ConfigSchema) error {
+	seen := make(map[string]struct{}, len(schemas))
+	for _, schema := range schemas {
+		if schema == nil {
+			continue
+		}
+		key := strings.TrimSpace(schema.GetKey())
+		if key == "" {
+			return fmt.Errorf("connection config key is required")
+		}
+		if _, exists := seen[key]; exists {
+			return fmt.Errorf("connection config key %q is duplicated", key)
+		}
+		seen[key] = struct{}{}
+		for _, field := range schema.GetAdminForm().GetFields() {
+			if field == nil {
+				continue
+			}
+			if strings.TrimSpace(field.GetExclusiveGroupField()) != "" {
+				return fmt.Errorf(
+					"connection config %q field %q uses exclusive_group_field, which watch provider setup does not support",
+					schema.GetKey(),
+					field.GetKey(),
+				)
+			}
+			if field.GetDynamicOptions() && len(field.GetOptions()) == 0 {
+				return fmt.Errorf(
+					"connection config %q field %q requires dynamic options, which watch provider setup does not support",
+					schema.GetKey(),
+					field.GetKey(),
+				)
+			}
+			if field.GetControl() == pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SELECT {
+				for _, option := range field.GetOptions() {
+					if option != nil && strings.TrimSpace(option.GetValue()) == "" {
+						return fmt.Errorf(
+							"connection config %q field %q has a blank select option value",
+							schema.GetKey(),
+							field.GetKey(),
+						)
+					}
+				}
+			}
+			if pattern := field.GetValidation().GetPattern(); pattern != "" {
+				if _, err := regexp.Compile(pattern); err != nil {
+					return fmt.Errorf(
+						"connection config %q field %q has invalid validation pattern: %w",
+						schema.GetKey(),
+						field.GetKey(),
+						err,
+					)
+				}
+			}
+		}
+		if err := validateConnectionSchemaIsRenderable(schema); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func validateConnectionSchemaIsRenderable(schema *pluginv1.ConfigSchema) error {
+	if strings.TrimSpace(schema.GetJsonSchema()) == "" && !schema.GetRequired() {
+		return nil
+	}
+	var document struct {
+		Type       string `json:"type"`
+		Properties map[string]struct {
+			Type  string `json:"type"`
+			Items *struct {
+				Type string `json:"type"`
+			} `json:"items"`
+		} `json:"properties"`
+	}
+	if err := json.Unmarshal([]byte(schema.GetJsonSchema()), &document); err != nil {
+		return fmt.Errorf("connection config %q has invalid json_schema: %w", schema.GetKey(), err)
+	}
+	if document.Type != "object" || document.Properties == nil {
+		return fmt.Errorf("connection config %q must have a renderable object json_schema", schema.GetKey())
+	}
+	explicit := make(map[string]*pluginv1.AdminFormField)
+	if form := schema.GetAdminForm(); form != nil {
+		for _, field := range form.GetFields() {
+			if field != nil {
+				explicit[field.GetKey()] = field
+			}
+		}
+	}
+	for key, property := range document.Properties {
+		field := explicit[key]
+		switch property.Type {
+		case "string", watchSyncJSONSchemaNumberType, "integer", watchSyncJSONSchemaBooleanType:
+			if field != nil && !connectionAdminFieldSupportsScalarType(field, property.Type) {
+				return fmt.Errorf(
+					"connection config %q field %q control %q cannot emit json_schema type %q",
+					schema.GetKey(),
+					key,
+					field.GetControl().String(),
+					property.Type,
+				)
+			}
+			continue
+		case "array":
+			if field != nil && field.GetControl() == pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_MULTI_SELECT &&
+				(property.Items == nil || property.Items.Type == "string" || property.Items.Type == watchSyncJSONSchemaNumberType ||
+					property.Items.Type == "integer" || property.Items.Type == watchSyncJSONSchemaBooleanType) {
+				continue
+			}
+		default:
+			// A property whose shape comes from enum/const/$ref cannot be
+			// inferred from type alone, but an explicit scalar form control is
+			// still a complete input mechanism. Direct object properties remain
+			// unsupported because none of these controls produces an object.
+			if property.Type != "object" && connectionAdminFieldRendersValue(field) {
+				continue
+			}
+		}
+		return fmt.Errorf(
+			"connection config %q property %q needs a renderable admin_form field because type %q cannot be inferred",
+			schema.GetKey(),
+			key,
+			property.Type,
+		)
+	}
+	return nil
+}
+
+func connectionAdminFieldSupportsScalarType(field *pluginv1.AdminFormField, propertyType string) bool {
+	switch field.GetControl() {
+	case pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SWITCH:
+		return propertyType == watchSyncJSONSchemaBooleanType
+	case pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_MULTI_SELECT:
+		return false
+	default:
+		return true
+	}
+}
+
+func connectionAdminFieldRendersValue(field *pluginv1.AdminFormField) bool {
+	if field == nil {
+		return false
+	}
+	switch field.GetControl() {
+	case pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT,
+		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXTAREA,
+		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_PASSWORD,
+		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_NUMBER,
+		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SWITCH,
+		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SELECT,
+		pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_MULTI_SELECT:
+		return true
+	default:
+		return false
+	}
+}
+
+func validateConnectionAdminFormValue(schema *pluginv1.ConfigSchema, value map[string]any) error {
+	if schema == nil || schema.GetAdminForm() == nil {
+		return nil
+	}
+	for _, field := range schema.GetAdminForm().GetFields() {
+		if field == nil || !connectionAdminFieldIsVisible(schema.GetAdminForm(), field, value) {
+			continue
+		}
+		raw, exists := value[field.GetKey()]
+		if !exists || connectionConfigValueIsEmpty(raw) {
+			if field.GetRequired() {
+				return fmt.Errorf("connection config %q field %q is required", schema.GetKey(), field.GetKey())
+			}
+			continue
+		}
+		if field.GetValidation() == nil {
+			continue
+		}
+		validation := field.GetValidation()
+		if text, ok := raw.(string); ok {
+			if pattern := validation.GetPattern(); pattern != "" {
+				matched, err := regexp.MatchString(pattern, text)
+				if err != nil {
+					return fmt.Errorf("connection config %q field %q has an invalid validation pattern", schema.GetKey(), field.GetKey())
+				}
+				if !matched {
+					return fmt.Errorf("connection config %q field %q is invalid", schema.GetKey(), field.GetKey())
+				}
+			}
+			length := utf8.RuneCountInString(text)
+			if minimum := int(validation.GetMinLength()); minimum > 0 && length < minimum {
+				return fmt.Errorf("connection config %q field %q must be at least %d characters", schema.GetKey(), field.GetKey(), minimum)
+			}
+			if maximum := int(validation.GetMaxLength()); maximum > 0 && length > maximum {
+				return fmt.Errorf("connection config %q field %q must be at most %d characters", schema.GetKey(), field.GetKey(), maximum)
+			}
+		}
+		if number, ok := connectionConfigNumber(raw); ok {
+			if validation.GetHasMin() && number < validation.GetMin() {
+				return fmt.Errorf("connection config %q field %q must be at least %g", schema.GetKey(), field.GetKey(), validation.GetMin())
+			}
+			if validation.GetHasMax() && number > validation.GetMax() {
+				return fmt.Errorf("connection config %q field %q must be at most %g", schema.GetKey(), field.GetKey(), validation.GetMax())
+			}
+		}
+	}
+	return nil
+}
+
+func connectionConfigValueIsEmpty(value any) bool {
+	switch typed := value.(type) {
+	case nil:
+		return true
+	case string:
+		return strings.TrimSpace(typed) == ""
+	case []any:
+		return len(typed) == 0
+	default:
+		return false
+	}
+}
+
+func connectionAdminFieldIsVisible(
+	form *pluginv1.AdminFormDescriptor,
+	field *pluginv1.AdminFormField,
+	values map[string]any,
+) bool {
+	if !connectionAdminConditionsMatch(field.GetShowWhen(), values, form.GetFields()) {
+		return false
+	}
+	contained := false
+	for _, section := range form.GetSections() {
+		if section == nil || !stringSliceContains(section.GetFieldKeys(), field.GetKey()) {
+			continue
+		}
+		contained = true
+		if connectionAdminConditionsMatch(section.GetShowWhen(), values, form.GetFields()) {
+			return true
+		}
+	}
+	return !contained
+}
+
+func connectionAdminConditionsMatch(
+	conditions []*pluginv1.AdminFormCondition,
+	values map[string]any,
+	fields []*pluginv1.AdminFormField,
+) bool {
+	for _, condition := range conditions {
+		if condition == nil {
+			continue
+		}
+		value, exists := values[condition.GetField()]
+		if !exists {
+			for _, field := range fields {
+				if field != nil && field.GetKey() == condition.GetField() && field.GetDefaultValue() != nil {
+					value = field.GetDefaultValue().AsInterface()
+					break
+				}
+			}
+		}
+		if !stringSliceContains(condition.GetEquals(), connectionAdminConditionString(value)) {
+			return false
+		}
+	}
+	return true
+}
+
+func connectionAdminConditionString(value any) string {
+	switch typed := value.(type) {
+	case nil:
+		return ""
+	case bool:
+		return strconv.FormatBool(typed)
+	case float64:
+		return strconv.FormatFloat(typed, 'f', -1, 64)
+	case string:
+		return typed
+	default:
+		return fmt.Sprint(typed)
+	}
+}
+
+func stringSliceContains(values []string, target string) bool {
+	for _, value := range values {
+		if value == target {
+			return true
+		}
+	}
+	return false
+}
+
+func connectionConfigNumber(value any) (float64, bool) {
+	switch typed := value.(type) {
+	case float64:
+		return typed, true
+	case float32:
+		return float64(typed), true
+	case int:
+		return float64(typed), true
+	case int8:
+		return float64(typed), true
+	case int16:
+		return float64(typed), true
+	case int32:
+		return float64(typed), true
+	case int64:
+		return float64(typed), true
+	case uint:
+		return float64(typed), true
+	case uint8:
+		return float64(typed), true
+	case uint16:
+		return float64(typed), true
+	case uint32:
+		return float64(typed), true
+	case uint64:
+		return float64(typed), true
+	case json.Number:
+		number, err := typed.Float64()
+		return number, err == nil
+	case string:
+		number, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
+		return number, err == nil
+	default:
+		return 0, false
+	}
+}
+
+func connectionConfigString(value any) (string, error) {
+	if text, ok := value.(string); ok {
+		return text, nil
+	}
+	encoded, err := json.Marshal(value)
+	if err != nil {
+		return "", err
+	}
+	return string(encoded), nil
+}
+
+func mergeWatchSyncProviderConfig(base, connection *pluginv1.WatchSyncProviderConfig) *pluginv1.WatchSyncProviderConfig {
+	merged := &pluginv1.WatchSyncProviderConfig{Values: map[string]string{}, SecretValues: map[string]string{}}
+	if base != nil {
+		for key, value := range base.GetValues() {
+			merged.Values[key] = value
+		}
+		for key, value := range base.GetSecretValues() {
+			merged.SecretValues[key] = value
+		}
+	}
+	if connection != nil {
+		for key, value := range connection.GetValues() {
+			delete(merged.SecretValues, key)
+			merged.Values[key] = value
+		}
+		for key, value := range connection.GetSecretValues() {
+			delete(merged.Values, key)
+			merged.SecretValues[key] = value
+		}
+	}
+	return merged
+}
+
 func (p *PluginProvider) persistUpdatedCredentials(
 	ctx context.Context,
 	conn Connection,
@@ -587,6 +1122,7 @@ func watchEventFromScrobble(event ScrobbleEvent, operation pluginv1.WatchSyncOpe
 		PositionSeconds:   event.PositionSeconds,
 		DurationSeconds:   event.DurationSeconds,
 		CompletionPercent: completion,
+		Completed:         event.Completed,
 		ProviderItemKey:   event.ProviderItemKey,
 		Media: mediaFromIdentity(event.MediaItemID, event.Kind, "", 0,
 			event.IMDbID, event.TMDBID, event.TVDBID, "", 0,

internal/watchsync/plugin_provider_test.go

@@ -34,6 +34,7 @@ type fakeWatchSyncPluginClient struct {
 	listResponses       []*pluginv1.WatchSyncListRemoteStateResponse
 	applyErr            error
 	applyRequest        *pluginv1.WatchSyncApplyEventsRequest
+	exchangeRequest     *pluginv1.WatchSyncExchangeAPIKeyRequest
 	refreshRequest      *pluginv1.WatchSyncRefreshCredentialsRequest
 	accountRequest      *pluginv1.WatchSyncGetAccountRequest
 	deviceStartRequest  *pluginv1.WatchSyncDeviceAuthorizationServiceStartRequest
@@ -57,7 +58,8 @@ func (f *fakeWatchSyncPluginClient) PollDeviceAuthorization(_ context.Context, r
 	return &pluginv1.WatchSyncDeviceAuthorizationServicePollResponse{}, nil
 }
 
-func (f *fakeWatchSyncPluginClient) ExchangeAPIKey(_ context.Context, _ *pluginv1.WatchSyncExchangeAPIKeyRequest) (*pluginv1.WatchSyncCredentialResponse, error) {
+func (f *fakeWatchSyncPluginClient) ExchangeAPIKey(_ context.Context, req *pluginv1.WatchSyncExchangeAPIKeyRequest) (*pluginv1.WatchSyncCredentialResponse, error) {
+	f.exchangeRequest = req
 	return f.exchangeResponse, nil
 }
 func (f *fakeWatchSyncPluginClient) RefreshCredentials(_ context.Context, req *pluginv1.WatchSyncRefreshCredentialsRequest) (*pluginv1.WatchSyncCredentialResponse, error) {
@@ -215,6 +217,476 @@ func TestPluginProviderRejectsMissingAccountIdentity(t *testing.T) {
 	}
 }
 
+func TestPluginProviderValidatesAndOverlaysConnectionConfig(t *testing.T) {
+	client := &fakeWatchSyncPluginClient{exchangeResponse: &pluginv1.WatchSyncCredentialResponse{
+		Credentials: &pluginv1.WatchSyncCredentials{AccessToken: testValidatedToken, TokenType: testBearerTokenType},
+		Account:     &pluginv1.WatchSyncAccount{ExternalSubject: "7", Username: testPluginUsername},
+	}}
+	schema := &pluginv1.ConfigSchema{
+		Key: "floppy", Title: "Floppy server", Required: true,
+		JsonSchema: `{"type":"object","properties":{"base_url":{"type":"string","format":"uri"}},"required":["base_url"],"additionalProperties":false}`,
+		AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+			Key: "base_url", Label: "Base URL", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT, Required: true,
+		}}},
+	}
+	provider, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		DisplayName: "Floppy", Descriptor: &pluginv1.WatchSyncProviderDescriptor{
+			AuthMethods: []pluginv1.WatchSyncAuthMethod{pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY},
+		},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{schema},
+		ResolveClient:          func(context.Context, int, string) (WatchSyncPluginClient, error) { return client, nil },
+		ResolveConfig: func(context.Context, int) (*pluginv1.WatchSyncProviderConfig, error) {
+			return &pluginv1.WatchSyncProviderConfig{Values: map[string]string{"floppy.base_url": "https://legacy.example.com"}}, nil
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _, err = provider.ConnectWithAPIKeyConfig(context.Background(), "token", ConnectionConfigValues{
+		"floppy": {"base_url": "https://personal.example.com"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := client.exchangeRequest.GetProviderConfig().GetValues()["floppy.base_url"]; got != "https://personal.example.com" {
+		t.Fatalf("base URL = %q", got)
+	}
+	views := provider.ConnectionConfigSchema()
+	if len(views) != 1 || views[0].AdminForm == nil || views[0].AdminForm.Fields[0].Control != "TEXT" {
+		t.Fatalf("connection config schema = %#v", views)
+	}
+	if _, _, err := provider.ConnectWithAPIKeyConfig(context.Background(), "token", nil); err == nil || !strings.Contains(err.Error(), "required") {
+		t.Fatalf("missing config error = %v", err)
+	}
+}
+
+func TestPluginProviderClassifiesAndRedactsConnectionSecrets(t *testing.T) {
+	client := &fakeWatchSyncPluginClient{exchangeResponse: &pluginv1.WatchSyncCredentialResponse{
+		Fault: &pluginv1.WatchSyncFault{
+			Code:        pluginv1.WatchSyncFaultCode_WATCH_SYNC_FAULT_CODE_INVALID_CREDENTIAL,
+			SafeMessage: "credentials " + testSecretValue + " were rejected",
+		},
+	}}
+	schema := &pluginv1.ConfigSchema{
+		Key: "account",
+		JsonSchema: `{
+			"type":"object",
+			"properties":{
+				"base_url":{"type":"string","format":"uri"},
+				"client_secret":{"type":"string","format":"password"}
+			},
+			"required":["base_url","client_secret"],
+			"additionalProperties":false
+		}`,
+		AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{
+			{Key: "base_url", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT},
+			// JSON Schema remains authoritative even when a form incorrectly
+			// presents a credential as ordinary text.
+			{Key: "client_secret", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT},
+		}},
+	}
+	provider, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{schema},
+		ResolveClient:          func(context.Context, int, string) (WatchSyncPluginClient, error) { return client, nil },
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _, err = provider.ConnectWithAPIKeyConfig(context.Background(), "input-token", ConnectionConfigValues{
+		"account": {
+			"base_url":      "https://floppy.example.com",
+			"client_secret": testSecretValue,
+		},
+	})
+	if !isWatchSyncInvalidCredentialError(err) {
+		t.Fatalf("error = %#v", err)
+	}
+	config := client.exchangeRequest.GetProviderConfig()
+	if config.GetValues()["account.base_url"] != "https://floppy.example.com" ||
+		config.GetSecretValues()["account.client_secret"] != testSecretValue {
+		t.Fatalf("provider config = %#v", config)
+	}
+	if _, exposed := config.GetValues()["account.client_secret"]; exposed {
+		t.Fatal("JSON-schema password was exposed as a public provider value")
+	}
+	if strings.Contains(err.Error(), testSecretValue) || !strings.Contains(err.Error(), "[REDACTED]") {
+		t.Fatalf("connection secrets were not redacted: %q", err)
+	}
+}
+
+func TestConnectionConfigValidationRedactsNestedSecrets(t *testing.T) {
+	const nestedSecret = "nested-connection-secret"
+	schema := &pluginv1.ConfigSchema{
+		Key:        "account",
+		JsonSchema: `{"type":"object","properties":{"advanced":{"type":"object","properties":{"password":{"type":"string","format":"password"}}}}}`,
+	}
+	secrets := connectionConfigSecrets(
+		[]*pluginv1.ConfigSchema{schema},
+		ConnectionConfigValues{"account": {"advanced": map[string]any{"password": nestedSecret}}},
+	)
+	err := sanitizedConnectionConfigError(errors.New("rejected "+nestedSecret), secrets)
+	if strings.Contains(err.Error(), nestedSecret) || !strings.Contains(err.Error(), "[REDACTED]") {
+		t.Fatalf("nested connection secret was not redacted: %q", err)
+	}
+}
+
+func TestPluginProviderRejectsUnresolvableDynamicConnectionOptions(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key: "server",
+			AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+				Key: "library", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SELECT,
+				DynamicOptions: true,
+			}}},
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "requires dynamic options") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
+func TestPluginProviderRejectsBlankConnectionSelectOption(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key:        "server",
+			JsonSchema: `{"type":"object","properties":{"mode":{"type":"string"}}}`,
+			AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+				Key: "mode", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SELECT,
+				Options: []*pluginv1.AdminFormOption{{Value: " ", Label: "Choose a mode"}},
+			}}},
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "blank select option value") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
+func TestPluginProviderRejectsDuplicateConnectionConfigKeys(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{
+			{Key: "server", JsonSchema: `{"type":"object","properties":{}}`},
+			{Key: "server", JsonSchema: `{"type":"object","properties":{}}`},
+		},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "duplicated") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
+func TestPluginProviderRejectsBlankConnectionConfigKey(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key: " \t", Required: true, JsonSchema: `{"type":"object","properties":{}}`,
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "key is required") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
+func TestPluginProviderRejectsUnsupportedConnectionConfigExclusivity(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key: "server", JsonSchema: `{"type":"object","properties":{"primary":{"type":"boolean"},"group":{"type":"string"}}}`,
+			AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+				Key: "primary", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SWITCH,
+				ExclusiveGroupField: "group",
+			}}},
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "exclusive_group_field") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
+func TestPluginProviderRejectsRequiredConnectionConfigTheWebCannotRender(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key: "server", Required: true,
+			JsonSchema: `{"type":"object","properties":{"headers":{"type":"object"}}}`,
+			AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+				Key: "headers", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXTAREA,
+			}}},
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "cannot be inferred") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
+func TestPluginProviderRejectsOptionalConnectionConfigTheWebCannotRender(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key:        "headers",
+			JsonSchema: `{"type":"object","properties":{"values":{"type":"object"}}}`,
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "cannot be inferred") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
+func TestPluginProviderAcceptsRenderableRequiredConnectionConfig(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{
+			{
+				Key: "server", Required: true,
+				JsonSchema: `{"type":"object","properties":{"base_url":{"type":"string"},"username":{"type":"string"}},"required":["base_url","username"]}`,
+				AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+					Key: "base_url", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT,
+				}}},
+			},
+			{
+				Key: "features", Required: true,
+				JsonSchema: `{"type":"object","properties":{"flags":{"type":"array","items":{"type":"boolean"}}}}`,
+				AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+					Key: "flags", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_MULTI_SELECT,
+					Options: []*pluginv1.AdminFormOption{{Value: "true", Label: "Enabled"}, {Value: "false", Label: "Disabled"}},
+				}}},
+			},
+			{
+				Key: "mode", Required: true,
+				JsonSchema: `{"type":"object","properties":{"value":{"enum":["standard","anime"]}}}`,
+				AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+					Key: "value", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SELECT,
+					Options: []*pluginv1.AdminFormOption{{Value: "standard", Label: "Standard"}, {Value: "anime", Label: "Anime"}},
+				}}},
+			},
+			{
+				Key: "reference", Required: true,
+				JsonSchema: `{"type":"object","properties":{"endpoint":{"$ref":"#/$defs/endpoint"}},"$defs":{"endpoint":{"type":"string","format":"uri"}}}`,
+				AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+					Key: "endpoint", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT,
+				}}},
+			},
+		},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestPluginProviderRejectsConnectionControlThatCannotEmitSchemaType(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key:        "server",
+			JsonSchema: `{"type":"object","properties":{"name":{"type":"string"}}}`,
+			AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{{
+				Key: "name", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SWITCH,
+			}}},
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "cannot emit json_schema type") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
+func TestPluginProviderEnforcesVisibleRequiredConnectionAdminFields(t *testing.T) {
+	client := &fakeWatchSyncPluginClient{exchangeResponse: &pluginv1.WatchSyncCredentialResponse{
+		Credentials: &pluginv1.WatchSyncCredentials{AccessToken: testValidatedToken},
+		Account:     &pluginv1.WatchSyncAccount{ExternalSubject: "7"},
+	}}
+	provider, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key:        "server",
+			JsonSchema: `{"type":"object","properties":{"advanced":{"type":"boolean"},"endpoint":{"type":"string"}},"required":["advanced"],"additionalProperties":false}`,
+			AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{
+				{Key: "advanced", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_SWITCH},
+				{
+					Key: "endpoint", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT, Required: true,
+					ShowWhen: []*pluginv1.AdminFormCondition{{Field: "advanced", Equals: []string{"true"}}},
+				},
+			}},
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return client, nil
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, _, err := provider.ConnectWithAPIKeyConfig(context.Background(), "token", ConnectionConfigValues{
+		"server": {"advanced": false},
+	}); err != nil {
+		t.Fatalf("hidden required field: %v", err)
+	}
+	if _, _, err := provider.ConnectWithAPIKeyConfig(context.Background(), "token", ConnectionConfigValues{
+		"server": {"advanced": true},
+	}); err == nil || !strings.Contains(err.Error(), `field "endpoint" is required`) {
+		t.Fatalf("visible required field error = %v", err)
+	}
+}
+
+func TestPluginProviderRejectsFlattenedConnectionConfigCollision(t *testing.T) {
+	client := &fakeWatchSyncPluginClient{}
+	provider, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{
+			{Key: "a", JsonSchema: `{"type":"object","properties":{"b.c":{"type":"string"}},"additionalProperties":false}`},
+			{Key: "a.b", JsonSchema: `{"type":"object","properties":{"c":{"type":"string"}},"additionalProperties":false}`},
+		},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return client, nil
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, _, err = provider.ConnectWithAPIKeyConfig(context.Background(), "token", ConnectionConfigValues{
+		"a":   {"b.c": "first"},
+		"a.b": {"c": "second"},
+	})
+	if err == nil || !strings.Contains(err.Error(), "conflicts") || !strings.Contains(err.Error(), "after flattening") {
+		t.Fatalf("error = %v", err)
+	}
+	if client.exchangeRequest != nil {
+		t.Fatal("ambiguous connection config reached the plugin")
+	}
+}
+
+func TestPluginProviderEnforcesConnectionAdminFormValidation(t *testing.T) {
+	client := &fakeWatchSyncPluginClient{exchangeResponse: &pluginv1.WatchSyncCredentialResponse{
+		Credentials: &pluginv1.WatchSyncCredentials{AccessToken: testValidatedToken},
+		Account:     &pluginv1.WatchSyncAccount{ExternalSubject: "7"},
+	}}
+	provider, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_API_KEY,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{
+			Key: "server", Required: true,
+			JsonSchema: `{"type":"object","properties":{"name":{"type":"string"},"port":{},"password":{"type":"string","format":"password"}},"required":["name","port","password"]}`,
+			AdminForm: &pluginv1.AdminFormDescriptor{Fields: []*pluginv1.AdminFormField{
+				{Key: "name", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_TEXT, Validation: &pluginv1.AdminFormValidation{Pattern: `^[a-z]+$`, MinLength: 3, MaxLength: 8}},
+				{Key: "port", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_NUMBER, Validation: &pluginv1.AdminFormValidation{HasMin: true, Min: 1, HasMax: true, Max: 65535}},
+				{Key: "password", Control: pluginv1.AdminFormControl_ADMIN_FORM_CONTROL_PASSWORD, Secret: true, Validation: &pluginv1.AdminFormValidation{MinLength: 8}},
+			}},
+		}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return client, nil
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tests := []struct {
+		name    string
+		values  map[string]any
+		message string
+	}{
+		{name: "pattern", values: map[string]any{"name": "Bad", "port": 443.0, "password": "long-enough"}, message: "is invalid"},
+		{name: "number", values: map[string]any{"name": "good", "port": 70000.0, "password": "long-enough"}, message: "at most 65535"},
+		{name: "numeric string", values: map[string]any{"name": "good", "port": "70000", "password": "long-enough"}, message: "at most 65535"},
+		{name: "secret length", values: map[string]any{"name": "good", "port": 443.0, "password": "leaky"}, message: "at least 8 characters"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, _, err := provider.ConnectWithAPIKeyConfig(context.Background(), "token", ConnectionConfigValues{"server": tt.values})
+			if err == nil || !strings.Contains(err.Error(), tt.message) {
+				t.Fatalf("error = %v", err)
+			}
+			if strings.Contains(err.Error(), "leaky") {
+				t.Fatalf("secret leaked in validation error: %q", err)
+			}
+		})
+	}
+}
+
+func TestPluginProviderRejectsConnectionConfigForDeviceAuthorization(t *testing.T) {
+	_, err := NewPluginProvider(PluginProviderOptions{
+		InstallationID: 4, ProviderKey: testPluginProviderKey, CapabilityID: testPluginCapabilityID,
+		Descriptor: &pluginv1.WatchSyncProviderDescriptor{AuthMethods: []pluginv1.WatchSyncAuthMethod{
+			pluginv1.WatchSyncAuthMethod_WATCH_SYNC_AUTH_METHOD_DEVICE_CODE,
+		}},
+		ConnectionConfigSchema: []*pluginv1.ConfigSchema{{Key: "server"}},
+		ResolveClient: func(context.Context, int, string) (WatchSyncPluginClient, error) {
+			return &fakeWatchSyncPluginClient{}, nil
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "connection config requires API-key authentication") {
+		t.Fatalf("error = %v", err)
+	}
+}
+
 func TestPluginProviderRefreshReturnsCredentialsAlongsideFault(t *testing.T) {
 	client := &fakeWatchSyncPluginClient{refreshResponse: &pluginv1.WatchSyncCredentialResponse{
 		Credentials: &pluginv1.WatchSyncCredentials{AccessToken: testRotatedAccessToken, TokenType: testBearerTokenType},
@@ -867,3 +1339,29 @@ func TestPluginProviderForwardsLiveScrobbleLifecycle(t *testing.T) {
 		t.Fatalf("scrobble event = %#v", got)
 	}
 }
+
+func TestPluginProviderForwardsAuthoritativeScrobbleCompletion(t *testing.T) {
+	completed := watchEventFromScrobble(ScrobbleEvent{
+		PlaybackSessionID: testPlaybackSessionID,
+		MediaItemID:       testMovieMediaID,
+		Kind:              historyimport.KindMovie,
+		PositionSeconds:   90,
+		DurationSeconds:   100,
+		Completed:         true,
+	}, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_STOP)
+	if !completed.GetCompleted() {
+		t.Fatal("completed event = false, want true")
+	}
+
+	incomplete := watchEventFromScrobble(ScrobbleEvent{
+		PlaybackSessionID: testPlaybackSessionID,
+		MediaItemID:       testMovieMediaID,
+		Kind:              historyimport.KindMovie,
+		PositionSeconds:   10,
+		DurationSeconds:   100,
+		Completed:         false,
+	}, pluginv1.WatchSyncOperation_WATCH_SYNC_OPERATION_SCROBBLE_STOP)
+	if incomplete.GetCompleted() {
+		t.Fatal("incomplete event = true, want false")
+	}
+}

internal/watchsync/registry.go

@@ -91,11 +91,15 @@ func (r *Registry) List() []ProviderSummary {
 
 	summaries := make([]ProviderSummary, 0, len(r.providers))
 	for key, provider := range r.providers {
-		summaries = append(summaries, ProviderSummary{
+		summary := ProviderSummary{
 			Key:          key,
 			DisplayName:  provider.DisplayName(),
 			Capabilities: provider.Capabilities(),
-		})
+		}
+		if configurable, ok := provider.(connectionConfigProvider); ok {
+			summary.ConnectionConfigSchema = configurable.ConnectionConfigSchema()
+		}
+		summaries = append(summaries, summary)
 	}
 	sort.Slice(summaries, func(i, j int) bool {
 		return summaries[i].Key < summaries[j].Key

internal/watchsync/repository.go

@@ -34,6 +34,7 @@ type Repository interface {
 	ListListEventConnections(ctx context.Context, userID int, profileID string, list ListKind) ([]Connection, error)
 	UpsertHistoryExports(ctx context.Context, exports []HistoryExport) error
 	ListPendingHistoryExports(ctx context.Context, connectionID string, limit int) ([]HistoryExport, error)
+	ListPendingHistoryExportsByHistoryIDs(ctx context.Context, connectionID string, historyIDs []string, limit int) ([]HistoryExport, error)
 	MarkHistoryExportStatus(ctx context.Context, id string, status string, lastError string) error
 	MarkHistoryExportSatisfiedByScrobble(ctx context.Context, connectionID string, historyID string) error
 	UpsertListItemStates(ctx context.Context, states []ListItemState) error
@@ -765,6 +766,60 @@ func (r *PostgresRepository) ListPendingHistoryExports(ctx context.Context, conn
 	return exports, nil
 }
 
+func (r *PostgresRepository) ListPendingHistoryExportsByHistoryIDs(
+	ctx context.Context,
+	connectionID string,
+	historyIDs []string,
+	limit int,
+) ([]HistoryExport, error) {
+	if len(historyIDs) == 0 {
+		return nil, nil
+	}
+	if limit <= 0 || limit > len(historyIDs) {
+		limit = len(historyIDs)
+	}
+	rows, err := r.pool.Query(ctx, `
+		SELECT id::text, connection_id::text, history_id, media_item_id, watched_at,
+			provider_item_key, status, attempt_count, last_attempt_at, last_error, created_at, updated_at
+		FROM watch_provider_history_exports
+		WHERE connection_id = $1::uuid
+		  AND history_id = ANY($2::text[])
+		  AND status IN ('pending', 'failed')
+		  AND attempt_count < 5
+		ORDER BY watched_at ASC
+		LIMIT $3
+	`, connectionID, historyIDs, limit)
+	if err != nil {
+		return nil, fmt.Errorf("list pending history exports by history ids: %w", err)
+	}
+	defer rows.Close()
+	var exports []HistoryExport
+	for rows.Next() {
+		var export HistoryExport
+		if err := rows.Scan(
+			&export.ID,
+			&export.ConnectionID,
+			&export.HistoryID,
+			&export.MediaItemID,
+			&export.WatchedAt,
+			&export.ProviderItemKey,
+			&export.Status,
+			&export.AttemptCount,
+			&export.LastAttemptAt,
+			&export.LastError,
+			&export.CreatedAt,
+			&export.UpdatedAt,
+		); err != nil {
+			return nil, fmt.Errorf("scan pending history export by history ids: %w", err)
+		}
+		exports = append(exports, export)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate pending history exports by history ids: %w", err)
+	}
+	return exports, nil
+}
+
 func (r *PostgresRepository) MarkHistoryExportStatus(ctx context.Context, id string, status string, lastError string) error {
 	_, err := r.pool.Exec(ctx, `
 		UPDATE watch_provider_history_exports

internal/watchsync/service.go

@@ -49,11 +49,13 @@ type watchStateImporter interface {
 const (
 	manualSyncCooldown = time.Hour
 	manualSyncTimeout  = 10 * time.Minute
-	// Built-in providers bind requests to this context and use HTTP client
+	// A completed stop may require a cold metadata lookup in a provider plugin.
-	// timeouts of at most 20 seconds. Keep dispatch below the reclaim lease;
+	// Match the plugin-host watch-sync deadline so the durable confirmation path
-	// the per-session queue remains occupied until the worker itself exits.
+	// does not cancel valid provider work before the RPC can finish. The reclaim
-	confirmedStopDispatchTimeout = 25 * time.Second
+	// lease must remain longer than that entire dispatch and finalization window
-	confirmedStopLease           = time.Minute
+	// so a second request cannot take over while the first is still valid.
+	confirmedStopDispatchTimeout = 2 * time.Minute
+	confirmedStopLease           = confirmedStopDispatchTimeout + 30*time.Second
 )
 
 var errConfirmedStopInProgress = errors.New("watch provider stop confirmation already in progress")
@@ -134,6 +136,9 @@ func (s *Service) GetConnectionStatus(ctx context.Context, userID int, profileID
 		SyncWatchlistOrderEnabled:    true,
 		ScrobbleEnabled:              true,
 	}
+	if configurable, ok := provider.(connectionConfigProvider); ok {
+		status.ConnectionConfigSchema = configurable.ConnectionConfigSchema()
+	}
 	if connected {
 		status.ProviderUsername = conn.ProviderUsername
 		status.ImportWatchedEnabled = conn.ImportWatchedEnabled
@@ -542,6 +547,17 @@ func (s *Service) ConnectAPIKey(
 	profileID string,
 	providerKey string,
 	apiKey string,
+) (Connection, error) {
+	return s.ConnectAPIKeyWithConfig(ctx, userID, profileID, providerKey, apiKey, nil)
+}
+
+func (s *Service) ConnectAPIKeyWithConfig(
+	ctx context.Context,
+	userID int,
+	profileID string,
+	providerKey string,
+	apiKey string,
+	connectionConfig ConnectionConfigValues,
 ) (Connection, error) {
 	if userID <= 0 {
 		return Connection{}, fmt.Errorf("user id is required")
@@ -562,7 +578,17 @@ func (s *Service) ConnectAPIKey(
 		return Connection{}, fmt.Errorf("provider %q does not support api-key auth", providerKey)
 	}
 
-	tokens, account, err := authProvider.ConnectWithAPIKey(ctx, apiKey)
+	var tokens TokenSet
+	var account ProviderAccount
+	var err error
+	if configured, ok := provider.(configuredAPIKeyAuthProvider); ok {
+		tokens, account, err = configured.ConnectWithAPIKeyConfig(ctx, apiKey, connectionConfig)
+	} else {
+		if len(connectionConfig) > 0 {
+			return Connection{}, fmt.Errorf("provider %q does not accept connection configuration", providerKey)
+		}
+		tokens, account, err = authProvider.ConnectWithAPIKey(ctx, apiKey)
+	}
 	if err != nil {
 		return Connection{}, err
 	}
@@ -1510,7 +1536,19 @@ func (s *Service) exportLocalPlays(
 	if err := s.repo.UpsertHistoryExports(ctx, exports); err != nil {
 		return err
 	}
-	pending, err := s.repo.ListPendingHistoryExports(ctx, conn.ID, 100)
+	historyIDs := make([]string, 0, len(exports))
+	for _, export := range exports {
+		historyIDs = append(historyIDs, export.HistoryID)
+	}
+	// A live watch event must not wait behind a connection's historical
+	// backlog. Scheduled sync still drains that backlog oldest-first, while
+	// this path selects only the events the user just created.
+	pending, err := s.repo.ListPendingHistoryExportsByHistoryIDs(
+		ctx,
+		conn.ID,
+		historyIDs,
+		watchedExportBatchSize(exporter, len(historyIDs)),
+	)
 	if err != nil {
 		return err
 	}
@@ -1603,20 +1641,25 @@ func (s *Service) persistConnectionError(ctx context.Context, conn Connection, m
 }
 
 func limitWatchedExportBatch(exporter WatchedExporter, plays []LocalPlay) ([]LocalPlay, bool) {
-	bounded, ok := exporter.(singleBatchWatchedExporter)
+	_, ok := exporter.(singleBatchWatchedExporter)
 	if !ok {
 		return plays, false
 	}
-	limit := bounded.ExportBatchSize()
+	limit := watchedExportBatchSize(exporter, len(plays))
-	if limit <= 0 {
-		limit = 1
-	}
 	if len(plays) > limit {
 		plays = plays[:limit]
 	}
 	return plays, true
 }
 
+func watchedExportBatchSize(exporter WatchedExporter, fallback int) int {
+	bounded, ok := exporter.(singleBatchWatchedExporter)
+	if !ok {
+		return max(1, fallback)
+	}
+	return max(1, bounded.ExportBatchSize())
+}
+
 func reconcileHistoryExports(connectionID string, local []LocalPlay, remote []RemotePlay) []HistoryExport {
 	remoteExact := make(map[string]struct{}, len(remote))
 	for _, play := range remote {

internal/watchsync/service_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"fmt"
 	"strconv"
 	"strings"
 	"sync"
@@ -45,6 +46,8 @@ type serviceFakeRepo struct {
 	settings               map[string]string
 	syncRuns               []SyncRun
 	historyExports         []HistoryExport
+	historyLookupIDs       []string
+	historyLookupLimit     int
 	listItemStates         []ListItemState
 	scrobbleConnections    []Connection
 	scrobbleSessions       []ScrobbleSession
@@ -346,6 +349,27 @@ func (r *serviceFakeRepo) ListPendingHistoryExports(_ context.Context, connectio
 	return exports, nil
 }
 
+func (r *serviceFakeRepo) ListPendingHistoryExportsByHistoryIDs(_ context.Context, connectionID string, historyIDs []string, limit int) ([]HistoryExport, error) {
+	r.historyLookupIDs = append([]string(nil), historyIDs...)
+	r.historyLookupLimit = limit
+	wanted := make(map[string]struct{}, len(historyIDs))
+	for _, historyID := range historyIDs {
+		wanted[historyID] = struct{}{}
+	}
+	var exports []HistoryExport
+	for _, export := range r.historyExports {
+		_, matches := wanted[export.HistoryID]
+		if export.ConnectionID == connectionID && matches &&
+			(export.Status == historyExportStatusPending || export.Status == historyExportStatusFailed) && export.AttemptCount < 5 {
+			exports = append(exports, export)
+			if limit > 0 && len(exports) >= limit {
+				break
+			}
+		}
+	}
+	return exports, nil
+}
+
 func (r *serviceFakeRepo) MarkHistoryExportStatus(_ context.Context, id string, status string, lastError string) error {
 	if r.markHistoryStatusErr != nil {
 		return r.markHistoryStatusErr
@@ -789,10 +813,20 @@ func (p progressBatchImporterStub) FetchProgressBatch(context.Context, ServerCon
 type watchedExporterStub struct {
 	exportErr    error
 	exportResult ExportResult
+	exported     *[]LocalPlay
 	key          string
 	source       userstore.WatchHistorySource
 }
 
+type singleBatchWatchedExporterStub struct {
+	watchedExporterStub
+	batchSize int
+}
+
+func (p singleBatchWatchedExporterStub) ExportBatchSize() int {
+	return p.batchSize
+}
+
 func (p watchedExporterStub) Key() string {
 	if p.key != "" {
 		return p.key
@@ -812,7 +846,10 @@ func (p watchedExporterStub) FetchHistory(context.Context, ServerConfig, Connect
 	return nil, nil
 }
 
-func (p watchedExporterStub) ExportHistory(context.Context, ServerConfig, Connection, []LocalPlay) (ExportResult, error) {
+func (p watchedExporterStub) ExportHistory(_ context.Context, _ ServerConfig, _ Connection, plays []LocalPlay) (ExportResult, error) {
+	if p.exported != nil {
+		*p.exported = append(*p.exported, plays...)
+	}
 	return p.exportResult, p.exportErr
 }
 
@@ -2587,6 +2624,12 @@ func TestServiceConfirmedStopSerializesConcurrentConfirmation(t *testing.T) {
 	}
 }
 
+func TestConfirmedStopLeaseExceedsDispatchTimeout(t *testing.T) {
+	if confirmedStopLease <= confirmedStopDispatchTimeout {
+		t.Fatalf("confirmed stop lease %s must exceed dispatch timeout %s", confirmedStopLease, confirmedStopDispatchTimeout)
+	}
+}
+
 func TestServiceConfirmedStopCannotCompleteReclaimedLease(t *testing.T) {
 	repo := newServiceFakeRepo()
 	repo.scrobbleConnections = []Connection{{
@@ -3130,6 +3173,74 @@ func TestServicePluginTransportFailureLeavesExportPending(t *testing.T) {
 	}
 }
 
+func TestServiceLocalWatchEventBypassesHistoricalExportBacklog(t *testing.T) {
+	repo := newServiceFakeRepo()
+	for i := 0; i < 100; i++ {
+		repo.historyExports = append(repo.historyExports, HistoryExport{
+			ID:           fmt.Sprintf("old-export-%d", i),
+			ConnectionID: "conn-1",
+			HistoryID:    fmt.Sprintf("old-history-%d", i),
+			Status:       historyExportStatusPending,
+			WatchedAt:    time.Date(2025, time.January, 1, 0, i, 0, 0, time.UTC),
+		})
+	}
+	var exported []LocalPlay
+	service := NewService(repo, NewRegistry())
+	play := LocalPlay{
+		HistoryID:       "new-history",
+		MediaItemID:     testMovieMediaID,
+		ProviderItemKey: testMovieProviderItemKey,
+		WatchedAt:       time.Date(2026, time.August, 6, 12, 0, 0, 0, time.UTC),
+	}
+	err := service.exportLocalPlays(context.Background(), Connection{ID: "conn-1"}, ServerConfig{}, watchedExporterStub{
+		exported:     &exported,
+		exportResult: ExportResult{Sent: []string{play.HistoryID}},
+	}, []LocalPlay{play})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(exported) != 1 || exported[0].HistoryID != play.HistoryID {
+		t.Fatalf("exported = %#v, want only the live event", exported)
+	}
+	if got := repo.historyExports[len(repo.historyExports)-1].Status; got != historyExportStatusSent {
+		t.Fatalf("new export status = %q, want %q", got, historyExportStatusSent)
+	}
+}
+
+func TestServiceLocalWatchEventBoundsHistoryLookupToProviderBatch(t *testing.T) {
+	repo := newServiceFakeRepo()
+	var exported []LocalPlay
+	plays := make([]LocalPlay, 0, 25)
+	for i := 0; i < 25; i++ {
+		plays = append(plays, LocalPlay{
+			HistoryID:       fmt.Sprintf("history-%d", i),
+			MediaItemID:     testMovieMediaID,
+			ProviderItemKey: testMovieProviderItemKey,
+			WatchedAt:       time.Date(2026, time.August, 6, 12, i, 0, 0, time.UTC),
+		})
+	}
+	service := NewService(repo, NewRegistry())
+	err := service.exportLocalPlays(
+		context.Background(),
+		Connection{ID: "conn-1"},
+		ServerConfig{},
+		singleBatchWatchedExporterStub{
+			watchedExporterStub: watchedExporterStub{exported: &exported},
+			batchSize:           1,
+		},
+		plays,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if repo.historyLookupLimit != 1 {
+		t.Fatalf("history lookup limit = %d, want 1", repo.historyLookupLimit)
+	}
+	if len(exported) != 1 {
+		t.Fatalf("exported %d plays, want 1", len(exported))
+	}
+}
+
 func TestServicePluginInvalidCredentialLeavesExportPendingAndRecordsConnectionError(t *testing.T) {
 	repo := newServiceFakeRepo()
 	client := &fakeWatchSyncPluginClient{applyResponse: &pluginv1.WatchSyncApplyEventsResponse{

internal/watchsync/types.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/Silo-Server/silo-server/internal/historyimport"
+	"github.com/Silo-Server/silo-server/internal/plugins"
 	"github.com/Silo-Server/silo-server/internal/userstore"
 )
 
@@ -74,6 +75,18 @@ type APIKeyAuthProvider interface {
 	ConnectWithAPIKey(ctx context.Context, apiKey string) (TokenSet, ProviderAccount, error)
 }
 
+// ConnectionConfigValues contains manifest-declared, per-connection setup
+// values keyed first by config-schema key and then by field key.
+type ConnectionConfigValues map[string]map[string]any
+
+type configuredAPIKeyAuthProvider interface {
+	ConnectWithAPIKeyConfig(ctx context.Context, apiKey string, config ConnectionConfigValues) (TokenSet, ProviderAccount, error)
+}
+
+type connectionConfigProvider interface {
+	ConnectionConfigSchema() []plugins.ConfigSchemaView
+}
+
 type WatchedImporter interface {
 	FetchWatched(ctx context.Context, cfg ServerConfig, conn Connection) ([]RemoteWatch, error)
 }
@@ -660,38 +673,40 @@ func (f RemoteFavorite) HistoryRecord() historyimport.Record {
 }
 
 type ProviderSummary struct {
-	Key          string       `json:"key"`
+	Key                    string                     `json:"key"`
-	DisplayName  string       `json:"display_name"`
+	DisplayName            string                     `json:"display_name"`
-	Capabilities Capabilities `json:"capabilities"`
+	Capabilities           Capabilities               `json:"capabilities"`
+	ConnectionConfigSchema []plugins.ConfigSchemaView `json:"connection_config_schema,omitempty"`
 }
 
 type ConnectionStatus struct {
-	Provider                     string       `json:"provider"`
+	Provider                     string                     `json:"provider"`
-	DisplayName                  string       `json:"display_name"`
+	DisplayName                  string                     `json:"display_name"`
-	Capabilities                 Capabilities `json:"capabilities"`
+	Capabilities                 Capabilities               `json:"capabilities"`
-	AuthMethod                   string       `json:"auth_method"`
+	AuthMethod                   string                     `json:"auth_method"`
-	Connected                    bool         `json:"connected"`
+	Connected                    bool                       `json:"connected"`
-	ProviderUsername             string       `json:"provider_username,omitempty"`
+	ProviderUsername             string                     `json:"provider_username,omitempty"`
-	ImportWatchedEnabled         bool         `json:"import_watched_enabled"`
+	ImportWatchedEnabled         bool                       `json:"import_watched_enabled"`
-	ImportProgressEnabled        bool         `json:"import_progress_enabled"`
+	ImportProgressEnabled        bool                       `json:"import_progress_enabled"`
-	ExportWatchedEnabled         bool         `json:"export_watched_enabled"`
+	ExportWatchedEnabled         bool                       `json:"export_watched_enabled"`
-	ExportUnwatchedEnabled       bool         `json:"export_unwatched_enabled"`
+	ExportUnwatchedEnabled       bool                       `json:"export_unwatched_enabled"`
-	ImportFavoritesEnabled       bool         `json:"import_favorites_enabled"`
+	ImportFavoritesEnabled       bool                       `json:"import_favorites_enabled"`
-	ExportFavoritesEnabled       bool         `json:"export_favorites_enabled"`
+	ExportFavoritesEnabled       bool                       `json:"export_favorites_enabled"`
-	SyncFavoriteRemovalsEnabled  bool         `json:"sync_favorite_removals_enabled"`
+	SyncFavoriteRemovalsEnabled  bool                       `json:"sync_favorite_removals_enabled"`
-	ImportWatchlistEnabled       bool         `json:"import_watchlist_enabled"`
+	ImportWatchlistEnabled       bool                       `json:"import_watchlist_enabled"`
-	ExportWatchlistEnabled       bool         `json:"export_watchlist_enabled"`
+	ExportWatchlistEnabled       bool                       `json:"export_watchlist_enabled"`
-	SyncWatchlistRemovalsEnabled bool         `json:"sync_watchlist_removals_enabled"`
+	SyncWatchlistRemovalsEnabled bool                       `json:"sync_watchlist_removals_enabled"`
-	SyncWatchlistOrderEnabled    bool         `json:"sync_watchlist_order_enabled"`
+	SyncWatchlistOrderEnabled    bool                       `json:"sync_watchlist_order_enabled"`
-	ScrobbleEnabled              bool         `json:"scrobble_enabled"`
+	ScrobbleEnabled              bool                       `json:"scrobble_enabled"`
-	CredentialsConfigured        bool         `json:"credentials_configured"`
+	CredentialsConfigured        bool                       `json:"credentials_configured"`
-	LastInboundSyncAt            *time.Time   `json:"last_inbound_sync_at,omitempty"`
+	ConnectionConfigSchema       []plugins.ConfigSchemaView `json:"connection_config_schema,omitempty"`
-	LastProgressSyncAt           *time.Time   `json:"last_progress_sync_at,omitempty"`
+	LastInboundSyncAt            *time.Time                 `json:"last_inbound_sync_at,omitempty"`
-	LastOutboundSyncAt           *time.Time   `json:"last_outbound_sync_at,omitempty"`
+	LastProgressSyncAt           *time.Time                 `json:"last_progress_sync_at,omitempty"`
-	LastFavoritesSyncAt          *time.Time   `json:"last_favorites_sync_at,omitempty"`
+	LastOutboundSyncAt           *time.Time                 `json:"last_outbound_sync_at,omitempty"`
-	LastWatchlistSyncAt          *time.Time   `json:"last_watchlist_sync_at,omitempty"`
+	LastFavoritesSyncAt          *time.Time                 `json:"last_favorites_sync_at,omitempty"`
-	LastScrobbleErrorAt          *time.Time   `json:"last_scrobble_error_at,omitempty"`
+	LastWatchlistSyncAt          *time.Time                 `json:"last_watchlist_sync_at,omitempty"`
-	LastError                    string       `json:"last_error,omitempty"`
+	LastScrobbleErrorAt          *time.Time                 `json:"last_scrobble_error_at,omitempty"`
+	LastError                    string                     `json:"last_error,omitempty"`
 }
 
 type ConnectionUpdate struct {

web/src/components/admin/plugins/PluginConfigForm.test.tsx

@@ -36,6 +36,30 @@ const schema: PluginConfigSchema = {
 };
 
 describe("PluginConfigForm secrets", () => {
+  it("derives a form when a plugin only supplies JSON Schema", () => {
+    render(
+      <PluginConfigForm
+        schema={{
+          key: "server",
+          title: "Server",
+          json_schema: JSON.stringify({
+            type: "object",
+            properties: {
+              base_url: { type: "string", title: "Base URL" },
+              api_key: { type: "string", format: "password" },
+            },
+            required: ["base_url", "api_key"],
+          }),
+          required: true,
+        }}
+        onSave={vi.fn()}
+      />,
+    );
+
+    expect(screen.getByLabelText("Base URL")).toBeInTheDocument();
+    expect(screen.getByLabelText("Api Key")).toHaveAttribute("type", "password");
+  });
+
   it("shows redacted saved state and only clears through an explicit action", async () => {
     const onSave = vi.fn();
     render(

web/src/components/admin/plugins/PluginConfigForm.tsx

@@ -10,11 +10,14 @@ import { ConnectionCheckAction } from "@/components/admin/ConnectionCheckAction"
 import { Button } from "@/components/ui/button";
 import { Label } from "@/components/ui/label";
 
+import { adminFormForConfigSchema, humanizeConfigKey } from "./configSchemaAdminForm";
 import { SchemaForm } from "./SchemaForm";
 import { buildSchemaValues } from "./schemaFormUtils";
 
 type PluginConfigValue = Record<string, unknown>;
 
+const EMPTY_FIELDS: PluginAdminFormField[] = [];
+
 type Props = {
   schema: PluginConfigSchema;
   value?: PluginConfigValue;
@@ -29,82 +32,7 @@ type Props = {
   isTesting?: boolean;
 };
 
-type SupportedField = PluginAdminFormField & {
+function defaultValueForField(field: PluginAdminFormField): string | boolean {
-  inferredType?: "string" | "number" | "integer" | "boolean";
-};
-
-type ParsedObjectSchema = {
-  supported: boolean;
-  fields: SupportedField[];
-};
-
-function humanizeKey(value: string) {
-  return value
-    .split("_")
-    .filter(Boolean)
-    .map((part) => part.charAt(0).toUpperCase() + part.slice(1))
-    .join(" ");
-}
-
-function parseJSONSchema(schema: PluginConfigSchema): ParsedObjectSchema {
-  try {
-    const parsed = JSON.parse(schema.json_schema) as {
-      type?: string;
-      required?: string[];
-      properties?: Record<
-        string,
-        {
-          type?: string;
-          title?: string;
-          description?: string;
-          writeOnly?: boolean;
-          format?: string;
-        }
-      >;
-    };
-    if (parsed.type !== "object" || !parsed.properties) {
-      return { supported: false, fields: [] };
-    }
-
-    const fields = Object.entries(parsed.properties).map(([key, property]) => {
-      const propertyType = property.type;
-      if (!propertyType || !["string", "number", "integer", "boolean"].includes(propertyType)) {
-        return null;
-      }
-      const isSensitive = property.writeOnly === true || property.format === "password";
-      const control =
-        propertyType === "boolean"
-          ? "SWITCH"
-          : propertyType === "number" || propertyType === "integer"
-            ? "NUMBER"
-            : isSensitive
-              ? "PASSWORD"
-              : "TEXT";
-      return {
-        key,
-        label: property.title || humanizeKey(key),
-        description: property.description,
-        control,
-        placeholder: "",
-        required: parsed.required?.includes(key) ?? false,
-        secret: isSensitive,
-        multiline: false,
-        options: [],
-        rows: 0,
-        inferredType: propertyType as "string" | "number" | "integer" | "boolean",
-      } satisfies SupportedField;
-    });
-
-    if (fields.some((field) => field == null)) {
-      return { supported: false, fields: [] };
-    }
-    return { supported: true, fields: fields.filter(Boolean) as SupportedField[] };
-  } catch {
-    return { supported: false, fields: [] };
-  }
-}
-
-function defaultValueForField(field: SupportedField): string | boolean {
   if (field.default_value !== undefined) {
     if (typeof field.default_value === "boolean") {
       return field.default_value;
@@ -122,7 +50,10 @@ function defaultValueForField(field: SupportedField): string | boolean {
   return "";
 }
 
-function valueForField(field: SupportedField, configValue?: PluginConfigValue): string | boolean {
+function valueForField(
+  field: PluginAdminFormField,
+  configValue?: PluginConfigValue,
+): string | boolean {
   const raw = configValue?.[field.key];
   if (typeof raw === "boolean") {
     return raw;
@@ -145,19 +76,12 @@ export function PluginConfigForm({
   isSaving = false,
   isTesting = false,
 }: Props) {
-  const parsedFallback = useMemo(() => parseJSONSchema(schema), [schema]);
+  const inferredDescriptor = useMemo(() => adminFormForConfigSchema(schema), [schema]);
-  const fields = useMemo<SupportedField[]>(() => {
+  const fields = inferredDescriptor?.fields ?? EMPTY_FIELDS;
-    if (schema.admin_form?.fields?.length) {
+  const supported = inferredDescriptor != null;
-      return schema.admin_form.fields;
-    }
-    return parsedFallback.fields;
-  }, [parsedFallback.fields, schema.admin_form?.fields]);
-
-  const supported =
-    fields.length > 0 && (schema.admin_form?.fields?.length ? true : parsedFallback.supported);
 
   const descriptor = useMemo<PluginAdminForm>(() => {
-    const base = schema.admin_form ?? { fields };
+    const base = inferredDescriptor ?? { fields };
     const configured = new Set(configuredSecrets);
     return {
       ...base,
@@ -167,7 +91,7 @@ export function PluginConfigForm({
           : field,
       ),
     };
-  }, [configuredSecrets, fields, schema.admin_form]);
+  }, [configuredSecrets, fields, inferredDescriptor]);
 
   const [values, setValues] = useState<PluginConfigValue>(() =>
     Object.fromEntries(fields.map((field) => [field.key, valueForField(field, value)])),
@@ -248,7 +172,7 @@ export function PluginConfigForm({
             return (
               <div key={key} className="flex items-center justify-between gap-3 text-xs">
                 <span className={clearing ? "text-destructive" : "text-muted-foreground"}>
-                  {field?.label || humanizeKey(key)}: {clearing ? "will be cleared" : "saved"}
+                  {field?.label || humanizeConfigKey(key)}: {clearing ? "will be cleared" : "saved"}
                   {required ? " (required)" : ""}
                 </span>
                 {!required ? (

web/src/components/admin/plugins/SchemaForm.test.tsx

@@ -98,6 +98,32 @@ describe("SchemaForm", () => {
       "true",
     );
   });
+  it("uses a controlling field default when rendering a conditional field", () => {
+    const d: PluginAdminForm = {
+      fields: [
+        {
+          key: "advanced_enabled",
+          label: "Advanced",
+          control: "SWITCH",
+          required: false,
+          secret: false,
+          multiline: false,
+          default_value: true,
+        },
+        {
+          key: "endpoint",
+          label: "Endpoint",
+          control: "TEXT",
+          required: false,
+          secret: false,
+          multiline: false,
+          show_when: [{ field: "advanced_enabled", equals: ["true"] }],
+        },
+      ],
+    };
+    render(<SchemaForm descriptor={d} values={{}} onChange={vi.fn()} />);
+    expect(screen.getByText("Endpoint")).toBeTruthy();
+  });
   it("reports validity through onValidityChange (#14)", () => {
     const onValidityChange = vi.fn();
     const d: PluginAdminForm = {
@@ -193,6 +219,43 @@ describe("SchemaForm collapsible sections", () => {
     fireEvent.click(screen.getByText("Show"));
     expect(screen.getByText("Verbose")).toBeTruthy();
   });
+
+  it("uses a controlling field default when rendering a conditional section", () => {
+    const d: PluginAdminForm = {
+      fields: [
+        {
+          key: "advanced_enabled",
+          label: "Advanced",
+          control: "SWITCH",
+          required: false,
+          secret: false,
+          multiline: false,
+          default_value: true,
+        },
+        {
+          key: "endpoint",
+          label: "Endpoint",
+          control: "TEXT",
+          required: false,
+          secret: false,
+          multiline: false,
+        },
+      ],
+      sections: [
+        {
+          key: "advanced",
+          title: "Advanced options",
+          collapsible: false,
+          collapsed_default: false,
+          field_keys: ["endpoint"],
+          show_when: [{ field: "advanced_enabled", equals: ["true"] }],
+        },
+      ],
+    };
+    render(<SchemaForm descriptor={d} values={{}} onChange={vi.fn()} />);
+    expect(screen.getByText("Advanced options")).toBeTruthy();
+    expect(screen.getByText("Endpoint")).toBeTruthy();
+  });
 });
 
 it("marks a show_when-gated field as nested when it is revealed", () => {

web/src/components/admin/plugins/SchemaForm.tsx

@@ -97,11 +97,13 @@ function ChipsSkeleton() {
 function SchemaFormSection({
   section,
   values,
+  fields,
   forceOpen,
   renderFields,
 }: {
   section: PluginAdminFormSection;
   values: Record<string, unknown>;
+  fields: PluginAdminFormField[];
   forceOpen: boolean;
   renderFields: (keys: string[]) => React.ReactNode;
 }) {
@@ -109,7 +111,7 @@ function SchemaFormSection({
   // (the section has unresolved errors) always wins so setup can't be hidden.
   const [userOpen, setUserOpen] = useState<boolean | null>(null);
 
-  if (!evaluateShowWhen(section.show_when, values)) {
+  if (!evaluateShowWhen(section.show_when, values, fields)) {
     return null;
   }
 
@@ -331,7 +333,9 @@ export function SchemaForm({
   // bordered, divided container so toggles read as a cohesive group instead of
   // a column of separate boxes. Honors show_when on each field.
   function renderFieldList(fields: PluginAdminFormField[]): React.ReactNode {
-    const visible = fields.filter((field) => evaluateShowWhen(field.show_when, values));
+    const visible = fields.filter((field) =>
+      evaluateShowWhen(field.show_when, values, descriptor.fields),
+    );
     const nodes: React.ReactNode[] = [];
     let run: PluginAdminFormField[] = [];
     // Key switch groups by their position in the list, not by their first
@@ -389,6 +393,7 @@ export function SchemaForm({
           key={section.key}
           section={section}
           values={values}
+          fields={descriptor.fields}
           forceOpen={section.field_keys.some((key) => mergedErrors[key] != null)}
           renderFields={(keys) => renderFieldList(resolveKeys(keys))}
         />

web/src/components/admin/plugins/configSchemaAdminForm.test.ts

@@ -0,0 +1,141 @@
+import { describe, expect, it } from "vitest";
+
+import type { PluginConfigSchema } from "@/api/types";
+
+import { adminFormForConfigSchema } from "./configSchemaAdminForm";
+
+function schema(overrides: Partial<PluginConfigSchema> = {}): PluginConfigSchema {
+  return {
+    key: "connection",
+    title: "Connection",
+    json_schema: JSON.stringify({
+      type: "object",
+      properties: {},
+      additionalProperties: false,
+    }),
+    required: true,
+    ...overrides,
+  };
+}
+
+describe("adminFormForConfigSchema", () => {
+  it("preserves primitive JSON Schema defaults on inferred fields", () => {
+    const form = adminFormForConfigSchema(
+      schema({
+        json_schema: JSON.stringify({
+          type: "object",
+          properties: {
+            base_url: { type: "string", default: "https://floppy.example.com" },
+            port: { type: "integer", default: 8080 },
+            verify_tls: { type: "boolean", default: true },
+          },
+        }),
+      }),
+    );
+
+    expect(form?.fields.map(({ key, default_value }) => ({ key, default_value }))).toEqual([
+      { key: "base_url", default_value: "https://floppy.example.com" },
+      { key: "port", default_value: 8080 },
+      { key: "verify_tls", default_value: true },
+    ]);
+  });
+
+  it("adds scalar schema properties omitted by a partial admin form", () => {
+    const form = adminFormForConfigSchema(
+      schema({
+        json_schema: JSON.stringify({
+          type: "object",
+          properties: {
+            base_url: { type: "string" },
+            username: { type: "string" },
+          },
+          required: ["base_url", "username"],
+        }),
+        admin_form: {
+          fields: [
+            {
+              key: "base_url",
+              label: "Custom URL",
+              control: "TEXT",
+              required: true,
+              secret: false,
+              multiline: false,
+            },
+          ],
+        },
+      }),
+    );
+
+    expect(form?.fields.map(({ key, label, required }) => ({ key, label, required }))).toEqual([
+      { key: "base_url", label: "Custom URL", required: true },
+      { key: "username", label: "Username", required: true },
+    ]);
+  });
+
+  it("preserves an explicit form when its schema cannot be inferred", () => {
+    const explicit = {
+      fields: [
+        {
+          key: "api_key",
+          label: "API Key",
+          control: "PASSWORD" as const,
+          required: true,
+          secret: true,
+          multiline: false,
+        },
+      ],
+    };
+    expect(adminFormForConfigSchema(schema({ json_schema: "", admin_form: explicit }))).toBe(
+      explicit,
+    );
+  });
+
+  it("applies JSON Schema sensitivity to matching explicit fields", () => {
+    const form = adminFormForConfigSchema(
+      schema({
+        json_schema: JSON.stringify({
+          type: "object",
+          properties: {
+            api_key: { type: "string", writeOnly: true },
+            password: { type: "string", format: "password" },
+            advanced: { type: "object", writeOnly: true },
+          },
+        }),
+        admin_form: {
+          fields: [
+            {
+              key: "api_key",
+              label: "API Key",
+              control: "TEXT",
+              required: false,
+              secret: false,
+              multiline: false,
+            },
+            {
+              key: "password",
+              label: "Password",
+              control: "TEXT",
+              required: false,
+              secret: false,
+              multiline: false,
+            },
+            {
+              key: "advanced",
+              label: "Advanced",
+              control: "TEXTAREA",
+              required: false,
+              secret: false,
+              multiline: true,
+            },
+          ],
+        },
+      }),
+    );
+
+    expect(form?.fields.map(({ key, secret }) => ({ key, secret }))).toEqual([
+      { key: "api_key", secret: true },
+      { key: "password", secret: true },
+      { key: "advanced", secret: true },
+    ]);
+  });
+});

web/src/components/admin/plugins/configSchemaAdminForm.ts

@@ -0,0 +1,98 @@
+import type { PluginAdminForm, PluginAdminFormField, PluginConfigSchema } from "@/api/types";
+
+export function humanizeConfigKey(value: string) {
+  return value
+    .split("_")
+    .filter(Boolean)
+    .map((part) => part.charAt(0).toUpperCase() + part.slice(1))
+    .join(" ");
+}
+
+export function adminFormForConfigSchema(schema: PluginConfigSchema): PluginAdminForm | null {
+  const explicitFields = schema.admin_form?.fields ?? [];
+  try {
+    const parsed = JSON.parse(schema.json_schema) as {
+      type?: string;
+      required?: string[];
+      properties?: Record<
+        string,
+        {
+          type?: string;
+          title?: string;
+          description?: string;
+          writeOnly?: boolean;
+          format?: string;
+          default?: unknown;
+        }
+      >;
+    };
+    if (parsed.type !== "object" || !parsed.properties) {
+      return explicitFields.length > 0 ? schema.admin_form! : null;
+    }
+
+    const inferredFields = Object.entries(parsed.properties).map(
+      ([key, property]): PluginAdminFormField | null => {
+        const propertyType = property.type;
+        if (!propertyType || !["string", "number", "integer", "boolean"].includes(propertyType)) {
+          return null;
+        }
+        const secret = property.writeOnly === true || property.format === "password";
+        const control =
+          propertyType === "boolean"
+            ? "SWITCH"
+            : propertyType === "number" || propertyType === "integer"
+              ? "NUMBER"
+              : secret
+                ? "PASSWORD"
+                : "TEXT";
+        return {
+          key,
+          label: property.title || humanizeConfigKey(key),
+          description: property.description,
+          control,
+          placeholder: "",
+          required: parsed.required?.includes(key) ?? false,
+          secret,
+          multiline: false,
+          default_value:
+            typeof property.default === "string" ||
+            typeof property.default === "number" ||
+            typeof property.default === "boolean"
+              ? property.default
+              : undefined,
+          options: [],
+          rows: 0,
+        };
+      },
+    );
+
+    if (explicitFields.length > 0) {
+      const explicitKeys = new Set(explicitFields.map((field) => field.key));
+      const sensitiveKeys = new Set(
+        Object.entries(parsed.properties)
+          .filter(([, property]) => property.writeOnly === true || property.format === "password")
+          .map(([key]) => key),
+      );
+      return {
+        ...schema.admin_form,
+        fields: [
+          ...explicitFields.map((field) =>
+            sensitiveKeys.has(field.key) ? { ...field, secret: true } : field,
+          ),
+          ...inferredFields.filter(
+            (field): field is PluginAdminFormField => field != null && !explicitKeys.has(field.key),
+          ),
+        ],
+      };
+    }
+
+    if (inferredFields.some((field) => field == null)) return null;
+
+    return {
+      ...schema.admin_form,
+      fields: inferredFields.filter((field): field is PluginAdminFormField => field != null),
+    };
+  } catch {
+    return explicitFields.length > 0 ? schema.admin_form! : null;
+  }
+}

web/src/components/admin/plugins/schemaFormUtils.test.ts

@@ -172,6 +172,7 @@ describe("parseFieldTypes (#15)", () => {
         root_folder: { type: "string" },
         tags: { type: "array", items: { type: "integer" } },
         labels: { type: "array", items: { type: "string" } },
+        flags: { type: "array", items: { type: "boolean" } },
         enabled: { type: "boolean" },
       },
     });
@@ -180,6 +181,7 @@ describe("parseFieldTypes (#15)", () => {
       root_folder: "string",
       tags: "array:int",
       labels: "array",
+      flags: "array:bool",
       enabled: "boolean",
     });
   });
@@ -443,3 +445,63 @@ describe("coerceFieldValue array:num coercion (CodeRabbit #5)", () => {
     expect(coerceFieldValue(numArrayField, ["abc"], "array:num")).toEqual(["abc"]);
   });
 });
+
+describe("coerceFieldValue array:bool coercion", () => {
+  it("coerces boolean multi-select values using their declared item type", () => {
+    expect(coerceFieldValue(numArrayField, ["true", "false", true], "array:bool")).toEqual([
+      true,
+      false,
+      true,
+    ]);
+  });
+});
+
+describe("section visibility", () => {
+  const sectionDescriptor: PluginAdminForm = {
+    fields: [
+      {
+        key: "advanced_enabled",
+        label: "Advanced",
+        control: "SWITCH",
+        required: false,
+        secret: false,
+        multiline: false,
+        default_value: false,
+      },
+      {
+        key: "endpoint",
+        label: "Endpoint",
+        control: "TEXT",
+        required: true,
+        secret: false,
+        multiline: false,
+      },
+    ],
+    sections: [
+      {
+        key: "advanced",
+        title: "Advanced",
+        collapsible: false,
+        collapsed_default: false,
+        field_keys: ["endpoint"],
+        show_when: [{ field: "advanced_enabled", equals: ["true"] }],
+      },
+    ],
+  };
+
+  it("does not validate required fields in a hidden section", () => {
+    expect(validateSchemaValues(sectionDescriptor, { endpoint: "stale" })).toEqual({});
+    expect(validateSchemaValues(sectionDescriptor, { advanced_enabled: true }).endpoint).toMatch(
+      /required/i,
+    );
+  });
+
+  it("does not persist stale values from a hidden section", () => {
+    expect(buildSchemaValues(sectionDescriptor, { endpoint: "stale" })).toEqual({
+      advanced_enabled: false,
+    });
+    expect(
+      buildSchemaValues(sectionDescriptor, { advanced_enabled: true, endpoint: "active" }),
+    ).toEqual({ advanced_enabled: true, endpoint: "active" });
+  });
+});

web/src/components/admin/plugins/schemaFormUtils.ts

@@ -45,7 +45,7 @@ export function validateSchemaValues(
 ): Record<string, string> {
   const errors: Record<string, string> = {};
   for (const field of descriptor.fields) {
-    if (!evaluateShowWhen(field.show_when, values, descriptor.fields)) continue;
+    if (!fieldIsVisible(descriptor, field, values)) continue;
     const raw = effectiveValue(field, values);
     if (field.required && isEmpty(raw)) {
       errors[field.key] = `${field.label || field.key} is required`;
@@ -120,6 +120,7 @@ export type FieldType =
   | "number"
   | "boolean"
   | "array"
+  | "array:bool"
   | "array:int"
   | "array:num";
 
@@ -150,6 +151,7 @@ export function parseFieldTypes(jsonSchema: string | undefined | null): Record<s
         items && typeof items === "object" ? (items as { type?: unknown }).type : undefined;
       if (itemType === "integer") out[key] = "array:int";
       else if (itemType === "number") out[key] = "array:num";
+      else if (itemType === "boolean") out[key] = "array:bool";
       else out[key] = "array";
     } else if (type === "string" || type === "integer" || type === "number" || type === "boolean") {
       out[key] = type;
@@ -191,10 +193,12 @@ export function coerceFieldValue(
         return raw;
       }
       case "array":
+      case "array:bool":
       case "array:int":
       case "array:num": {
         const arr = Array.isArray(raw) ? raw : [];
         if (fieldType === "array") return arr;
+        if (fieldType === "array:bool") return arr.map((v) => coerceBoolean(v));
         if (fieldType === "array:int") return arr.map((v) => coerceNumericString(v));
         return arr.map((v) => coerceNumberString(v));
       }
@@ -237,6 +241,23 @@ export function effectiveValue(
   return values[field.key] !== undefined ? values[field.key] : field.default_value;
 }
 
+export function fieldIsVisible(
+  descriptor: PluginAdminForm,
+  field: PluginAdminFormField,
+  values: Record<string, unknown>,
+): boolean {
+  if (!evaluateShowWhen(field.show_when, values, descriptor.fields)) return false;
+  const containingSections = (descriptor.sections ?? []).filter((section) =>
+    section.field_keys.includes(field.key),
+  );
+  return (
+    containingSections.length === 0 ||
+    containingSections.some((section) =>
+      evaluateShowWhen(section.show_when, values, descriptor.fields),
+    )
+  );
+}
+
 export function buildSchemaValues(
   descriptor: PluginAdminForm,
   draft: Record<string, unknown>,
@@ -244,7 +265,7 @@ export function buildSchemaValues(
 ): Record<string, unknown> {
   const out: Record<string, unknown> = {};
   for (const field of descriptor.fields) {
-    if (!evaluateShowWhen(field.show_when, draft, descriptor.fields)) continue; // don't persist hidden fields' stale values
+    if (!fieldIsVisible(descriptor, field, draft)) continue; // don't persist hidden fields' stale values
     // Fall back to the declared default for untouched fields so an unmodified
     // default persists exactly as it is displayed.
     const rawSource = draft[field.key] !== undefined ? draft[field.key] : field.default_value;

web/src/hooks/queries/watchProviders.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it, vi } from "vitest";
 import {
+  connectWatchProviderAPIKey,
   fetchWatchProviders,
   pollWatchProviderDeviceAuth,
   startWatchProviderDeviceAuth,
@@ -42,6 +43,7 @@ vi.mock("@/api/client", () => ({
     return {
       path,
       method: init?.method ?? "GET",
+      body: init?.body,
     };
   }),
 }));
@@ -61,6 +63,18 @@ describe("watch provider queries", () => {
       path: "/watch-providers/trakt/auth/poll",
       method: "POST",
     });
+    await expect(
+      connectWatchProviderAPIKey("plugin:4:floppy", "token", {
+        floppy: { base_url: "https://floppy.example.com" },
+      }),
+    ).resolves.toMatchObject({
+      path: "/watch-providers/plugin:4:floppy/auth/api-key",
+      method: "POST",
+      body: JSON.stringify({
+        api_key: "token",
+        connection_config: { floppy: { base_url: "https://floppy.example.com" } },
+      }),
+    });
     await expect(
       updateWatchProviderConnection("trakt", { scrobble_enabled: true }),
     ).resolves.toMatchObject({

web/src/hooks/queries/watchProviders.ts

@@ -3,11 +3,15 @@ import { api, ApiClientError } from "@/api/client";
 import { favoriteKeys, watchlistKeys, watchProviderKeys } from "./keys";
 import { toast } from "sonner";
 import { storage } from "@/utils/storage";
+import type { PluginConfigSchema } from "@/api/types";
+
+export type WatchProviderConnectionConfig = Record<string, Record<string, unknown>>;
 
 export interface WatchProviderSummary {
   key: string;
   display_name: string;
   capabilities: WatchProviderCapabilities;
+  connection_config_schema?: PluginConfigSchema[];
 }
 
 export const WatchProviderAuthMethod = {
@@ -52,6 +56,7 @@ export interface WatchProviderConnection {
   sync_watchlist_order_enabled: boolean;
   scrobble_enabled: boolean;
   credentials_configured: boolean;
+  connection_config_schema?: PluginConfigSchema[];
   last_inbound_sync_at?: string;
   last_progress_sync_at?: string;
   last_outbound_sync_at?: string;
@@ -159,10 +164,14 @@ export function pollWatchProviderDeviceAuth(provider: string, authSessionId: str
   });
 }
 
-export function connectWatchProviderAPIKey(provider: string, apiKey: string) {
+export function connectWatchProviderAPIKey(
+  provider: string,
+  apiKey: string,
+  connectionConfig: WatchProviderConnectionConfig = {},
+) {
   return api<WatchProviderConnection>(`/watch-providers/${provider}/auth/api-key`, {
     method: "POST",
-    body: JSON.stringify({ api_key: apiKey }),
+    body: JSON.stringify({ api_key: apiKey, connection_config: connectionConfig }),
   });
 }
 
@@ -248,7 +257,13 @@ export function usePollWatchProviderDeviceAuth(provider: string) {
 export function useConnectWatchProviderAPIKey(provider: string) {
   const queryClient = useQueryClient();
   return useMutation({
-    mutationFn: (apiKey: string) => connectWatchProviderAPIKey(provider, apiKey),
+    mutationFn: ({
+      apiKey,
+      connectionConfig,
+    }: {
+      apiKey: string;
+      connectionConfig?: WatchProviderConnectionConfig;
+    }) => connectWatchProviderAPIKey(provider, apiKey, connectionConfig),
     onSuccess: (connection) => {
       const profileId = getActiveProfileId();
       queryClient.setQueryData(watchProviderKeys.connection(profileId, provider), connection);

web/src/pages/settings/WatchProvidersSettings.tsx

@@ -32,6 +32,14 @@ import {
   type WatchProviderSyncRun,
 } from "@/hooks/queries/watchProviders";
 import { Input } from "@/components/ui/input";
+import { SchemaForm } from "@/components/admin/plugins/SchemaForm";
+import type { PluginConfigSchema } from "@/api/types";
+import type { WatchProviderConnectionConfig } from "@/hooks/queries/watchProviders";
+import {
+  buildConnectionConfig,
+  connectionSchemasAreValid,
+  renderableConnectionSchemas,
+} from "./watchProviderConnectionConfig";
 
 function formatRelativeTime(value?: string) {
   if (!value) return "Never";
@@ -214,17 +222,29 @@ function AuthCodeBlock({
 
 function APIKeyBlock({
   displayName,
+  providerKey,
+  configSchemas,
   pending,
   onSubmit,
   onCancel,
 }: {
   displayName: string;
+  providerKey: string;
+  configSchemas: PluginConfigSchema[];
   pending: boolean;
-  onSubmit: (apiKey: string) => void;
+  onSubmit: (apiKey: string, connectionConfig: WatchProviderConnectionConfig) => void;
   onCancel: () => void;
 }) {
   const [value, setValue] = useState("");
+  const [connectionConfig, setConnectionConfig] = useState<WatchProviderConnectionConfig>({});
+  const [configValidity, setConfigValidity] = useState<Record<string, boolean>>({});
   const trimmed = value.trim();
+  const renderableSchemas = renderableConnectionSchemas(configSchemas);
+  const configValid = connectionSchemasAreValid(
+    renderableSchemas,
+    connectionConfig,
+    configValidity,
+  );
 
   return (
     <div className="border-primary/30 bg-primary/5 rounded-xl border border-dashed p-4">
@@ -245,6 +265,29 @@ function APIKeyBlock({
           <X className="h-4 w-4" />
         </Button>
       </div>
+      {renderableSchemas.map((schema) => (
+        <div key={schema.key} className="mt-4 space-y-2">
+          <div>
+            <div className="text-sm font-medium">{schema.title || schema.key}</div>
+            {schema.description ? (
+              <div className="text-muted-foreground mt-0.5 text-xs leading-snug">
+                {schema.description}
+              </div>
+            ) : null}
+          </div>
+          <SchemaForm
+            descriptor={schema.admin_form}
+            values={connectionConfig[schema.key] ?? {}}
+            onChange={(next) =>
+              setConnectionConfig((current) => ({ ...current, [schema.key]: next }))
+            }
+            idPrefix={`watch-provider-${providerKey}-${schema.key}`}
+            onValidityChange={(valid) =>
+              setConfigValidity((current) => ({ ...current, [schema.key]: valid }))
+            }
+          />
+        </div>
+      ))}
       <div className="mt-3 flex flex-col gap-2 sm:flex-row sm:items-stretch">
         <Input
           type="password"
@@ -258,8 +301,10 @@ function APIKeyBlock({
         <Button
           type="button"
           size="sm"
-          disabled={pending || trimmed.length === 0}
+          disabled={pending || trimmed.length === 0 || !configValid}
-          onClick={() => onSubmit(trimmed)}
+          onClick={() =>
+            onSubmit(trimmed, buildConnectionConfig(renderableSchemas, connectionConfig))
+          }
           className="sm:flex-none"
         >
           {pending ? <Loader2 className="h-4 w-4 animate-spin" /> : <Check className="h-4 w-4" />}
@@ -454,12 +499,15 @@ function WatchProviderCard({ providerKey }: { providerKey: string }) {
     });
   };
 
-  const handleSubmitAPIKey = (apiKey: string) => {
+  const handleSubmitAPIKey = (apiKey: string, connectionConfig: WatchProviderConnectionConfig) => {
-    connectAPIKey.mutate(apiKey, {
+    connectAPIKey.mutate(
-      onSuccess: () => {
+      { apiKey, connectionConfig },
-        setApiKeyPrompt(false);
+      {
+        onSuccess: () => {
+          setApiKeyPrompt(false);
+        },
       },
-    });
+    );
   };
 
   const handleCancelAPIKey = () => {
@@ -549,6 +597,8 @@ function WatchProviderCard({ providerKey }: { providerKey: string }) {
         <div className="mt-4">
           <APIKeyBlock
             displayName={displayName}
+            providerKey={providerKey}
+            configSchemas={connection.connection_config_schema ?? []}
             pending={connectAPIKey.isPending}
             onSubmit={handleSubmitAPIKey}
             onCancel={handleCancelAPIKey}
@@ -622,14 +672,18 @@ function WatchProviderCard({ providerKey }: { providerKey: string }) {
               disabled={isBusy}
               onChange={(checked) => updateConnection.mutate({ export_watched_enabled: checked })}
             />
-            <ToggleRow
+            {connection.capabilities.export_unwatched ? (
-              id={`watch-provider-${providerKey}-export-unwatched`}
+              <ToggleRow
-              label="Send unwatched changes"
+                id={`watch-provider-${providerKey}-export-unwatched`}
-              description="When you mark something unwatched, remove matching history from this provider."
+                label="Send unwatched changes"
-              checked={connection.export_unwatched_enabled}
+                description="When you mark something unwatched, remove matching history from this provider."
-              disabled={isBusy}
+                checked={connection.export_unwatched_enabled}
-              onChange={(checked) => updateConnection.mutate({ export_unwatched_enabled: checked })}
+                disabled={isBusy}
-            />
+                onChange={(checked) =>
+                  updateConnection.mutate({ export_unwatched_enabled: checked })
+                }
+              />
+            ) : null}
             {connection.capabilities.import_favorites ||
             connection.capabilities.export_favorites ? (
               <ToggleRow

web/src/pages/settings/watchProviderConnectionConfig.test.ts

@@ -0,0 +1,73 @@
+import { describe, expect, it } from "vitest";
+
+import type { PluginConfigSchema } from "@/api/types";
+
+import {
+  activeConnectionSchemas,
+  buildConnectionConfig,
+  connectionSchemasAreValid,
+  renderableConnectionSchemas,
+} from "./watchProviderConnectionConfig";
+
+const optionalSchema: PluginConfigSchema = {
+  key: "optional",
+  title: "Optional server settings",
+  description: "Only sent when configured.",
+  json_schema: JSON.stringify({
+    type: "object",
+    properties: { base_url: { type: "string" } },
+    required: ["base_url"],
+  }),
+  required: false,
+  admin_form: {
+    fields: [
+      {
+        key: "base_url",
+        label: "Server URL",
+        control: "TEXT",
+        required: true,
+        secret: false,
+        multiline: false,
+      },
+    ],
+  },
+};
+
+describe("watch provider connection config", () => {
+  it("omits an untouched optional schema without blocking Connect", () => {
+    expect(activeConnectionSchemas([optionalSchema], {})).toEqual([]);
+    expect(connectionSchemasAreValid([optionalSchema], {}, { optional: false })).toBe(true);
+    expect(buildConnectionConfig([optionalSchema], {})).toEqual({});
+  });
+
+  it("validates and submits an optional schema after the user enters a value", () => {
+    const drafts = { optional: { base_url: "https://floppy.example.com" } };
+    expect(connectionSchemasAreValid([optionalSchema], drafts, { optional: false })).toBe(false);
+    expect(connectionSchemasAreValid([optionalSchema], drafts, { optional: true })).toBe(true);
+    expect(buildConnectionConfig([optionalSchema], drafts)).toEqual({
+      optional: { base_url: "https://floppy.example.com" },
+    });
+  });
+
+  it("keeps required schemas active before any values are entered", () => {
+    const requiredSchema = { ...optionalSchema, key: "required", required: true };
+    expect(activeConnectionSchemas([requiredSchema], {})).toEqual([requiredSchema]);
+    expect(connectionSchemasAreValid([requiredSchema], {}, {})).toBe(false);
+  });
+
+  it("derives a usable form for a required JSON-schema-only block", () => {
+    const headless = { ...optionalSchema, required: true, admin_form: undefined };
+    const schemas = renderableConnectionSchemas([headless]);
+    expect(schemas).toHaveLength(1);
+    const renderable = schemas[0]!;
+    expect(renderable.admin_form.fields).toEqual([
+      expect.objectContaining({
+        key: "base_url",
+        label: "Base Url",
+        control: "TEXT",
+        required: true,
+      }),
+    ]);
+    expect(connectionSchemasAreValid([renderable], {}, {})).toBe(false);
+  });
+});

web/src/pages/settings/watchProviderConnectionConfig.ts

@@ -0,0 +1,59 @@
+import type { PluginConfigSchema } from "@/api/types";
+import { adminFormForConfigSchema } from "@/components/admin/plugins/configSchemaAdminForm";
+import { buildSchemaValues, parseFieldTypes } from "@/components/admin/plugins/schemaFormUtils";
+import type { WatchProviderConnectionConfig } from "@/hooks/queries/watchProviders";
+
+export type RenderableConnectionSchema = PluginConfigSchema & {
+  admin_form: NonNullable<PluginConfigSchema["admin_form"]>;
+};
+
+export function renderableConnectionSchemas(
+  schemas: PluginConfigSchema[],
+): RenderableConnectionSchema[] {
+  return schemas.flatMap((schema) => {
+    const adminForm = adminFormForConfigSchema(schema);
+    return adminForm == null ? [] : [{ ...schema, admin_form: adminForm }];
+  });
+}
+
+function hasEnteredValue(value: unknown): boolean {
+  if (value == null) return false;
+  if (typeof value === "string") return value.trim().length > 0;
+  if (Array.isArray(value)) return value.some(hasEnteredValue);
+  if (typeof value === "object") return Object.values(value).some(hasEnteredValue);
+  return true;
+}
+
+export function activeConnectionSchemas(
+  schemas: PluginConfigSchema[],
+  drafts: WatchProviderConnectionConfig,
+): RenderableConnectionSchema[] {
+  return schemas.filter(
+    (schema): schema is RenderableConnectionSchema =>
+      schema.admin_form != null && (schema.required || hasEnteredValue(drafts[schema.key])),
+  );
+}
+
+export function connectionSchemasAreValid(
+  schemas: PluginConfigSchema[],
+  drafts: WatchProviderConnectionConfig,
+  validity: Record<string, boolean>,
+): boolean {
+  return activeConnectionSchemas(schemas, drafts).every((schema) => validity[schema.key] ?? false);
+}
+
+export function buildConnectionConfig(
+  schemas: PluginConfigSchema[],
+  drafts: WatchProviderConnectionConfig,
+): WatchProviderConnectionConfig {
+  return Object.fromEntries(
+    activeConnectionSchemas(schemas, drafts).map((schema) => [
+      schema.key,
+      buildSchemaValues(
+        schema.admin_form,
+        drafts[schema.key] ?? {},
+        parseFieldTypes(schema.json_schema),
+      ),
+    ]),
+  );
+}