feat(plugins)!: load product-owned Connectivity executables

cmd/connectivity_cancel_test.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/formancehq/fctl/v4/cmd"
 	"github.com/formancehq/fctl/v4/internal/interactive"
 )
 
@@ -32,11 +31,11 @@ func TestConnectivityCancelAndDeclineNeverWrite(t *testing.T) {
 				}
 				return []string{"false"}, nil
 			})
-			root := cmd.NewRootCommand()
 			args := []string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", action}
 			if action == "delete" {
 				args = append(args, "ingestion")
 			}
+			root := connectivityRoot(t, args)
 			root.SetArgs(args)
 			root.SetOut(io.Discard)
 			root.SetErr(io.Discard)

cmd/connectivity_external_test.go

@@ -0,0 +1,33 @@
+package cmd_test
+
+import (
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/spf13/cobra"
+
+	"github.com/formancehq/fctl/v4/cmd"
+)
+
+// Product integration is opt-in: the host must never import or build the
+// Connectivity module. Supply a product-built executable at exact version 1.2.3.
+func connectivityRoot(t *testing.T, args []string) *cobra.Command {
+	t.Helper()
+	binary := os.Getenv("FCTL_CONNECTIVITY_TEST_BINARY")
+	if binary == "" {
+		t.Skip("set FCTL_CONNECTIVITY_TEST_BINARY to product executable built for 1.2.3")
+	}
+	prefix := []string{}
+	for i := 0; i < len(args) && args[i] != "connectivity"; i++ {
+		prefix = append(prefix, args[i])
+	}
+	install := append(append([]string{}, prefix...), "plugins", "install", "--service", "connectivity", "--binary", binary, "--service-version", "1.2.3")
+	out, trace, err := executeExternalCLI(t, install)
+	if err != nil || !strings.Contains(out, "connectivity") {
+		t.Fatalf("install: %s %s %v", out, trace, err)
+	}
+	// Cached metadata attaches forms without discovery; the actual execution
+	// still checks the live service version and uses the host HTTP broker.
+	return cmd.NewRootCommandWithArgs(t.Context(), append(prefix, "--help"))
+}

cmd/connectivity_test.go

@@ -14,7 +14,6 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/formancehq/fctl/v4/cmd"
 	"github.com/formancehq/fctl/v4/internal/interactive"
 )
 
@@ -30,6 +29,10 @@ func TestConnectivityCreationFormPaginatesChoices(t *testing.T) {
 	t.Setenv("FCTL_NO_INPUT", "")
 	pages, writes := 0, 0
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/_info" {
+			writeConnectivityResponse(t, w, `{"version":"1.2.3"}`)
+			return
+		}
 		if r.Method == http.MethodGet && r.URL.Path == "/connectors" {
 			pages++
 			response := `{"cursor":{"pageSize":15,"hasMore":true,"next":"opaque +/=","data":[{"metadata":{"name":"adyen"},"spec":{"displayName":"Adyen"}}]}}`
@@ -81,8 +84,9 @@ func TestConnectivityCreationFormPaginatesChoices(t *testing.T) {
 		}
 		return values, nil
 	})
-	root := cmd.NewRootCommand()
-	root.SetArgs([]string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "create"})
+	args := []string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "create"}
+	root := connectivityRoot(t, args)
+	root.SetArgs(args)
 	var output bytes.Buffer
 	root.SetOut(&output)
 	root.SetErr(io.Discard)
