From d3ba0d9b38ffb483669f24a9511c1afeef8deaf1 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Tue, 30 Jun 2026 18:02:05 -0400 Subject: [PATCH 1/2] feat: add compute chain step --- README.md | 120 +++++++++++ go.mod | 2 +- go.sum | 6 +- internal/plugin.go | 3 + internal/plugin_test.go | 2 +- internal/steps_chain.go | 361 ++++++++++++++++++++++++++++++++++ internal/steps_chain_test.go | 243 +++++++++++++++++++++++ internal/steps_stream_test.go | 2 +- internal/steps_test.go | 2 +- plugin.json | 6 +- 10 files changed, 736 insertions(+), 11 deletions(-) create mode 100644 internal/steps_chain.go create mode 100644 internal/steps_chain_test.go diff --git a/README.md b/README.md index ff07e14..00602fd 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,10 @@ Examples: `workflow-plugin-compute-core/protocol.ProviderContract`; this plugin submits or waits on the resulting generic workflow-compute task without embedding provider business logic. +- A content workflow submits a generic provider chain that fetches media + through a storage provider, transforms it through a media provider, and then + forwards artifact/content/stream references to the next provider without + embedding storage or media semantics in this plugin. - A live-video workflow submits `step.compute_stream` to create a `video-stream` task, while `workflow-plugin-stream` owns the MediaMTX provider contract, runtime adapter, ingest descriptor, auth hook, and stream proof @@ -127,6 +131,122 @@ For fanout work, use `step.compute_map` with a deterministic `tasks` list. The step submits every task, polls the core task/proof APIs, and stops the Workflow pipeline if any task fails, stalls, times out, or produces a non-accepted proof. +## Composable Provider Chains + +`step.compute_chain` submits an ordered list of workflow-compute tasks and waits +for each task by default. It is for workflows where one provider step produces a +proof preview that later provider steps consume, such as S3 content fetch, +media transform, and stream publish. + +Each chain entry has an explicit `id`, workload routing fields, optional +`depends_on`, optional `wait: false`, and optional `input_mappings`. Mappings +only copy prior step output into generic provider handoff fields: + +- `provider.artifact_refs` +- `provider.content_inputs` +- `provider.stream_inputs` + +The compute plugin does not fetch S3 objects, run ffmpeg, mint stream tokens, +or implement provider SDKs. S3 behavior belongs in `workflow-plugin-aws`, media +behavior belongs in `workflow-plugin-media`, stream transport belongs in +`workflow-plugin-stream`, and the wfcompute runtime enforces task/proof +authorization for forwarded refs. Secrets and credentials must remain scoped +refs such as `secret://...`, `content://...`, `stream://...`, or +`artifact://...`; raw AWS keys, publish tokens, and bearer credentials do not +belong in chain config. + +Example: + +```yaml +steps: + transcode_and_publish: + type: step.compute_chain + config: + server_url: https://compute.example.com + auth_token_ref: secret:WFCOMPUTE_TOKEN + poll_interval: 2s + timeout: 30m + require_proof: true + steps: + - id: fetch_media + task_id: fetch-media-1 + org_id: gocodealone + pool_id: media + policy_id: content-fetch + timeout_seconds: 300 + workload: + kind: provider + provider: + provider_config: + plugin_id: workflow-plugin-aws + provider_id: s3-content-source + contract_id: workflow-plugin-aws.s3-content-source.v1 + version: v1.0.0 + config_ref: config://providers/aws/media + operation: fetch + image_ref: ghcr.io/gocodealone/workflow-plugin-aws-provider@sha256:... + input: + bucket_ref: config://media/source-bucket + object_key: inputs/source.mp4 + + - id: transcode + task_id: transcode-1 + org_id: gocodealone + pool_id: media + policy_id: media-transform + timeout_seconds: 1800 + depends_on: + - fetch_media + input_mappings: + - from_step: fetch_media + from: result_preview.content_inputs + to: provider.content_inputs + workload: + kind: provider + provider: + provider_config: + plugin_id: workflow-plugin-media + provider_id: media-batch-transform + contract_id: workflow-plugin-media.batch-transform.v1 + version: v1.0.0 + config_ref: config://providers/media/ffmpeg + operation: batch_transform + image_ref: ghcr.io/gocodealone/workflow-plugin-media-provider@sha256:... + input: + outputs: + - name: hls_720p + preset: hls-720p + + - id: publish_stream + task_id: publish-stream-1 + org_id: gocodealone + pool_id: streamers + policy_id: stream-publish + timeout_seconds: 600 + depends_on: + - transcode + input_mappings: + - from_step: transcode + from: result_preview.artifact_refs + to: provider.artifact_refs + - from_step: transcode + from: result_preview.stream_inputs + to: provider.stream_inputs + workload: + kind: provider + provider: + provider_config: + plugin_id: workflow-plugin-stream + provider_id: mediamtx + contract_id: workflow-plugin-stream.video-stream.v1 + version: v1.0.0 + config_ref: config://providers/stream/main + operation: publish + image_ref: ghcr.io/gocodealone/workflow-plugin-stream-provider@sha256:... + input: + rendition: 720p +``` + ## Stream Workloads `step.compute_stream` submits a `workflow-plugin-compute-core` `video-stream` diff --git a/go.mod b/go.mod index c7fd0da..8134ef9 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.26.3 require ( github.com/GoCodeAlone/workflow v0.64.0 github.com/GoCodeAlone/workflow-compute v0.0.0-20260611013203-3a1a07f1da28 - github.com/GoCodeAlone/workflow-plugin-compute-core v0.8.1 + github.com/GoCodeAlone/workflow-plugin-compute-core v0.8.3 ) require ( diff --git a/go.sum b/go.sum index 6c2d4f3..cbf0ab9 100644 --- a/go.sum +++ b/go.sum @@ -16,10 +16,8 @@ github.com/GoCodeAlone/workflow v0.64.0 h1:2CpbYPwIqdGDb3xi3YJpwcteIum4ehBSrnRql github.com/GoCodeAlone/workflow v0.64.0/go.mod h1:659GGDrw3QJ7b625y9rf8QhKIpt1VCoEG0MxKu5tGQs= github.com/GoCodeAlone/workflow-compute v0.0.0-20260611013203-3a1a07f1da28 h1:aZSKBNYl8Caa8g6s97CGXcY6fYkAwGZfr7qY5NLWe3o= github.com/GoCodeAlone/workflow-compute v0.0.0-20260611013203-3a1a07f1da28/go.mod h1:wNbq90VefaFm2XC7GBGSm1S3ESML+JJgH4gdbO+zN64= -github.com/GoCodeAlone/workflow-plugin-compute-core v0.6.0 h1:khmZdWjqybwuXlUHmxDIP3Ce6ce6F77QXXvV4HAox7w= -github.com/GoCodeAlone/workflow-plugin-compute-core v0.6.0/go.mod h1:1T6uCpUWPCNk6XPYgKq5CL/7LkD24MphKYsYVzF4jnI= -github.com/GoCodeAlone/workflow-plugin-compute-core v0.8.1 h1:K/Y1ZZbDG3yhNBThYnZ4G2vhixBEaUosLxUdJg+lwlI= -github.com/GoCodeAlone/workflow-plugin-compute-core v0.8.1/go.mod h1:Dsw3miQN8eenpDaLzo/p0TolGtI7NyETkf9abAHZKSM= +github.com/GoCodeAlone/workflow-plugin-compute-core v0.8.3 h1:CIu6hIUXJx6PbmTq1xP9RKNuXJ9kBsIiJQoRuBvyj2U= +github.com/GoCodeAlone/workflow-plugin-compute-core v0.8.3/go.mod h1:Dsw3miQN8eenpDaLzo/p0TolGtI7NyETkf9abAHZKSM= github.com/GoCodeAlone/yaegi v0.17.2 h1:WK6Y6e0t1a6U7r+S2dN3CGWW1PizYD3zO0zneToZPxM= github.com/GoCodeAlone/yaegi v0.17.2/go.mod h1:z5Pr6Wse6QJcQvpgxTxzMAevFarH0N37TG88Y9dprx0= github.com/IBM/sarama v1.47.0 h1:GcQFEd12+KzfPYeLgN69Fh7vLCtYRhVIx0rO4TZO318= diff --git a/internal/plugin.go b/internal/plugin.go index 1173b74..8eddfbb 100644 --- a/internal/plugin.go +++ b/internal/plugin.go @@ -50,6 +50,7 @@ func (p *computePlugin) StepTypes() []string { "step.compute_wait", "step.compute_map", "step.compute_stream", + "step.compute_chain", } } @@ -63,6 +64,8 @@ func (p *computePlugin) CreateStep(typeName, name string, config map[string]any) return newMapStep(name, config) case "step.compute_stream": return newComputeStreamStep(name, config) + case "step.compute_chain": + return newComputeChainStep(name, config) default: return nil, fmt.Errorf("compute plugin: unknown step type %q", typeName) } diff --git a/internal/plugin_test.go b/internal/plugin_test.go index 7957d36..7b2732b 100644 --- a/internal/plugin_test.go +++ b/internal/plugin_test.go @@ -55,7 +55,7 @@ func TestT545_PluginManifestDeclaresManagedRuntimeDependencies(t *testing.T) { t.Fatalf("decode plugin.json: %v", err) } want := map[string]string{ - "workflow-plugin-compute-core": ">=0.8.1", + "workflow-plugin-compute-core": ">=0.8.3", "workflow-plugin-compute-container": ">=0.5.1", } for _, dependency := range manifest.Dependencies { diff --git a/internal/steps_chain.go b/internal/steps_chain.go new file mode 100644 index 0000000..e8c0339 --- /dev/null +++ b/internal/steps_chain.go @@ -0,0 +1,361 @@ +package internal + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/GoCodeAlone/workflow-compute/pkg/protocol" + coreprotocol "github.com/GoCodeAlone/workflow-plugin-compute-core/protocol" + sdk "github.com/GoCodeAlone/workflow/plugin/external/sdk" +) + +type computeChainConfig struct { + connectionConfig + Steps []computeChainTaskConfig `json:"steps"` + PollInterval string `json:"poll_interval,omitempty"` + Timeout string `json:"timeout,omitempty"` + RequireProof *bool `json:"require_proof,omitempty"` +} + +type computeChainTaskConfig struct { + ID string `json:"id"` + TaskID string `json:"task_id,omitempty"` + ProductID string `json:"product_id,omitempty"` + OrgID string `json:"org_id"` + PoolID string `json:"pool_id"` + PolicyID string `json:"policy_id"` + TimeoutSeconds int `json:"timeout_seconds"` + Labels map[string]string `json:"labels,omitempty"` + ResiduePolicy protocol.ResiduePolicy `json:"residue_policy,omitzero"` + DependsOn []string `json:"depends_on,omitempty"` + InputMappings []computeChainInputMapping `json:"input_mappings,omitempty"` + Wait *bool `json:"wait,omitempty"` + Workload protocol.WorkloadSpec `json:"workload"` +} + +type computeChainInputMapping struct { + FromStep string `json:"from_step"` + From string `json:"from"` + To string `json:"to"` +} + +type computeChainStep struct { + name string + config computeChainConfig +} + +func newComputeChainStep(name string, raw map[string]any) (*computeChainStep, error) { + var cfg computeChainConfig + if err := decodeStrictMap(raw, &cfg); err != nil { + return nil, fmt.Errorf("step.compute_chain %q: %w", name, err) + } + if err := cfg.validate(name); err != nil { + return nil, err + } + return &computeChainStep{name: name, config: cfg}, nil +} + +func (c computeChainConfig) validate(name string) error { + if err := c.connectionConfig.validate(); err != nil { + return fmt.Errorf("step.compute_chain %q: %w", name, err) + } + if len(c.Steps) == 0 { + return fmt.Errorf("step.compute_chain %q: steps is required", name) + } + if c.PollInterval != "" { + if d, err := time.ParseDuration(c.PollInterval); err != nil { + return fmt.Errorf("step.compute_chain %q: poll_interval must be duration: %w", name, err) + } else if d <= 0 { + return fmt.Errorf("step.compute_chain %q: poll_interval must be positive", name) + } + } + if c.Timeout != "" { + if d, err := time.ParseDuration(c.Timeout); err != nil { + return fmt.Errorf("step.compute_chain %q: timeout must be duration: %w", name, err) + } else if d <= 0 { + return fmt.Errorf("step.compute_chain %q: timeout must be positive", name) + } + } + seen := make(map[string]struct{}, len(c.Steps)) + for i, step := range c.Steps { + if step.ID == "" { + return fmt.Errorf("step.compute_chain %q: steps[%d].id is required", name, i) + } + if _, ok := seen[step.ID]; ok { + return fmt.Errorf("step.compute_chain %q: duplicate step id %q", name, step.ID) + } + taskCfg := step.taskConfig() + if err := taskCfg.validate(); err != nil { + return fmt.Errorf("step.compute_chain %q: steps[%d] %q: %w", name, i, step.ID, err) + } + if err := step.Workload.Validate(); err != nil { + return fmt.Errorf("step.compute_chain %q: steps[%d] %q workload: %w", name, i, step.ID, err) + } + for _, dep := range step.DependsOn { + if _, ok := seen[dep]; !ok { + return fmt.Errorf("step.compute_chain %q: steps[%d] %q depends_on %q must reference an earlier step", name, i, step.ID, dep) + } + } + deps := stringSet(step.DependsOn) + for j, mapping := range step.InputMappings { + if err := mapping.validate(); err != nil { + return fmt.Errorf("step.compute_chain %q: steps[%d] %q input_mappings[%d]: %w", name, i, step.ID, j, err) + } + if _, ok := seen[mapping.FromStep]; !ok { + return fmt.Errorf("step.compute_chain %q: steps[%d] %q input_mappings[%d].from_step %q must reference an earlier step", name, i, step.ID, j, mapping.FromStep) + } + if _, ok := deps[mapping.FromStep]; !ok { + return fmt.Errorf("step.compute_chain %q: steps[%d] %q input_mappings[%d].from_step %q must be listed in depends_on", name, i, step.ID, j, mapping.FromStep) + } + } + seen[step.ID] = struct{}{} + } + return nil +} + +func (c computeChainConfig) requireProof() bool { + return c.RequireProof == nil || *c.RequireProof +} + +func (c computeChainTaskConfig) taskConfig() taskConfig { + return taskConfig{ + ID: c.TaskID, + ProductID: c.ProductID, + OrgID: c.OrgID, + PoolID: c.PoolID, + PolicyID: c.PolicyID, + TimeoutSeconds: c.TimeoutSeconds, + Labels: c.Labels, + ResiduePolicy: c.ResiduePolicy, + } +} + +func (c computeChainTaskConfig) wait() bool { + return c.Wait == nil || *c.Wait +} + +func (m computeChainInputMapping) validate() error { + var errs []error + if m.FromStep == "" { + errs = append(errs, errors.New("from_step is required")) + } + if m.From == "" { + errs = append(errs, errors.New("from is required")) + } + switch m.To { + case "provider.artifact_refs", "provider.content_inputs", "provider.stream_inputs": + default: + errs = append(errs, fmt.Errorf("unsupported to value %q", m.To)) + } + return errors.Join(errs...) +} + +func (s *computeChainStep) Execute(ctx context.Context, _ map[string]any, _ map[string]map[string]any, _ map[string]any, metadata map[string]any, runtimeConfig map[string]any) (*sdk.StepResult, error) { + client, err := s.config.connectionConfig.client(ctx, metadata, runtimeConfig) + if err != nil { + return errorResult(err.Error()), nil + } + + timeout := durationOrDefault(s.config.Timeout, 30*time.Minute) + chainCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + outputs := make([]map[string]any, 0, len(s.config.Steps)) + outputsByID := make(map[string]map[string]any, len(s.config.Steps)) + for _, chainTask := range s.config.Steps { + workload := chainTask.Workload + if err := applyComputeChainInputMappings(&workload, chainTask.InputMappings, outputsByID); err != nil { + return chainErrorResult(outputs, chainTask.ID, err.Error()), nil + } + if err := workload.Validate(); err != nil { + return chainErrorResult(outputs, chainTask.ID, fmt.Sprintf("workload: %v", err)), nil + } + task, err := client.submitTask(chainCtx, buildTask(chainTask.taskConfig(), workload)) + if err != nil { + return chainErrorResult(outputs, chainTask.ID, err.Error()), nil + } + + output := taskOutput(task) + output["step_id"] = chainTask.ID + if chainTask.wait() { + output, err = s.waitForTask(chainCtx, client, task.ID) + if err != nil { + return chainErrorResult(append(outputs, output), chainTask.ID, err.Error()), nil + } + output["step_id"] = chainTask.ID + } + outputs = append(outputs, output) + outputsByID[chainTask.ID] = output + } + + return &sdk.StepResult{Output: map[string]any{"steps": outputs}}, nil +} + +func (s *computeChainStep) waitForTask(ctx context.Context, client *computeClient, taskID string) (map[string]any, error) { + pollInterval := durationOrDefault(s.config.PollInterval, time.Second) + for { + task, found, stalls, err := client.taskSnapshot(ctx, taskID) + if err != nil { + return taskOutput(protocol.Task{ID: taskID}), err + } + if !found { + return taskOutput(protocol.Task{ID: taskID}), fmt.Errorf("task %q not found", taskID) + } + output := taskOutput(task) + actionableStalls := actionableStalls(stalls, s.config.requireProof()) + if task.Status == protocol.TaskFailed || task.Status == protocol.TaskStalled || len(actionableStalls) > 0 { + if len(actionableStalls) > 0 { + addStallOutput(output, actionableStalls[0]) + } + msg := taskWaitError(task, actionableStalls) + output["error"] = msg + return output, errors.New(msg) + } + if isTerminalTaskStatus(task.Status) { + proof, hasProof, err := client.findProof(ctx, task.ID) + if err != nil { + return output, err + } + if hasProof { + addProofOutput(output, proof) + } + if hasProof && proof.Verifier.Status != protocol.VerificationAccepted { + msg := fmt.Sprintf("task %q proof %q is %s", task.ID, proof.ID, proof.Verifier.Status) + output["error"] = msg + return output, errors.New(msg) + } + if hasProof || !s.config.requireProof() { + return output, nil + } + } + timer := time.NewTimer(pollInterval) + select { + case <-ctx.Done(): + timer.Stop() + return output, fmt.Errorf("timed out waiting for task %q", taskID) + case <-timer.C: + } + } +} + +func applyComputeChainInputMappings(workload *protocol.WorkloadSpec, mappings []computeChainInputMapping, outputs map[string]map[string]any) error { + if len(mappings) == 0 { + return nil + } + if workload.Kind != protocol.WorkloadProvider || workload.Provider == nil { + return errors.New("input_mappings require provider workload") + } + for _, mapping := range mappings { + source, ok := outputs[mapping.FromStep] + if !ok { + return fmt.Errorf("source step %q has no output", mapping.FromStep) + } + value, ok := lookupChainValue(source, mapping.From) + if !ok { + return fmt.Errorf("source step %q output %q is missing", mapping.FromStep, mapping.From) + } + switch mapping.To { + case "provider.artifact_refs": + refs, err := stringSliceFromAny(value) + if err != nil { + return fmt.Errorf("%s: %w", mapping.To, err) + } + workload.Provider.ArtifactRefs = append(workload.Provider.ArtifactRefs, refs...) + case "provider.content_inputs": + inputs, err := typedSliceFromAny[coreprotocol.ContentInputRef](value) + if err != nil { + return fmt.Errorf("%s: %w", mapping.To, err) + } + workload.Provider.ContentInputs = append(workload.Provider.ContentInputs, inputs...) + case "provider.stream_inputs": + inputs, err := typedSliceFromAny[coreprotocol.StreamInputRef](value) + if err != nil { + return fmt.Errorf("%s: %w", mapping.To, err) + } + workload.Provider.StreamInputs = append(workload.Provider.StreamInputs, inputs...) + default: + return fmt.Errorf("unsupported mapping destination %q", mapping.To) + } + } + return nil +} + +func lookupChainValue(values map[string]any, path string) (any, bool) { + current := any(values) + for _, part := range splitDottedPath(path) { + object, ok := current.(map[string]any) + if !ok { + return nil, false + } + current, ok = object[part] + if !ok { + return nil, false + } + } + return current, true +} + +func splitDottedPath(path string) []string { + parts := make([]string, 0, 1) + start := 0 + for i, r := range path { + if r == '.' { + parts = append(parts, path[start:i]) + start = i + 1 + } + } + return append(parts, path[start:]) +} + +func stringSliceFromAny(value any) ([]string, error) { + switch v := value.(type) { + case string: + return []string{v}, nil + case []string: + return append([]string(nil), v...), nil + case []any: + out := make([]string, 0, len(v)) + for i, item := range v { + ref, ok := item.(string) + if !ok { + return nil, fmt.Errorf("[%d] must be string", i) + } + out = append(out, ref) + } + return out, nil + default: + return nil, fmt.Errorf("must be string or string array") + } +} + +func typedSliceFromAny[T any](value any) ([]T, error) { + data, err := json.Marshal(value) + if err != nil { + return nil, err + } + var out []T + if err := json.Unmarshal(data, &out); err != nil { + return nil, err + } + return out, nil +} + +func stringSet(values []string) map[string]struct{} { + set := make(map[string]struct{}, len(values)) + for _, value := range values { + set[value] = struct{}{} + } + return set +} + +func chainErrorResult(steps []map[string]any, stepID, msg string) *sdk.StepResult { + output := map[string]any{ + "error": fmt.Sprintf("step %q: %s", stepID, msg), + "step_id": stepID, + "steps": steps, + } + return &sdk.StepResult{StopPipeline: true, Output: output} +} diff --git a/internal/steps_chain_test.go b/internal/steps_chain_test.go new file mode 100644 index 0000000..265cdbe --- /dev/null +++ b/internal/steps_chain_test.go @@ -0,0 +1,243 @@ +package internal + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/GoCodeAlone/workflow-compute/pkg/protocol" +) + +func TestComputeChainSubmitsSequentialProviderTasksAndMapsInputs(t *testing.T) { + var submitted []protocol.Task + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1/tasks": + var task protocol.Task + if err := json.NewDecoder(r.Body).Decode(&task); err != nil { + t.Fatalf("decode task: %v", err) + } + submitted = append(submitted, task) + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{"task": task}) + case r.Method == http.MethodGet && r.URL.Path == "/v1/tasks": + tasks := make([]protocol.Task, len(submitted)) + copy(tasks, submitted) + for i := range tasks { + tasks[i].Status = protocol.TaskSucceeded + } + _ = json.NewEncoder(w).Encode(map[string]any{"tasks": tasks}) + case r.Method == http.MethodGet && r.URL.Path == "/v1/proofs": + proofs := make([]protocol.ProofReceipt, 0, len(submitted)) + for _, task := range submitted { + proof := proofReceipt(task.ID) + proof.ID = task.ID + "-proof" + if task.ID == "fetch-task" { + proof.ResultPreview = map[string]any{ + "artifact_refs": []any{"artifact://pool-1/tasks/fetch-task/proofs/fetch-task-proof/source.mp4"}, + "content_inputs": []any{map[string]any{ + "name": "source-file", + "ref": "content://workloads/source.mp4", + "target_path": "inputs/source.mp4", + "content_type": "video/mp4", + }}, + "stream_inputs": []any{map[string]any{ + "name": "live-source", + "handle": map[string]any{ + "url": "rtmp://stream.example.test/live/source", + "protocol": "rtmp", + "auth_token_ref": "secret://streams/live-source", + "codecs": []any{"h264", "aac"}, + "expires_at": "2026-06-30T22:00:00Z", + }, + }}, + } + } + proofs = append(proofs, proof) + } + _ = json.NewEncoder(w).Encode(map[string]any{"proofs": proofs}) + default: + t.Fatalf("request: %s %s", r.Method, r.URL.Path) + } + })) + defer srv.Close() + + step, err := newComputeChainStep("chain", chainConfigMap(srv.URL)) + if err != nil { + t.Fatalf("newComputeChainStep: %v", err) + } + result, err := step.Execute(context.Background(), nil, nil, nil, nil, runtimeSecrets()) + if err != nil { + t.Fatalf("Execute: %v", err) + } + if result.StopPipeline { + t.Fatalf("unexpected stop: %+v", result.Output) + } + if len(submitted) != 2 { + t.Fatalf("submitted tasks: got %d", len(submitted)) + } + transform := submitted[1].Workload.Provider + if transform == nil { + t.Fatalf("second task provider workload: %+v", submitted[1].Workload) + } + if len(transform.ArtifactRefs) != 1 || transform.ArtifactRefs[0] != "artifact://pool-1/tasks/fetch-task/proofs/fetch-task-proof/source.mp4" { + t.Fatalf("artifact refs: %+v", transform.ArtifactRefs) + } + if len(transform.ContentInputs) != 1 || transform.ContentInputs[0].Ref != "content://workloads/source.mp4" { + t.Fatalf("content inputs: %+v", transform.ContentInputs) + } + if len(transform.StreamInputs) != 1 || transform.StreamInputs[0].Handle.URL != "rtmp://stream.example.test/live/source" { + t.Fatalf("stream inputs: %+v", transform.StreamInputs) + } + + steps := result.Output["steps"].([]map[string]any) + if len(steps) != 2 || steps[0]["step_id"] != "fetch" || steps[1]["step_id"] != "transform" { + t.Fatalf("chain output: %+v", result.Output) + } +} + +func TestComputeChainRejectsUnknownConfig(t *testing.T) { + cfg := chainConfigMap("https://compute.example.test") + cfg["unknown"] = true + if _, err := newComputeChainStep("chain", cfg); err == nil { + t.Fatal("expected strict unknown-field error") + } +} + +func TestComputeChainRequiresMappingSourcesInDependsOn(t *testing.T) { + cfg := chainConfigMap("https://compute.example.test") + steps := cfg["steps"].([]any) + transform := steps[1].(map[string]any) + delete(transform, "depends_on") + _, err := newComputeChainStep("chain", cfg) + if err == nil || !strings.Contains(err.Error(), "must be listed in depends_on") { + t.Fatalf("expected depends_on validation error, got %v", err) + } +} + +func TestComputeChainWaitFalseSkipsProofPolling(t *testing.T) { + var proofRequests int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1/tasks": + var task protocol.Task + if err := json.NewDecoder(r.Body).Decode(&task); err != nil { + t.Fatalf("decode task: %v", err) + } + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{"task": task}) + case r.Method == http.MethodGet && r.URL.Path == "/v1/proofs": + proofRequests++ + _ = json.NewEncoder(w).Encode(map[string]any{"proofs": []any{}}) + default: + t.Fatalf("request: %s %s", r.Method, r.URL.Path) + } + })) + defer srv.Close() + + cfg := chainConfigMap(srv.URL) + cfg["steps"] = []any{map[string]any{ + "id": "enqueue", + "task_id": "enqueue-task", + "org_id": "org-1", + "pool_id": "pool-1", + "policy_id": "policy-1", + "timeout_seconds": 60, + "wait": false, + "workload": commandWorkloadMap(), + }} + step, err := newComputeChainStep("chain", cfg) + if err != nil { + t.Fatalf("newComputeChainStep: %v", err) + } + result, err := step.Execute(context.Background(), nil, nil, nil, nil, runtimeSecrets()) + if err != nil { + t.Fatalf("Execute: %v", err) + } + if result.StopPipeline { + t.Fatalf("unexpected stop: %+v", result.Output) + } + if proofRequests != 0 { + t.Fatalf("wait=false should not poll proofs, got %d requests", proofRequests) + } +} + +func chainConfigMap(serverURL string) map[string]any { + return map[string]any{ + "server_url": serverURL, + "auth_token_ref": "secret:compute-token", + "poll_interval": "1ms", + "timeout": "5s", + "require_proof": true, + "steps": []any{ + map[string]any{ + "id": "fetch", + "task_id": "fetch-task", + "org_id": "org-1", + "pool_id": "pool-1", + "policy_id": "policy-1", + "timeout_seconds": 60, + "workload": providerWorkloadMap("fetch"), + }, + map[string]any{ + "id": "transform", + "task_id": "transform-task", + "org_id": "org-1", + "pool_id": "pool-1", + "policy_id": "policy-1", + "timeout_seconds": 60, + "depends_on": []any{"fetch"}, + "input_mappings": []any{ + map[string]any{ + "from_step": "fetch", + "from": "result_preview.artifact_refs", + "to": "provider.artifact_refs", + }, + map[string]any{ + "from_step": "fetch", + "from": "result_preview.content_inputs", + "to": "provider.content_inputs", + }, + map[string]any{ + "from_step": "fetch", + "from": "result_preview.stream_inputs", + "to": "provider.stream_inputs", + }, + }, + "workload": providerWorkloadMap("transform"), + }, + }, + } +} + +func providerWorkloadMap(operation string) map[string]any { + return map[string]any{ + "kind": "provider", + "provider": map[string]any{ + "provider_config": map[string]any{ + "plugin_id": "workflow-plugin-generic-provider", + "provider_id": "transformer", + "contract_id": "generic-transform.v1", + "version": "v1.0.0", + "config_ref": "config://providers/generic-transformer/main", + }, + "operation": operation, + "image_ref": testProviderImageRef, + "input": map[string]any{ + "value": operation, + }, + }, + } +} + +func commandWorkloadMap() map[string]any { + return map[string]any{ + "kind": "command", + "command": map[string]any{ + "args": []any{"true"}, + }, + } +} diff --git a/internal/steps_stream_test.go b/internal/steps_stream_test.go index 80eb5e5..9eaac0e 100644 --- a/internal/steps_stream_test.go +++ b/internal/steps_stream_test.go @@ -97,7 +97,7 @@ func streamConfigMap(serverURL string) map[string]any { func TestStepTypesIncludeComputeStream(t *testing.T) { got := NewPlugin().(interface{ StepTypes() []string }).StepTypes() - want := []string{"step.compute_dispatch", "step.compute_wait", "step.compute_map", "step.compute_stream"} + want := []string{"step.compute_dispatch", "step.compute_wait", "step.compute_map", "step.compute_stream", "step.compute_chain"} if len(got) != len(want) { t.Fatalf("step types: got %#v", got) } diff --git a/internal/steps_test.go b/internal/steps_test.go index 793b7b0..1a234ba 100644 --- a/internal/steps_test.go +++ b/internal/steps_test.go @@ -18,7 +18,7 @@ const testProviderImageRef = "ghcr.io/gocodealone/generic-provider@sha256:aaaaaa func TestStepTypes(t *testing.T) { steps := NewPlugin().(interface{ StepTypes() []string }) got := steps.StepTypes() - want := []string{"step.compute_dispatch", "step.compute_wait", "step.compute_map", "step.compute_stream"} + want := []string{"step.compute_dispatch", "step.compute_wait", "step.compute_map", "step.compute_stream", "step.compute_chain"} if !reflect.DeepEqual(got, want) { t.Fatalf("step types: got %#v", got) } diff --git a/plugin.json b/plugin.json index 0b559dd..08a75a0 100644 --- a/plugin.json +++ b/plugin.json @@ -1,7 +1,7 @@ { "name": "workflow-plugin-compute", "version": "0.1.10", - "description": "Workflow adapter for workflow-compute dispatch, wait, map, provider, pool, setup, and provider catalog integration", + "description": "Workflow adapter for workflow-compute dispatch, wait, map, chain, provider, pool, setup, and provider catalog integration", "author": "GoCodeAlone", "license": "MIT", "type": "external", @@ -13,7 +13,7 @@ "capabilities": { "configProvider": false, "moduleTypes": ["compute.provider", "compute.pool", "compute.provider_catalog"], - "stepTypes": ["step.compute_dispatch", "step.compute_wait", "step.compute_map", "step.compute_stream"], + "stepTypes": ["step.compute_dispatch", "step.compute_wait", "step.compute_map", "step.compute_stream", "step.compute_chain"], "triggerTypes": [], "cliCommands": [ { @@ -25,7 +25,7 @@ "dependencies": [ { "name": "workflow-plugin-compute-core", - "constraint": ">=0.8.1" + "constraint": ">=0.8.3" }, { "name": "workflow-plugin-compute-container", From 30eb7e21f565fc97e52eff474412b150381a7bc8 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Tue, 30 Jun 2026 18:09:09 -0400 Subject: [PATCH 2/2] fix: preserve chain failure step id --- internal/steps_chain.go | 2 +- internal/steps_chain_test.go | 52 ++++++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/internal/steps_chain.go b/internal/steps_chain.go index e8c0339..08d82a0 100644 --- a/internal/steps_chain.go +++ b/internal/steps_chain.go @@ -182,10 +182,10 @@ func (s *computeChainStep) Execute(ctx context.Context, _ map[string]any, _ map[ output["step_id"] = chainTask.ID if chainTask.wait() { output, err = s.waitForTask(chainCtx, client, task.ID) + output["step_id"] = chainTask.ID if err != nil { return chainErrorResult(append(outputs, output), chainTask.ID, err.Error()), nil } - output["step_id"] = chainTask.ID } outputs = append(outputs, output) outputsByID[chainTask.ID] = output diff --git a/internal/steps_chain_test.go b/internal/steps_chain_test.go index 265cdbe..d7a9aea 100644 --- a/internal/steps_chain_test.go +++ b/internal/steps_chain_test.go @@ -165,6 +165,58 @@ func TestComputeChainWaitFalseSkipsProofPolling(t *testing.T) { } } +func TestComputeChainFailureOutputKeepsStepID(t *testing.T) { + var submitted []protocol.Task + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1/tasks": + var task protocol.Task + if err := json.NewDecoder(r.Body).Decode(&task); err != nil { + t.Fatalf("decode task: %v", err) + } + submitted = append(submitted, task) + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{"task": task}) + case r.Method == http.MethodGet && r.URL.Path == "/v1/tasks": + tasks := make([]protocol.Task, len(submitted)) + copy(tasks, submitted) + for i := range tasks { + tasks[i].Status = protocol.TaskFailed + } + _ = json.NewEncoder(w).Encode(map[string]any{"tasks": tasks}) + default: + t.Fatalf("request: %s %s", r.Method, r.URL.Path) + } + })) + defer srv.Close() + + cfg := chainConfigMap(srv.URL) + cfg["steps"] = []any{map[string]any{ + "id": "failing", + "task_id": "failing-task", + "org_id": "org-1", + "pool_id": "pool-1", + "policy_id": "policy-1", + "timeout_seconds": 60, + "workload": commandWorkloadMap(), + }} + step, err := newComputeChainStep("chain", cfg) + if err != nil { + t.Fatalf("newComputeChainStep: %v", err) + } + result, err := step.Execute(context.Background(), nil, nil, nil, nil, runtimeSecrets()) + if err != nil { + t.Fatalf("Execute: %v", err) + } + if !result.StopPipeline { + t.Fatalf("expected stop pipeline, got %+v", result.Output) + } + steps := result.Output["steps"].([]map[string]any) + if len(steps) != 1 || steps[0]["step_id"] != "failing" { + t.Fatalf("failure output must retain step_id, got %+v", result.Output) + } +} + func chainConfigMap(serverURL string) map[string]any { return map[string]any{ "server_url": serverURL,