feat(wfctl): bounded pipeline records and registry reports

cmd/wfctl/local_external_plugins.go

@@ -5,10 +5,12 @@ import (
 	"io"
 	"log"
 	"log/slog"
+	"os"
 	"sort"
 	"strings"
 
 	"github.com/GoCodeAlone/workflow"
+	"github.com/GoCodeAlone/workflow/plugin"
 	pluginexternal "github.com/GoCodeAlone/workflow/plugin/external"
 )
 
@@ -56,6 +58,53 @@ func newLocalExternalPluginStdLogger(logger *slog.Logger) *log.Logger {
 	return log.New(localExternalPluginLogWriter{logger: logger}, "", 0)
 }
 
+func loadPipelineExternalPlugins(eng *workflow.StdEngine, pluginDir string, names []string, logger *slog.Logger) (func(), error) {
+	installed, err := inspectPipelineInstallations(pluginDir)
+	if err != nil {
+		return nil, err
+	}
+	manager := pluginexternal.NewExternalPluginManager(pluginDir, newLocalExternalPluginStdLogger(logger))
+	manager.SetProcessOutput(os.Stdout, os.Stderr)
+	for _, name := range names {
+		adapter, err := manager.LoadPlugin(name)
+		if err != nil {
+			manager.Shutdown()
+			return nil, fmt.Errorf("load selected external plugin %q: %w", name, err)
+		}
+		if adapter.EngineManifest().Name != name {
+			manager.Shutdown()
+			return nil, fmt.Errorf("external plugin runtime installation identity mismatch")
+		}
+		if len(adapter.ConfigFragmentBytes()) != 0 {
+			manager.Shutdown()
+			return nil, fmt.Errorf("record mode rejects external config fragments")
+		}
+		factories := adapter.StepFactories()
+		if err := validatePipelineRuntimeTypes(installed.manifests[name], factories); err != nil {
+			manager.Shutdown()
+			return nil, err
+		}
+		if err := eng.LoadPlugin(adapter); err != nil {
+			manager.Shutdown()
+			return nil, fmt.Errorf("register selected external plugin: %w", err)
+		}
+		registerPipelineFailureFactories(eng, factories)
+	}
+	return manager.Shutdown, nil
+}
+
+func validatePipelineRuntimeTypes(manifest *plugin.PluginManifest, factories map[string]plugin.StepFactory) error {
+	if manifest == nil || len(factories) != len(manifest.StepTypes) {
+		return fmt.Errorf("external runtime step ownership differs from installed manifest")
+	}
+	for _, name := range manifest.StepTypes {
+		if _, exists := factories[name]; !exists {
+			return fmt.Errorf("external runtime step ownership differs from installed manifest")
+		}
+	}
+	return nil
+}
+
 type localExternalPluginLogWriter struct {
 	logger *slog.Logger
 }

cmd/wfctl/main.go