@@ -97,8 +101,12 @@ func TestConnectivityCreationFormPaginatesChoices(t *testing.T) {
 func TestConnectivityExplicitBodiesBypassForms(t *testing.T) {
 	body := `{"suspend":false,"version":null,"startSequence":9007199254740993}`
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/_info" {
+			writeConnectivityResponse(t, w, `{"version":"1.2.3"}`)
+			return
+		}
 		data, err := io.ReadAll(r.Body)
-		if err != nil || string(data) != body || r.Method != http.MethodPatch || r.Header.Get("Content-Type") != "application/merge-patch+json" {
+		if err != nil || string(data) != `{"spec":`+body+`}` || r.Method != http.MethodPatch || r.Header.Get("Content-Type") != "application/merge-patch+json" {
 			t.Errorf("method=%s type=%s body=%s err=%v", r.Method, r.Header.Get("Content-Type"), data, err)
 		}
 		writeConnectivityResponse(t, w, `{"metadata":{"name":"ingestion"}}`)
@@ -110,8 +118,9 @@ func TestConnectivityExplicitBodiesBypassForms(t *testing.T) {
 	}
 	for _, input := range []string{body, "@" + file, "-"} {
 		t.Run(input[:1], func(t *testing.T) {
-			root := cmd.NewRootCommand()
-			root.SetArgs([]string{"--config-dir", t.TempDir(), "--no-input", "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "patch", "ingestion", "--data", input})
+			args := []string{"--config-dir", t.TempDir(), "--no-input", "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "patch", "ingestion", "--data", input}
+			root := connectivityRoot(t, args)
+			root.SetArgs(args)
 			root.SetIn(strings.NewReader(body))
 			root.SetOut(io.Discard)
 			root.SetErr(io.Discard)
@@ -129,13 +138,17 @@ func TestConnectivityPatchFormAndInstanceSelection(t *testing.T) {
 	patch := `{"suspend":true,"version":null}`
 	writes := 0
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/_info" {
+			writeConnectivityResponse(t, w, `{"version":"1.2.3"}`)
+			return
+		}
 		if r.Method == http.MethodGet && r.URL.Path == "/connectorinstances" {
 			writeConnectivityResponse(t, w, `{"cursor":{"hasMore":false,"data":[{"metadata":{"name":"ingestion"},"spec":{"connector":"stripe","ledger":"books"},"status":{"phase":"Running"}}]}}`)
 			return
 		}
 		writes++
 		data, err := io.ReadAll(r.Body)
-		if err != nil || r.Method != http.MethodPatch || r.URL.Path != "/connectorinstances/ingestion" || string(data) != patch {
+		if err != nil || r.Method != http.MethodPatch || r.URL.Path != "/connectorinstances/ingestion" || string(data) != `{"spec":`+patch+`}` {
 			t.Errorf("patch body=%s err=%v", data, err)
 		}
 		writeConnectivityResponse(t, w, `{}`)
@@ -155,8 +168,9 @@ func TestConnectivityPatchFormAndInstanceSelection(t *testing.T) {
 		}
 		return values, nil
 	})
-	root := cmd.NewRootCommand()
-	root.SetArgs([]string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "patch"})
+	args := []string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "patch"}
+	root := connectivityRoot(t, args)
+	root.SetArgs(args)
 	root.SetOut(io.Discard)
 	root.SetErr(io.Discard)
 	if err := root.ExecuteContext(interactive.WithRunner(t.Context(), runner)); err != nil {

cmd/plugins_install.go

@@ -21,7 +21,7 @@ func newPluginsInstallCommand(settings *connection.Settings) *cobra.Command {
 		Annotations: map[string]string{"fctl.target": "stack"},
 		Example:     "  fctl plugins install --binary ./build/fctl-plugin-ledger --profile local",
 	}
-	install.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger or auth")
+	install.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger, auth or connectivity")
 	install.Flags().StringVar(&binary, "binary", "", "Explicit path to the trusted executable")
 	install.Flags().StringVar(&localVersion, "service-version", "", "Exact service version; defaults to selected service /_info discovery")
 	install.RunE = func(cmd *cobra.Command, _ []string) error {

cmd/plugins_show.go

@@ -9,7 +9,7 @@ import (
 func newPluginsShowCommand(settings *connection.Settings) *cobra.Command {
 	var service string
 	show := &cobra.Command{Use: "show", Short: "Show the service plugin locked for the selected target", Args: cobra.NoArgs}
-	show.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger or auth")
+	show.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger, auth or connectivity")
 	show.RunE = func(cmd *cobra.Command, _ []string) error {
 		if err := distributionService(service); err != nil {
 			return err

cmd/plugins_sync.go

@@ -17,7 +17,7 @@ func newPluginsSyncCommand(settings *connection.Settings) *cobra.Command {
 		Annotations: map[string]string{"fctl.target": "stack"},
 		Example:     "  fctl plugins sync --organization ORG --stack STACK\n  fctl plugins sync --catalogue registry.yaml --service-version 3.0.0 --profile local",
 	}
-	sync.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger or auth")
+	sync.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger, auth or connectivity")
 	sync.Flags().StringVar(&catalogue, "catalogue", "", "YAML or JSON catalogue file or HTTPS URL; defaults to the official registry (FCTL_PLUGIN_CATALOGUE)")
 	sync.Flags().StringVar(&serviceVersion, "service-version", "", "Exact service version; defaults to selected service /_info discovery")
 	sync.Flags().IntVar(&revision, "revision", 0, "Plugin revision; defaults to the highest for this exact service version")

cmd/plugins_target.go

@@ -25,6 +25,8 @@ func supportedPluginService(service string) (serviceDescriptor, error) {
 	switch service {
 	case "auth":
 		return serviceDescriptor{name: "auth", title: "Auth"}, nil
+	case "connectivity":
+		return serviceDescriptor{name: "connectivity", title: "Connectivity"}, nil
 	case "ledger":
 		return serviceDescriptor{name: "ledger", title: "Ledger"}, nil
 	default:
@@ -49,6 +51,9 @@ func pluginServiceTarget(settings *connection.Settings, cmd *cobra.Command, serv
 	if descriptor.name == "auth" {
 		endpoint = options.AuthURL
 	}
+	if descriptor.name == "connectivity" {
+		endpoint = options.ConnectivityURL
+	}
 	if options.AuthMode == "cloud" {
 		endpoint = cmp.Or(options.Issuer, cloud.DefaultIssuer)
 	} else if endpoint == "" && options.StackURL != "" {

cmd/root.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"os/signal"
 
@@ -19,7 +20,6 @@ import (
 	"github.com/formancehq/fctl/v4/internal/plugin"
 	"github.com/formancehq/fctl/v4/plugins/auth"
 	cloudplugin "github.com/formancehq/fctl/v4/plugins/cloud"
-	"github.com/formancehq/fctl/v4/plugins/connectivity"
 	"github.com/formancehq/fctl/v4/plugins/ledger"
 )
 
@@ -51,7 +51,8 @@ func newRootCommand(ctx context.Context, args []string) *cobra.Command {
 	root.PersistentFlags().Bool("no-input", false, "Disable interactive forms and selections (FCTL_NO_INPUT, CI)")
 	authFactory, authPreparationErr := prepareServicePlugin(ctx, root, settings, args, "auth")
 	ledgerFactory, ledgerPreparationErr := prepareServicePlugin(ctx, root, settings, args, "ledger")
-	preparationErrors := map[string]error{"auth": authPreparationErr, "ledger": ledgerPreparationErr}
+	connectivityFactory, connectivityPreparationErr := prepareServicePlugin(ctx, root, settings, args, "connectivity")
+	preparationErrors := map[string]error{"auth": authPreparationErr, "ledger": ledgerPreparationErr, "connectivity": connectivityPreparationErr}
 	root.PersistentPreRunE = func(cmd *cobra.Command, _ []string) error {
 		if err := preparationErrors[serviceCommand(cmd)]; err != nil {
 			return err
@@ -70,14 +71,22 @@ func newRootCommand(ctx context.Context, args []string) *cobra.Command {
 	if ledgerFactory == nil {
 		ledgerFactory = ledger.New
 	}
-	for _, factory := range []plugin.Factory{cloudplugin.New, authFactory, ledgerFactory, connectivity.New} {
+	for _, factory := range []plugin.Factory{cloudplugin.New, authFactory, ledgerFactory, connectivityFactory} {
+		if factory == nil {
+			continue
+		}
 		if err := registry.Register(context.WithoutCancel(ctx), factory(nil), factory); err != nil {
 			panic(err) // Embedded metadata is a build-time invariant, never user input.
 		}
 	}
 	if err := plugin.NewCommandWithRequest(registry, resolve).AddTo(root); err != nil {
 		panic(err)
 	}
+	if connectivityFactory == nil {
+		root.AddCommand(&cobra.Command{Use: "connectivity", Short: "Manage Connectivity with an installed product plugin", RunE: func(*cobra.Command, []string) error {
+			return fmt.Errorf("install the Connectivity plugin with fctl plugins install --service connectivity --binary PATH, or configure FCTL_PLUGIN_CATALOGUE")
+		}})
+	}
 	if err := cloudtools.AddTo(root, settings); err != nil {
 		panic(err)
 	}

cmd/service_loader_test.go

@@ -44,7 +44,7 @@ func loaderFlag(t *testing.T, root *cobra.Command, name, value string) {
 }
 
 func TestServiceLoaderTargets(t *testing.T) {
-	for _, service := range []string{"auth", "ledger"} {
+	for _, service := range []string{"auth", "ledger", "connectivity"} {
 		t.Run(service, func(t *testing.T) { testServiceLoaderTarget(t, service) })
 	}
 	root, settings := loaderSettings(t)
@@ -72,6 +72,7 @@ func testServiceLoaderTarget(t *testing.T, service string) {
 	}
 	loaderFlag(t, root, "auth-url", "https://auth.example")
 	loaderFlag(t, root, "ledger-url", "https://ledger.example")
+	loaderFlag(t, root, "connectivity-url", "https://connectivity.example")
 	target, err = pluginServiceTarget(settings, root, service)
 	if err != nil || target.Endpoint != "https://"+service+".example" {
 		t.Fatalf("direct target: %+v %v", target, err)
@@ -90,7 +91,7 @@ func testServiceLoaderTarget(t *testing.T, service string) {
 
 func TestServiceLoaderManifestBoundary(t *testing.T) {
 	root, _ := loaderSettings(t)
-	for _, service := range []string{"auth", "ledger"} {
+	for _, service := range []string{"auth", "ledger", "connectivity"} {
 		manifest := loaderManifest(service)
 		if err := validateServiceManifest(t.Context(), root, service, manifest); err != nil {
 			t.Fatal(err)
@@ -138,10 +139,10 @@ func TestServiceLoaderOfflineRootsAndCacheIsolation(t *testing.T) {
 		t.Fatal(err)
 	}
 	target := pluginmanager.Target{Profile: "", Organization: "org", Stack: "stack", Endpoint: settings.Options.Issuer}
-	for _, service := range []string{"auth", "ledger"} {
+	for _, service := range []string{"auth", "ledger", "connectivity"} {
 		installLoaderFixture(t, manager, target, service)
 	}
-	for _, service := range []string{"auth", "ledger"} {
+	for _, service := range []string{"auth", "ledger", "connectivity"} {
 		lock, err := cachedServiceLock(settings, root, manager, service)
 		if err != nil || lock.Service != service {
 			t.Fatalf("cached %s: %+v %v", service, lock, err)
@@ -259,6 +260,7 @@ func TestAuthLoaderExactVersionAndStrictTargets(t *testing.T) {
 func TestLedgerLoaderCompatibility(t *testing.T) {
 	root, settings := loaderSettings(t)
 	loaderFlag(t, root, "ledger-url", "https://ledger.example")
+	loaderFlag(t, root, "connectivity-url", "https://connectivity.example")
 	target, err := ledgerTarget(settings, root)
 	if err != nil {
 		t.Fatal(err)
@@ -296,3 +298,41 @@ func TestLedgerLoaderCompatibility(t *testing.T) {
 		t.Fatalf("legacy version: %s %v", version, err)
 	}
 }
+
+func TestConnectivityLoaderExactVersionAndStrictTargets(t *testing.T) {
+	t.Setenv("FCTL_PLUGIN_CATALOGUE", "")
+	base := http.DefaultTransport
+	http.DefaultTransport = discoveryTransport(func(request *http.Request) (*http.Response, error) {
+		if request.URL.String() != "http://127.0.0.1:1/_info" {
+			t.Fatalf("wrong Connectivity version endpoint: %s", request.URL)
+		}
+		return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"version":"v1.0.1"}`))}, nil
+	})
+	t.Cleanup(func() { http.DefaultTransport = base })
+	prep := discoveryPreparation(t, discoveryTransport(func(*http.Request) (*http.Response, error) {
+		return &http.Response{StatusCode: http.StatusServiceUnavailable, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(""))}, nil
+	}))
+	prep.service = serviceDescriptor{name: "connectivity", title: "Connectivity"}
+	loaderFlag(t, prep.root, "auth-mode", "none")
+	loaderFlag(t, prep.root, "connectivity-url", "http://127.0.0.1:1")
+	target, version, err := pluginServiceSyncTarget(t.Context(), prep.settings, prep.root, "connectivity", "")
+	if err != nil || version != "1.0.1" {
+		t.Fatalf("exact Connectivity discovery: %+v %s %v", target, version, err)
+	}
+	empty := &pluginmanager.Catalogue{SchemaVersion: pluginmanager.SchemaVersion}
+	if _, err := prep.matchVersion(t.Context(), pluginmanager.DefaultCatalogue, empty); !errors.Is(err, pluginmanager.ErrNotInstalled) {
+		t.Fatalf("official unprepared missing release: %v", err)
+	}
+	t.Setenv("FCTL_PLUGIN_CATALOGUE", "https://custom.example/catalogue")
+	if _, err := prep.resolve(t.Context(), pluginBootstrap{commands: []string{"connectivity", "probe"}}); !errors.Is(err, pluginmanager.ErrCatalogueUnavailable) {
+		t.Fatalf("custom unavailable target fell back: %v", err)
+	}
+	t.Setenv("FCTL_PLUGIN_CATALOGUE", "")
+	installLoaderFixture(t, prep.manager, target, "connectivity")
+	if _, err := prep.matchVersion(t.Context(), pluginmanager.DefaultCatalogue, empty); !errors.Is(err, pluginmanager.ErrNoRelease) {
+		t.Fatalf("prepared target accepted missing exact release: %v", err)
+	}
+	if _, err := prep.matchVersion(t.Context(), pluginmanager.DefaultCatalogue, nil); !errors.Is(err, pluginmanager.ErrCatalogueUnavailable) {
+		t.Fatalf("prepared unavailable target fell back: %v", err)
+	}
+}

docs/connectivity.md

@@ -1,14 +1,11 @@
 # Connectivity plugin
 
-The embedded `plugins/connectivity` module implements all 14 operations in
-[connectivity-api/openapi.yaml](https://github.com/formancehq/connectivity/blob/ce2324887f5b5ec4e3c2ec934ac874656d4c5348/misc/connectivity-api/openapi.yaml),
-API version `0.1.0`, pinned to commit `ce2324887f5b5ec4e3c2ec934ac874656d4c5348`.
-The repository also provides a generated client under `pkg/client/http`.
-This plugin uses the public `pluginsdk/httpclient` adapter to preserve raw
-request/response JSON, including merge-patch nulls, unknown fields and exact
-numbers, without introducing typed-model defaults such as `replicas`.
-It imports no server, Cobra, UI or core packages.
-The host owns authentication, endpoints, storage, forms and presentation.
+The external product-owned module
+[`github.com/formancehq/connectivity/misc/fctl-plugin`](https://github.com/formancehq/connectivity/tree/main/misc/fctl-plugin)
+implements all 14 Connectivity API operations. It is installed through the
+shared loader; fctl has no Go dependency on the product module and no embedded
+Connectivity fallback. The product owns forms, payload validation and HTTP
+mapping; fctl owns credentials, terminal interaction and cached manifests.
 
 ## Profiles
 
@@ -102,9 +99,24 @@ Deletion maps the empty 204 response to JSON null.
 
 Local fixtures cover all routes/methods, media types, errors, cursor encoding,
 int64 precision, size limits, forms, selection and authentication boundaries.
-The pinned operation snapshot is `plugins/connectivity/testdata/operations.json`.
+The operation snapshot and current API contract tests live in the product module.
 An independent consumer verifies the public SDK boundary without core or UI
 dependencies. A real PTY check covers creation, resource selection, default-No
 deletion confirmation, styled forms and plain tables. Creation produces one
 write and exact JSON on stdout even with forced terminal colors.
 Deployment availability and ingestion require separate live runtime checks.
+
+## Install the exact product executable
+
+```sh
+fctl plugins install --service connectivity --binary ./fctl-plugin-connectivity --profile connectivity-local
+fctl plugins sync --service connectivity --profile connectivity-local
+fctl plugins show --service connectivity --profile connectivity-local
+```
+
+Without an installed matching executable, Connectivity commands require plugin
+installation. Help/completion use cached metadata without launching a process or
+contacting the service. Each execution checks `/_info` again and rejects version
+drift before starting the plugin. Patch input remains a spec-only JSON object;
+the product wraps it as `{"spec":...}` for the API. Read-only replicas and
+connectivityRef are rejected locally; use suspend.

docs/modules.md

@@ -1,6 +1,6 @@
 # Auth and Ledger modules
 
-The root registers Cloud, Auth, Ledger and Connectivity plugin factories. This page describes
+The root registers Cloud, Auth and Ledger defaults plus installed external service plugins, including Connectivity. This page describes
 the service modules; [Cloud management](cloud.md) describes control-plane commands.
 Each plugin implements
 the public `pluginsdk.Plugin` manifest/execution contract. The core builds Cobra