feat: plugin ecosystem + step.workflow_call (#331)

* feat(wfctl): add plugin install --url for direct URL installs

Adds --url flag to wfctl plugin install that downloads a tar.gz archive
from a direct URL, extracts plugin.json to identify the plugin name,
installs to the plugin directory, and records the SHA-256 checksum in
the lockfile.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat(wfctl): enhanced plugin init scaffold with full project structure

Extends wfctl plugin init to generate cmd/workflow-plugin-<name>/main.go,
internal/provider.go, internal/steps.go, go.mod, .goreleaser.yml, CI/release
GitHub Actions workflows, Makefile, and README.md. Adds --module flag for
custom Go module paths. Preserves existing plugin.json and .go skeleton for
backward compatibility.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat: engine auto-fetch for declared external plugins on startup

Add AutoFetchPlugin and AutoFetchDeclaredPlugins to plugin/autofetch.go,
which shell out to wfctl to download plugins not found locally. Extend
WorkflowConfig with a new PluginsConfig / ExternalPluginDecl type so
configs can declare plugins with autoFetch: true and an optional version
constraint. StdEngine gains SetExternalPluginDir and calls
AutoFetchDeclaredPlugins in BuildFromConfig before module loading.
The server's buildEngine registers the plugin dir so auto-fetch is active
at runtime. If wfctl is absent, a warning is logged and startup continues.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: resolve all CI lint failures

- Use struct conversion for staticIndexEntry → PluginSummary (staticcheck S1016)
- Remove unused updateLockfile and writePluginJSON functions
- Add nilerr annotations for intentional nil returns in integrity.go
- Add gosec annotation for exec.Command in autofetch.go
- Fix TestLoadRegistryConfigDefault to use DefaultRegistryConfig() directly

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address Copilot review feedback on engine PR

- integrity.go: fail closed when lockfile exists but is unreadable or
  unparseable, preventing integrity enforcement bypass
- autofetch.go: extract stripVersionConstraint helper; detect compound
  version constraints and fall back to latest; check both pluginName and
  workflow-plugin-<name> alternate form for installed-check; log restart
  warning when new plugins are downloaded (they require a server restart)
- autofetch_test.go: test stripVersionConstraint directly instead of
  duplicating the logic inline; add compound-constraint cases
- engine.go: clarify comment that auto-fetched plugins need a restart

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: address all Copilot review comments on engine PR (#330)

- generator.go: use plugin/external/sdk imports and types (PluginProvider,
  StepInstance, StepResult, StepTypes/CreateStep) instead of plugin/sdk
- PLUGIN_AUTHORING.md: update examples to match external SDK interfaces
- plugin_install.go: hash installed binary (not archive) for lockfile,
  add hashFileSHA256 helper, add install mode mutual exclusivity check,
  update installFromLocal to write lockfile, normalize plugin names
- plugin_lockfile.go: add registry param to updateLockfileWithChecksum,
  pass version/registry in installFromLockfile, remove dir on mismatch
- registry_source.go: validate URL in NewStaticRegistrySource
- config.go: clarify Version field forwarding semantics

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address remaining Copilot review comments on engine PR (#330)

- registry_source.go: use explicit field assignment for PluginSummary
  instead of struct type conversion (clearer, avoids tag confusion)
- plugin_lockfile.go: don't pass @Version in installFromLockfile to
  prevent lockfile overwrite before checksum verification
- plugin_install.go: add verifyInstalledPlugin() call in installFromURL
  for parity with registry installs
- engine.go: add TODO to move auto-fetch before plugin discovery so
  newly fetched plugins are available without restart
- integrity_test.go: add tests for unreadable and malformed lockfile
  to verify fail-closed behavior

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: suppress S1016 lint, add generator project structure tests

- registry_source.go: add nolint:gosimple for S1016 — explicit field
  assignment preferred for clarity across different struct tags
- generator_test.go: add TestGenerateProjectStructure verifying all
  generated files exist and use correct external SDK imports/types

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: move auto-fetch before plugin discovery so fetched plugins load immediately

Auto-fetch was running inside BuildFromConfig, which executes after
external plugins are already discovered and loaded. Plugins downloaded
by auto-fetch required a server restart to take effect.

Move auto-fetch to buildEngine in cmd/server/main.go, before
DiscoverPlugins/LoadPlugin. Remove the now-unused externalPluginDir
field and SetExternalPluginDir from the engine.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: add step.workflow_call for cross-pipeline dispatch

Enables pipelines to call other pipelines by name with full context
forwarding. Supports template-resolved workflow names and stop_pipeline
option. Required for WebSocket message routing to game-specific pipelines.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: address all 9 Copilot review comments on PR #331

- registry_source.go: switch GitHubRegistrySource to use registryHTTPClient
  (timeout-configured) for both ListPlugins and FetchManifest; update comment
- autofetch.go: scan for AutoFetch=true entries before exec.LookPath to avoid
  misleading startup warning when no plugins need auto-fetch
- autofetch.go: add internal autoFetchPlugin helper that accepts *slog.Logger
  and emits structured log entries when a logger is available; public
  AutoFetchPlugin delegates to it; AutoFetchDeclaredPlugins passes its logger
- autofetch_test.go: rename TestAutoFetchPlugin_CorrectArgs to
  TestAutoFetchPlugin_SkipsWhenExists to match what the test actually asserts
- integrity.go: replace os.ReadFile + sha256.Sum256 with streaming
  os.Open + io.Copy into sha256.New() to keep memory bounded for large binaries
- integrity_test.go: add t.Skip guard in UnreadableLockfile test when the file
  is actually readable (Windows / root environments)
- pipeline_step_workflow_call.go: use resolved workflowName (not s.workflow)
  in async return payload and sync error message for consistency
- docs/PLUGIN_AUTHORING.md: clarify the distinction between plugin.json name
  (short, used by engine) and the registry/provider manifest name
  (workflow-plugin- prefixed), and which is used for dependency resolution
- config/config.go: MergeApplicationConfig now merges Plugins.External from
  each referenced workflow file into the combined config, deduplicated by name

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: remove duplicate hashFileSHA256 from merge conflict resolution

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: address new PR #331 review comments

- docs/PLUGIN_AUTHORING.md: close unclosed code fence after StepProvider example
- cmd/wfctl/plugin_install.go: streaming hashFileSHA256 via io.Copy + sha256.New()
- cmd/wfctl/plugin_install.go: warn on hash failure instead of silent empty checksum
- config/merge_test.go: add TestMergeApplicationConfig_PluginDedup covering first-definition-wins dedup

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: address 4 new PR #331 review comments

- pipeline_step_workflow_call.go: propagate stop_pipeline in async mode
- integrity.go: "open plugin binary" instead of "read" for os.Open error
- plugin_install.go: lowercase "warning:" for consistency
- merge_test.go: use filepath.Join and check writeFileContent errors

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: intel352

cmd/wfctl/multi_registry.go

@@ -29,12 +29,12 @@ func NewMultiRegistry(cfg *RegistryConfig) *MultiRegistry {
 		case "github":
 			sources = append(sources, NewGitHubRegistrySource(sc))
 		case "static":
-			staticSrc, staticErr := NewStaticRegistrySource(sc)
-			if staticErr != nil {
-				fmt.Fprintf(os.Stderr, "warning: %v, skipping\n", staticErr)
+			src, err := NewStaticRegistrySource(sc)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "warning: %v, skipping\n", err)
 				continue
 			}
-			sources = append(sources, staticSrc)
+			sources = append(sources, src)
 		default:
 			// Skip unknown types
 			fmt.Fprintf(os.Stderr, "warning: unknown registry type %q for %q, skipping\n", sc.Type, sc.Name)

cmd/wfctl/plugin_install.go

@@ -81,19 +81,19 @@ func runPluginInstall(args []string) error {
 		return err
 	}
 
-	// Enforce mutual exclusivity: at most one of --url, --local, or positional args.
-	exclusiveCount := 0
+	// Validate mutual exclusivity of install modes.
+	modes := 0
 	if *directURL != "" {
-		exclusiveCount++
+		modes++
 	}
 	if *localPath != "" {
-		exclusiveCount++
+		modes++
 	}
 	if fs.NArg() > 0 {
-		exclusiveCount++
+		modes++
 	}
-	if exclusiveCount > 1 {
-		return fmt.Errorf("--url, --local, and <name> are mutually exclusive; specify only one")
+	if modes > 1 {
+		return fmt.Errorf("specify only one of: <name>, --url, or --local")
 	}
 
 	if *directURL != "" {
@@ -165,13 +165,15 @@ func runPluginInstall(args []string) error {
 
 	// Update .wfctl.yaml lockfile if name@version was provided.
 	if _, ver := parseNameVersion(nameArg); ver != "" {
-		// Hash the installed binary (not the archive) so verifyInstalledChecksum matches.
+		pluginName = normalizePluginName(pluginName)
+		binaryChecksum := ""
 		binaryPath := filepath.Join(pluginDirVal, pluginName, pluginName)
-		sha, hashErr := hashFileSHA256(binaryPath)
-		if hashErr != nil {
-			fmt.Fprintf(os.Stderr, "warning: could not hash installed binary: %v\n", hashErr)
+		if cs, hashErr := hashFileSHA256(binaryPath); hashErr == nil {
+			binaryChecksum = cs
+		} else {
+			fmt.Fprintf(os.Stderr, "warning: could not hash binary %s: %v (lockfile will have no checksum)\n", binaryPath, hashErr)
 		}
-		updateLockfileWithChecksum(pluginName, manifest.Version, manifest.Repository, sourceName, sha)
+		updateLockfileWithChecksum(pluginName, manifest.Version, manifest.Repository, sourceName, binaryChecksum)
 	}
 
 	return nil
@@ -516,7 +518,7 @@ func installFromURL(url, pluginDir string) error {
 	}
 
 	if err := ensurePluginBinary(destDir, pluginName); err != nil {
-		return fmt.Errorf("normalize binary name: %w", err)
+		return fmt.Errorf("could not normalize binary name: %w", err)
 	}
 
 	// Validate the installed plugin (same checks as registry installs).
@@ -536,6 +538,20 @@ func installFromURL(url, pluginDir string) error {
 	return nil
 }
 
+// hashFileSHA256 computes the SHA-256 hex digest of the file at path using streaming I/O.
+func hashFileSHA256(path string) (string, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return "", fmt.Errorf("hash file %s: %w", path, err)
+	}
+	defer f.Close()
+	h := sha256.New()
+	if _, err := io.Copy(h, f); err != nil {
+		return "", fmt.Errorf("hash file %s: %w", path, err)
+	}
+	return hex.EncodeToString(h.Sum(nil)), nil
+}
+
 // verifyInstalledChecksum reads the plugin binary and verifies its SHA-256 checksum.
 func verifyInstalledChecksum(pluginDir, pluginName, expectedSHA256 string) error {
 	binaryPath := filepath.Join(pluginDir, pluginName)
@@ -590,13 +606,11 @@ func installFromLocal(srcDir, pluginDir string) error {
 		return err
 	}
 
-	// Update lockfile with binary checksum for consistency with other install paths.
-	installedBinary := filepath.Join(destDir, pluginName)
-	sha, hashErr := hashFileSHA256(installedBinary)
+	binaryChecksum, hashErr := hashFileSHA256(filepath.Join(destDir, pluginName))
 	if hashErr != nil {
-		fmt.Fprintf(os.Stderr, "warning: could not hash installed binary: %v\n", hashErr)
+		fmt.Fprintf(os.Stderr, "warning: could not compute binary checksum: %v\n", hashErr)
 	}
-	updateLockfileWithChecksum(pluginName, pj.Version, "", "", sha)
+	updateLockfileWithChecksum(pluginName, pj.Version, "", "", binaryChecksum)
 
 	fmt.Printf("Installed %s v%s from %s to %s\n", pluginName, pj.Version, srcDir, destDir)
 	return nil
@@ -693,16 +707,6 @@ func parseGitHubRepoURL(repoURL string) (owner, repo string, err error) {
 	return parts[1], repoName, nil
 }
 
-// hashFileSHA256 returns the hex-encoded SHA-256 hash of the file at path.
-func hashFileSHA256(path string) (string, error) {
-	data, err := os.ReadFile(path)
-	if err != nil {
-		return "", fmt.Errorf("hash file %s: %w", path, err)
-	}
-	h := sha256.Sum256(data)
-	return hex.EncodeToString(h[:]), nil
-}
-
 // extractTarGz decompresses and extracts a .tar.gz archive into destDir.
 // It guards against path traversal (zip-slip) attacks.
 func extractTarGz(data []byte, destDir string) error {

cmd/wfctl/plugin_lockfile.go

@@ -90,8 +90,10 @@ func installFromLockfile(pluginDir, cfgPath string) error {
 		if entry.SHA256 != "" {
 			pluginInstallDir := filepath.Join(pluginDir, name)
 			if verifyErr := verifyInstalledChecksum(pluginInstallDir, name, entry.SHA256); verifyErr != nil {
-				fmt.Fprintf(os.Stderr, "CHECKSUM MISMATCH for %s: %v — removing plugin\n", name, verifyErr)
-				_ = os.RemoveAll(pluginInstallDir)
+				fmt.Fprintf(os.Stderr, "CHECKSUM MISMATCH for %s: %v\n", name, verifyErr)
+				if removeErr := os.RemoveAll(pluginInstallDir); removeErr != nil {
+					fmt.Fprintf(os.Stderr, "warning: could not remove plugin dir: %v\n", removeErr)
+				}
 				failed = append(failed, name)
 				continue
 			}

cmd/wfctl/registry_source.go

@@ -64,7 +64,7 @@ func (g *GitHubRegistrySource) ListPlugins() ([]string, error) {
 		req.Header.Set("Authorization", "Bearer "+token)
 	}
 
-	resp, err := http.DefaultClient.Do(req)
+	resp, err := registryHTTPClient.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("list registry plugins from %s: %w", g.name, err)
 	}
@@ -90,7 +90,11 @@ func (g *GitHubRegistrySource) FetchManifest(name string) (*RegistryManifest, er
 		"https://raw.githubusercontent.com/%s/%s/%s/plugins/%s/manifest.json",
 		g.owner, g.repo, g.branch, name,
 	)
-	resp, err := http.Get(url) //nolint:gosec // URL constructed from configured registry
+	req, err := http.NewRequest(http.MethodGet, url, nil) //nolint:gosec // URL constructed from configured registry
+	if err != nil {
+		return nil, fmt.Errorf("build request: %w", err)
+	}
+	resp, err := registryHTTPClient.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("fetch manifest for %q from %s: %w", name, g.name, err)
 	}
@@ -206,8 +210,13 @@ func (s *StaticRegistrySource) SearchPlugins(query string) ([]PluginSearchResult
 			strings.Contains(strings.ToLower(e.Name), q) ||
 			strings.Contains(strings.ToLower(e.Description), q) {
 			results = append(results, PluginSearchResult{
-				PluginSummary: PluginSummary(e),
-				Source:        s.name,
+				PluginSummary: PluginSummary{ //nolint:staticcheck // S1016: explicit fields for clarity across struct tag boundaries
+					Name:        e.Name,
+					Version:     e.Version,
+					Description: e.Description,
+					Tier:        e.Tier,
+				},
+				Source: s.name,
 			})
 		}
 	}
@@ -226,7 +235,8 @@ func (s *StaticRegistrySource) ListPlugins() ([]string, error) {
 	return names, nil
 }
 
-// registryHTTPClient is used for all registry HTTP requests with a reasonable timeout.
+// registryHTTPClient is used for all registry HTTP requests (both GitHub and static
+// sources) with a reasonable timeout to avoid hangs on network issues.
 var registryHTTPClient = &http.Client{Timeout: 30 * time.Second}
 
 // fetch performs an HTTP GET with optional auth token.

config/config.go

@@ -419,6 +419,25 @@ func MergeApplicationConfig(appCfg *ApplicationConfig) (*WorkflowConfig, error)
 		}
 
 		combined.Modules = append(combined.Modules, wfCfg.Modules...)
+
+		// Merge external plugin declarations — deduplicate by name (first definition wins).
+		if wfCfg.Plugins != nil && len(wfCfg.Plugins.External) > 0 {
+			if combined.Plugins == nil {
+				combined.Plugins = &PluginsConfig{}
+			}
+			existingPlugins := make(map[string]struct{}, len(combined.Plugins.External))
+			for _, ep := range combined.Plugins.External {
+				existingPlugins[ep.Name] = struct{}{}
+			}
+			for _, ep := range wfCfg.Plugins.External {
+				if _, exists := existingPlugins[ep.Name]; exists {
+					continue
+				}
+				combined.Plugins.External = append(combined.Plugins.External, ep)
+				existingPlugins[ep.Name] = struct{}{}
+			}
+		}
+
 		for k, v := range wfCfg.Workflows {
 			if existing, exists := combined.Workflows[k]; exists {
 				// If the existing value is nil (e.g. `http:` with no body in YAML),

config/merge_test.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"os"
+	"path/filepath"
 	"reflect"
 	"strings"
 	"testing"
@@ -533,3 +534,68 @@ func writeFileContent(path, content string) error {
 func contains(s, substr string) bool {
 	return strings.Contains(s, substr)
 }
+
+func TestMergeApplicationConfig_PluginDedup(t *testing.T) {
+	dir := t.TempDir()
+
+	// Workflow A declares plugin "foo"
+	wfA := filepath.Join(dir, "a.yaml")
+	if err := writeFileContent(wfA, `
+plugins:
+  external:
+    - name: foo
+      version: "1.0"
+      repository: "https://example.com/foo"
+modules: []
+`); err != nil {
+		t.Fatalf("write a.yaml: %v", err)
+	}
+
+	// Workflow B declares plugin "foo" (duplicate) and "bar"
+	wfB := filepath.Join(dir, "b.yaml")
+	if err := writeFileContent(wfB, `
+plugins:
+  external:
+    - name: foo
+      version: "2.0"
+      repository: "https://example.com/foo-v2"
+    - name: bar
+      version: "1.0"
+      repository: "https://example.com/bar"
+modules: []
+`); err != nil {
+		t.Fatalf("write b.yaml: %v", err)
+	}
+
+	appCfg := &ApplicationConfig{
+		Application: ApplicationInfo{
+			Workflows: []WorkflowRef{
+				{File: wfA},
+				{File: wfB},
+			},
+		},
+	}
+
+	cfg, err := MergeApplicationConfig(appCfg)
+	if err != nil {
+		t.Fatalf("MergeApplicationConfig: %v", err)
+	}
+
+	if cfg.Plugins == nil {
+		t.Fatal("expected Plugins to be non-nil after merge")
+	}
+	if len(cfg.Plugins.External) != 2 {
+		t.Fatalf("expected 2 plugins (foo + bar), got %d", len(cfg.Plugins.External))
+	}
+
+	// First definition wins — foo should have version "1.0" from workflow A
+	var fooVer string
+	for _, p := range cfg.Plugins.External {
+		if p.Name == "foo" {
+			fooVer = p.Version
+		}
+	}
+	if fooVer != "1.0" {
+		t.Errorf("expected foo version 1.0 (first definition wins), got %s", fooVer)
+	}
+}

docs/PLUGIN_AUTHORING.md

@@ -101,7 +101,17 @@ func (p *Provider) CreateModule(typeName, name string, config map[string]any) (s
 
 ## Plugin Manifest
 
-The `plugin.json` declares what your plugin provides. The name should match what you passed to `wfctl plugin init`:
+The `plugin.json` at the project root declares what your plugin provides. The `name`
+field **must match the short name** you passed to `wfctl plugin init` (e.g. `my-plugin`).
+This is the name used by the engine for plugin discovery, the `requires.plugins` dependency
+check, and `wfctl plugin install`.
+
+> **Note:** The scaffolded `internal/provider.go` returns a manifest with the name prefixed
+> as `workflow-plugin-<short-name>` (e.g. `workflow-plugin-my-plugin`). That longer form is
+> the canonical name used in the **public registry** (`workflow-registry`) and in release
+> artifact URLs. When referencing your plugin in a workflow config's `requires.plugins` or
+> `plugins.external`, use the same short name you put in `plugin.json` — the engine resolves
+> both forms automatically.
 
 ```json
 {

module/pipeline_step_workflow_call.go

@@ -31,6 +31,7 @@ type WorkflowCallStep struct {
 	name          string
 	workflow      string           // target pipeline name
 	mode          WorkflowCallMode // "sync" (default) or "async"
+	stopPipeline  bool             // if true, stop parent pipeline after this step completes
 	inputMapping  map[string]string
 	outputMapping map[string]string
 	timeout       time.Duration
@@ -86,6 +87,10 @@ func NewWorkflowCallStepFactory(lookup PipelineLookupFn) StepFactory {
 			}
 		}
 
+		if v, ok := cfg["stop_pipeline"].(bool); ok {
+			step.stopPipeline = v
+		}
+
 		return step, nil
 	}
 }
@@ -101,9 +106,13 @@ func (s *WorkflowCallStep) Execute(ctx context.Context, pc *PipelineContext) (*S
 	if s.lookup == nil {
 		return nil, fmt.Errorf("workflow_call step %q: no pipeline lookup function configured", s.name)
 	}
-	target, ok := s.lookup(s.workflow)
+	workflowName, resolveErr := s.tmpl.Resolve(s.workflow, pc)
+	if resolveErr != nil {
+		return nil, fmt.Errorf("workflow_call step %q: failed to resolve workflow name %q: %w", s.name, s.workflow, resolveErr)
+	}
+	target, ok := s.lookup(workflowName)
 	if !ok {
-		return nil, fmt.Errorf("workflow_call step %q: pipeline %q not found — ensure it is defined in the application config", s.name, s.workflow)
+		return nil, fmt.Errorf("workflow_call step %q: pipeline %q not found — ensure it is defined in the application config", s.name, workflowName)
 	}
 
 	// Build trigger data from input mapping or fall back to passing all current data
@@ -130,7 +139,7 @@ func (s *WorkflowCallStep) Execute(ctx context.Context, pc *PipelineContext) (*S
 			defer cancel()
 			_, _ = target.Execute(asyncCtx, data) //nolint:errcheck
 		}(ctx, triggerData)
-		return &StepResult{Output: map[string]any{"workflow": s.workflow, "mode": "async", "dispatched": true}}, nil
+		return &StepResult{Output: map[string]any{"workflow": workflowName, "mode": "async", "dispatched": true}, Stop: s.stopPipeline}, nil
 	}
 
 	// Sync mode: apply timeout and wait for result
@@ -139,7 +148,7 @@ func (s *WorkflowCallStep) Execute(ctx context.Context, pc *PipelineContext) (*S
 
 	childCtx, err := target.Execute(syncCtx, triggerData)
 	if err != nil {
-		return nil, fmt.Errorf("workflow_call step %q: workflow %q failed: %w", s.name, s.workflow, err)
+		return nil, fmt.Errorf("workflow_call step %q: workflow %q failed: %w", s.name, workflowName, err)
 	}
 
 	// Map outputs back to parent context
@@ -153,7 +162,7 @@ func (s *WorkflowCallStep) Execute(ctx context.Context, pc *PipelineContext) (*S
 		output["result"] = childCtx.Current
 	}
 
-	return &StepResult{Output: output}, nil
+	return &StepResult{Output: output, Stop: s.stopPipeline}, nil
 }
 
 // Ensure interface satisfaction at compile time.