@@ -164,6 +164,9 @@ var commands = map[string]func([]string) error{
 }
 
 func main() {
+	if handled, code := pipelineRecordEarlyDispatch(os.Args[1:]); handled {
+		os.Exit(code)
+	}
 	// Load the embedded config. All command definitions and pipeline wiring
 	// live in wfctl.yaml — no hardcoded routing in this file.
 	cfg, err := config.LoadFromBytes(wfctlConfigBytes)

cmd/wfctl/main_test.go

@@ -30,6 +30,9 @@ import (
 // are exercised by platform/differ_cache_test.go via its own
 // in-memory swap.
 func TestMain(m *testing.M) {
+	if len(os.Args) == 2 && os.Args[1] == pipelineRecordChildCommand {
+		os.Exit(runPipelineRecordChild())
+	}
 	if err := os.Setenv("WFCTL_DIFFCACHE", "disabled"); err != nil {
 		panic("setenv WFCTL_DIFFCACHE: " + err.Error())
 	}

cmd/wfctl/pipeline.go

@@ -113,6 +113,10 @@ func runPipelineRun(args []string) error {
 	pluginDir := fs.String("plugin-dir", "", "Directory containing installed external plugins")
 	inputJSON := fs.String("input", "", "Input data as JSON object")
 	verbose := fs.Bool("verbose", false, "Show detailed step output")
+	outputMode := fs.String("output", "human", "Output mode: human or record")
+	resultStep := fs.String("result-step", "", "Literal reachable result step for record output")
+	recordPrefix := fs.String("record-prefix", "", "Success record prefix")
+	errorPrefix := fs.String("error-prefix", "", "Error record prefix")
 	var vars stringSliceFlag
 	fs.Var(&vars, "var", "Variable in key=value format (repeatable)")
 	fs.Usage = func() {
@@ -144,13 +148,26 @@ Options:
 	if *pluginDir == "" {
 		*pluginDir = strings.TrimSpace(os.Getenv("WFCTL_PLUGIN_DIR"))
 	}
+	if *outputMode != "human" && *outputMode != "record" {
+		return fmt.Errorf("--output must be human or record")
+	}
+	if *outputMode == "record" {
+		if err := validatePipelineRecordSelectors(*resultStep, *recordPrefix, *errorPrefix); err != nil {
+			return err
+		}
+	} else if *resultStep != "" || *recordPrefix != "" || *errorPrefix != "" {
+		return fmt.Errorf("record selectors require --output record")
+	}
 
 	// Build initial trigger data from --input JSON
 	triggerData := make(map[string]any)
 	if *inputJSON != "" {
 		if err := json.Unmarshal([]byte(*inputJSON), &triggerData); err != nil {
 			return fmt.Errorf("invalid --input JSON: %w", err)
 		}
+		if triggerData == nil {
+			return fmt.Errorf("invalid --input JSON: expected an object")
+		}
 	}
 
 	// Inject --var entries into trigger data
@@ -161,6 +178,11 @@ Options:
 		}
 		triggerData[kv[:idx]] = kv[idx+1:]
 	}
+	if *outputMode == "record" {
+		return runPipelineRecord(pipelineRecordRequest{Pipeline: *pipelineName, ResultStep: *resultStep,
+			PluginDir: *pluginDir, Input: triggerData}, pipelineRecordOptions{ConfigPath: *configPath,
+			SuccessPrefix: *recordPrefix, ErrorPrefix: *errorPrefix})
+	}
 
 	// Load config
 	cfg, err := config.LoadFromFile(*configPath)
@@ -182,6 +204,15 @@ Options:
 	}
 
 	// Set up a logger — suppress engine noise unless --verbose
+	closure, err := selectPipelineClosure(cfg, *pipelineName, false)
+	if err != nil {
+		return err
+	}
+	if _, err := inspectPipelineInstallations(*pluginDir); err != nil {
+		return err
+	}
+	cfg = closure.config
+
 	logLevel := slog.LevelError
 	if *verbose {
 		logLevel = slog.LevelDebug
@@ -289,6 +320,9 @@ func (ps *progressStep) Execute(ctx context.Context, pc *module.PipelineContext)
 	fmt.Printf("Step %d/%d: %s ... ", ps.index+1, ps.total, ps.inner.Name())
 
 	result, err := ps.inner.Execute(ctx, pc)
+	if err == nil {
+		err = stoppedPipelineError(result)
+	}
 	elapsed := time.Since(start)
 
 	if err != nil {

cmd/wfctl/pipeline_broker_host_test.go

@@ -0,0 +1,50 @@
+//go:build linux || darwin
+
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"slices"
+	"strings"
+	"testing"
+)
+
+func TestPipelineRecordHostSandboxBroker(t *testing.T) {
+	root := pipelineRecordHostTempDir(t)
+	binary := filepath.Join(root, "wfctl")
+	buildPipelineRecordHostBinary(t, binary, ".")
+	// This substitutes container transport only. The actual Workflow sandbox
+	// step, templates, WFD2 channel, journal, and parent Docker adapter execute.
+	dir, _ := newPipelineDockerFixture(t)
+	t.Setenv("XDG_STATE_HOME", filepath.Join(root, "state"))
+	t.Setenv("GITHUB_TOKEN", "host-private-canary")
+	stdout, stderr, err := runPipelineRecordHost(t, binary, "testdata/pipeline-record/sandbox.yaml", "", "--var", "message=dynamic-message")
+	if err != nil || string(stdout) != "SDK_RECORD_V1 {\"message\":\"captured output\",\"source\":\"broker\"}\n" {
+		t.Fatalf("Workflow sandbox broker path failed: %v\nstdout=%s\nstderr=%s", err, stdout, stderr)
+	}
+	calls := pipelineDockerCalls(t, dir)
+	var created bool
+	for _, call := range calls {
+		if !slices.Contains(call.Args, "create") {
+			continue
+		}
+		created = true
+		args := strings.Join(call.Args, " ")
+		for _, value := range []string{"--user 65532:65532", "--read-only", "--cap-drop ALL", "--security-opt no-new-privileges", "--pids-limit 64", "--cpus 1", "PUBLIC_VALUE=dynamic-message", "printf '%s' 'dynamic-message'"} {
+			if !strings.Contains(args, value) {
+				t.Errorf("broker omitted strict or resolved argument %q: %q", value, args)
+			}
+		}
+		if strings.Contains(args, "host-private-canary") || strings.Contains(args, "GITHUB_TOKEN") {
+			t.Fatal("host token entered the container request")
+		}
+	}
+	if !created {
+		t.Fatal("sandbox transport never received create")
+	}
+	entries, err := os.ReadDir(filepath.Join(root, "state", "wfctl", "pipeline-cleanup"))
+	if err != nil || len(entries) != 1 || entries[0].Name() != "lock" {
+		t.Fatalf("completed sandbox left retained journal state: %v, %v", entries, err)
+	}
+}