diff --git a/README.md b/README.md index 4db8a3e4..b91599ec 100644 --- a/README.md +++ b/README.md @@ -120,7 +120,6 @@ func Build(ctx context.Context, cfg *Config) (*App, error) { if err := artifacts.Validate(ctx, artifacts.Current(), corechecks.Checks()); err != nil { return nil, err } - registry := artifacts.CurrentRegistry() // 3. Payment paymentRepo := NewPaymentRepo(db) @@ -152,7 +151,8 @@ func Build(ctx context.Context, cfg *Config) (*App, error) { onTaskCompleted := func(parentWorkflowID, parentStepID string, vars map[string]any) error { return consignmentService.HandleTaskCompletion(ctx, parentWorkflowID, parentStepID, vars) } - tm = orchestrator.NewTaskManager(taskStore, registry, pluginRegistry, microRunner, onTaskCompleted, taskRenderer) + // Each task resolves its templates through the manifest version its workflow carries. + tm = orchestrator.NewTaskManager(taskStore, artifacts, pluginRegistry, nil /* extensions */, microRunner, onTaskCompleted, taskRenderer) microRunner.StartWorker() // 6. Macro workflow runner @@ -161,6 +161,10 @@ func Build(ctx context.Context, cfg *Config) (*App, error) { onMacroTaskActivation, onMacroCompletion, ) macroRunner.StartWorker() + // Start each top-level workflow on the current manifest; the engine carries the version to + // every child workflow and task: + // vars[workflow.VarManifestVersion] = artifacts.Current() + // macroRunner.StartWorkflow(ctx, id, def, vars) // 7. Auth authnManager, _ := authn.NewManager(userProfileSvc, cfg.Authn) diff --git a/taskflow/README.md b/taskflow/README.md index ec8956d7..9e03e2e5 100644 --- a/taskflow/README.md +++ b/taskflow/README.md @@ -84,7 +84,9 @@ workflowRunner := workflow.NewTemporalManager( onTaskCompleted := func(parentWorkflowID, parentRunID, parentNodeID string, vars map[string]any) error { return parentWorkflowManager.CompleteActivation(ctx, parentWorkflowID, parentRunID, parentNodeID, vars) } -tm = orchestrator.NewTaskManager(store, artifactRegistry, pluginRegistry, workflowRunner, onTaskCompleted, taskRenderer) +// artifacts is an *artifact.Set: each task resolves its templates through the manifest version +// its parent workflow carries (workflow.VarManifestVersion), and records it on the task. +tm = orchestrator.NewTaskManager(store, artifacts, pluginRegistry, nil /* extensions */, workflowRunner, onTaskCompleted, taskRenderer) // 5. Start the Temporal worker if err := workflowRunner.StartWorker(); err != nil { diff --git a/taskflow/docs/integration-guide.md b/taskflow/docs/integration-guide.md index a2acb77d..b65d0bbd 100644 --- a/taskflow/docs/integration-guide.md +++ b/taskflow/docs/integration-guide.md @@ -13,7 +13,7 @@ To run the orchestrator, you provide concrete implementations of four dependenci | Dependency | Interface | What it does | |-------------------|--------------------------------------|----------------------------------------------------------------| | Task store | `store.TaskStore` | Persists `TaskRecord`s | -| Template registry | `orchestrator.TaskTemplateRegistry` | Looks up task / step / workflow / render-config definitions | +| Artifacts | `*artifact.Set` | Task / step / workflow / render-config definitions, per manifest version | | Plugin registry | `*plugins.Registry` (concrete) | Maps `TaskType` → plugin handler | | Temporal manager | `engine.TemporalManager` | Starts task workflows and resumes parked activities | | Renderer | `renderer.Renderer` | Turns `(state, data)` + render config → a UI view | @@ -22,10 +22,10 @@ To run the orchestrator, you provide concrete implementations of four dependenci Then you call: ```go -tm := orchestrator.NewTaskManager(db, registry, pluginsReg, taskWorkflowManager, onTaskCompleted, rdr) +tm := orchestrator.NewTaskManager(db, artifacts, pluginsReg, nil /* extensions */, taskWorkflowManager, onTaskCompleted, rdr) ``` -…and wire `tm.StartTask`, `tm.StartTaskStep`, and `tm.HandleTaskCompletion` into your Temporal handlers. `demo/main.go` is the complete worked example. +…and wire `tm.StartTask`, `tm.StartTaskStep`, and `tm.HandleTaskCompletion` into your Temporal handlers. Start every parent workflow with `workflow.VarManifestVersion` set to `artifacts.Current()`: `StartTask` records that version on the task and refuses a payload without one, and every later step, submission and render of the task resolves through it. `demo/main.go` is the complete worked example. --- @@ -182,7 +182,7 @@ taskHandler := func(p engine.TaskPayload) (map[string]any, error) { // ... construct managers ... -tm = orchestrator.NewTaskManager(db, registry, pluginsReg, taskWorkflowManager, onTaskCompleted, rdr) +tm = orchestrator.NewTaskManager(db, artifacts, pluginsReg, nil /* extensions */, taskWorkflowManager, onTaskCompleted, rdr) ``` --- diff --git a/taskflow/orchestrator/complete_step_test.go b/taskflow/orchestrator/complete_step_test.go index a8cdcfcd..58c799c5 100644 --- a/taskflow/orchestrator/complete_step_test.go +++ b/taskflow/orchestrator/complete_step_test.go @@ -23,7 +23,7 @@ const wakeTaskID = "5f0c9b1e-3d2a-4c6b-8e7f-0a1b2c3d4e5f" func wakeUpFixture(t *testing.T, db store.TaskStore) (*TaskManager, *mockTemporalManager) { t.Helper() ctx := context.Background() - db.InitTask(ctx, store.TaskRecord{TaskID: wakeTaskID, TaskType: "TEST", State: "STARTING", TaskWorkflowID: "task-wf", Data: map[string]any{}}) + db.InitTask(ctx, store.TaskRecord{ManifestVersion: 1, TaskID: wakeTaskID, TaskType: "TEST", State: "STARTING", TaskWorkflowID: "task-wf", Data: map[string]any{}}) if _, err := db.ClaimStep(ctx, wakeTaskID, store.StepClaim{ StepID: stepA, Seq: 4, ActiveTaskTemplateID: "generic_user_input", State: store.StateStartingStep, Data: map[string]any{"in": "value"}, @@ -220,7 +220,8 @@ func TestCompleteTaskStep_CallbackBeforeTheRenderWriteWins(t *testing.T) { } if _, err := tm.StartTaskStep(context.Background(), engine.TaskPayload{ - WorkflowID: "task-wf", ActivationID: stepA, Seq: 4, TaskTemplateID: "generic_user_input", + ManifestVersion: 1, + WorkflowID: "task-wf", ActivationID: stepA, Seq: 4, TaskTemplateID: "generic_user_input", }); err == nil { t.Fatal("expected ErrResultPending from the suspended plugin") } @@ -234,7 +235,7 @@ func TestCompleteTaskStep_CallbackBeforeTheRenderWriteWins(t *testing.T) { // startTaskStepManager is a task manager over db whose USER_INPUT plugin is a probe. func startTaskStepManager(t *testing.T, db *safeMockTaskStore) (*TaskManager, *probePlugin) { t.Helper() - db.InitTask(context.Background(), store.TaskRecord{TaskID: wakeTaskID, TaskType: "TEST", State: "STARTING", TaskWorkflowID: "task-wf", Data: map[string]any{}}) + db.InitTask(context.Background(), store.TaskRecord{ManifestVersion: 1, TaskID: wakeTaskID, TaskType: "TEST", State: "STARTING", TaskWorkflowID: "task-wf", Data: map[string]any{}}) tm, _, probe := startTaskStepFixtureOver(t, db) return tm, probe } diff --git a/taskflow/orchestrator/completion_test.go b/taskflow/orchestrator/completion_test.go index 5d39648d..11c4116a 100644 --- a/taskflow/orchestrator/completion_test.go +++ b/taskflow/orchestrator/completion_test.go @@ -18,7 +18,8 @@ func completionFixture(t *testing.T, onCompleted TaskCompletedCallback) (*TaskMa t.Helper() db := newSafeMockTaskStore() db.InitTask(context.Background(), store.TaskRecord{ - TaskID: wakeTaskID, TaskWorkflowID: "task-wf", State: "STARTING", Data: map[string]any{}, + ManifestVersion: 1, + TaskID: wakeTaskID, TaskWorkflowID: "task-wf", State: "STARTING", Data: map[string]any{}, ParentWorkflowID: "parent-wf", ParentStepID: "parent-step", }) if _, err := db.ClaimStep(context.Background(), wakeTaskID, store.StepClaim{StepID: stepB, Seq: 6, State: store.StateStartingStep}); err != nil { diff --git a/taskflow/orchestrator/manager.go b/taskflow/orchestrator/manager.go index 7c5dd1a6..9278b27d 100644 --- a/taskflow/orchestrator/manager.go +++ b/taskflow/orchestrator/manager.go @@ -92,7 +92,7 @@ var ErrStepIDRequired = errors.New("step ID is required") type TaskManager struct { db store.TaskStore renderer renderer.Renderer - registry *artifact.Registry + artifacts *artifact.Set pluginsRegistry *plugins.Registry extensionsRegistry *extensions.Registry onTaskCompleted TaskCompletedCallback @@ -103,14 +103,15 @@ type TaskManager struct { // NewTaskManager creates a TaskManager instance. // // - db — the persistence/in-memory task store. -// - registry — artifact registry holding task templates, step templates, workflow definitions, and render configs. +// - artifacts — the artifact manifests holding task templates, step templates, workflow definitions, and render configs. +// Every task resolves them through the manifest version its parent workflow carries (see engine.VarManifestVersion). // - pluginsRegistry — registry containing task execution plugin handlers. // - taskWorkflowManager — the TemporalManager used to start and complete Task sub-workflows. // - onTaskCompleted — callback invoked when a Task workflow finishes; // typically invokes Parent.CompleteActivation to resume the parent workflow using stored coordinates. func NewTaskManager( db store.TaskStore, - registry *artifact.Registry, + artifacts *artifact.Set, pluginsRegistry *plugins.Registry, extensionsRegistry *extensions.Registry, taskWorkflowManager engine.TemporalManager, @@ -120,7 +121,7 @@ func NewTaskManager( ) *TaskManager { tm := &TaskManager{ db: db, - registry: registry, + artifacts: artifacts, pluginsRegistry: pluginsRegistry, extensionsRegistry: extensionsRegistry, onTaskCompleted: onTaskCompleted, @@ -137,21 +138,46 @@ func NewTaskManager( return tm } +// registryFor returns the registry of manifest version v. A task without a version cannot resolve +// templates: falling back to the current version would change templates under a running task. That +// is a host wiring fault no retry repairs, so it fails without retrying; a version this replica has +// not loaded yet may still appear, so that error stays retryable. +// +// Temporal reads retryability off the returned error itself, not anything it wraps, so callers +// running as an Activity return the error unwrapped. +func (tm *TaskManager) registryFor(ctx context.Context, taskTemplateID string, v int) (*artifact.Registry, error) { + if v <= 0 { + return nil, temporal.NewNonRetryableApplicationError( + fmt.Sprintf("template %q: no artifact manifest version (got %d); the workflow was started without %s", taskTemplateID, v, engine.VarManifestVersion), + "NoManifestVersion", artifact.ErrUnknownManifestVersion) + } + reg, err := tm.artifacts.Registry(ctx, v) + if err != nil { + return nil, fmt.Errorf("template %q: %w", taskTemplateID, err) + } + return reg, nil +} + // StartTask is called by the parent workflow engine when it activates a TASK node. // It looks up the template registry, creates a single DB record with parent // coordinates, and kicks off the Task's internal workflow. func (tm *TaskManager) StartTask(ctx context.Context, payload engine.TaskPayload) (map[string]any, error) { - template, err := tasktemplate.Load(ctx, tm.registry, payload.TaskTemplateID) + reg, err := tm.registryFor(ctx, payload.TaskTemplateID, payload.ManifestVersion) + if err != nil { + return nil, err + } + + template, err := tasktemplate.Load(ctx, reg, payload.TaskTemplateID) if err != nil { return nil, fmt.Errorf("load task template %q: %w", payload.TaskTemplateID, err) } - wfDef, err := workflowdef.Load(ctx, tm.registry, template.WorkflowID) + wfDef, err := workflowdef.Load(ctx, reg, template.WorkflowID) if err != nil { return nil, fmt.Errorf("load workflow %q referenced by task template %q: %w", template.WorkflowID, template.ID, err) } - renderConfig, err := generictemplate.Load(ctx, tm.registry, template.RenderConfigID) + renderConfig, err := generictemplate.Load(ctx, reg, template.RenderConfigID) if err != nil { return nil, fmt.Errorf("load render config %q referenced by task template %q: %w", template.RenderConfigID, template.ID, err) } @@ -170,6 +196,8 @@ func (tm *TaskManager) StartTask(ctx context.Context, payload engine.TaskPayload maputil.SetNestedKey(initialData, k, v) } initialData["_task_id"] = taskID + // The task's own workflow resolves its steps through the same manifest version. + initialData[engine.VarManifestVersion] = payload.ManifestVersion record := store.TaskRecord{ TaskID: taskID, @@ -179,6 +207,7 @@ func (tm *TaskManager) StartTask(ctx context.Context, payload engine.TaskPayload ParentWorkflowID: payload.WorkflowID, ParentStepID: payload.ActivationID, // the parent's Activity to complete CaseID: payload.CaseID, // the same at every level of nesting (see engine.VarCaseID) + ManifestVersion: payload.ManifestVersion, TaskWorkflowID: taskWorkflowID, Data: initialData, CreatedAt: time.Now(), @@ -228,7 +257,14 @@ func (tm *TaskManager) StartTaskStep(ctx context.Context, payload engine.TaskPay return nil, fmt.Errorf("[StartTaskStep] no task record found for workflow %s", payload.WorkflowID) } - stepTemplate, err := steptemplate.Load(ctx, tm.registry, payload.TaskTemplateID) + // The record's version is the one the task started on; every step uses it. + reg, err := tm.registryFor(ctx, payload.TaskTemplateID, record.ManifestVersion) + if err != nil { + return nil, err + } + ctx = artifact.WithManifestVersion(ctx, record.ManifestVersion) + + stepTemplate, err := steptemplate.Load(ctx, reg, payload.TaskTemplateID) if err != nil { return nil, fmt.Errorf("[StartTaskStep] load step template %q: %w", payload.TaskTemplateID, err) } @@ -401,7 +437,14 @@ func (tm *TaskManager) CompleteTaskStep(ctx context.Context, taskID, stepID stri } seq := record.Seq - stepTemplate, err := steptemplate.Load(ctx, tm.registry, record.ActiveTaskTemplateID) + reg, err := tm.registryFor(ctx, record.ActiveTaskTemplateID, record.ManifestVersion) + if err != nil { + return fmt.Errorf("task %s: %w", taskID, err) + } + // Extensions (a notification's template, for example) resolve through the task's version. + ctx = artifact.WithManifestVersion(ctx, record.ManifestVersion) + + stepTemplate, err := steptemplate.Load(ctx, reg, record.ActiveTaskTemplateID) if err != nil { return fmt.Errorf("failed to load active step template %q: %w", record.ActiveTaskTemplateID, err) } @@ -533,7 +576,10 @@ func (tm *TaskManager) GetTaskRenderInfo(context context.Context, taskID string) return TaskView{}, fmt.Errorf("task record %s not found", taskID) } - view, err := tm.renderer.Render(context, record.RenderConfig, renderer.Facts{State: record.State, Data: record.Data}) + // The renderer's template provider resolves the render config's templates through the task's + // version, so a form does not change under an open task. + renderCtx := artifact.WithManifestVersion(context, record.ManifestVersion) + view, err := tm.renderer.Render(renderCtx, record.RenderConfig, renderer.Facts{State: record.State, Data: record.Data}) if err != nil { return TaskView{}, fmt.Errorf("rendering task %s: %w", taskID, err) } diff --git a/taskflow/orchestrator/manager_test.go b/taskflow/orchestrator/manager_test.go index 859a953b..73845fc5 100644 --- a/taskflow/orchestrator/manager_test.go +++ b/taskflow/orchestrator/manager_test.go @@ -80,7 +80,7 @@ func (m *mockTemporalManager) GetStatus(ctx context.Context, workflowID string) return nil, nil } -func (m *mockTemporalManager) RegisterDefinitionHandler(_ func(templateID string) (engine.WorkflowDefinition, error)) { +func (m *mockTemporalManager) RegisterDefinitionHandler(_ engine.DefinitionHandler) { } func (m *mockTemporalManager) RegisterAdminParkHandler(_ engine.AdminParkHandler) { @@ -212,13 +212,14 @@ func newTestPluginsRegistry() *plugins.Registry { return pr } -func newTestTaskManager(db store.TaskStore, registry *artifact.Registry, tm engine.TemporalManager, cb TaskCompletedCallback) *TaskManager { +func newTestTaskManager(db store.TaskStore, registry *artifact.Set, tm engine.TemporalManager, cb TaskCompletedCallback) *TaskManager { return NewTaskManager(db, registry, newTestPluginsRegistry(), nil, tm, cb, noopRenderer{}) } func noopCallback(_, _ string, _ map[string]any) error { return nil } -func newTestRegistry() *artifact.Registry { +// newTestRegistry returns a one-manifest artifact set; tasks in these tests run on version 1. +func newTestRegistry() *artifact.Set { m := testutil.MemLoader{ "task_test_template.json": []byte(`{ "id": "test_template", @@ -258,14 +259,19 @@ func newTestRegistry() *artifact.Registry { } }`), } - reg := artifact.NewRegistry(m) - reg.RegisterArtifact("test_template", "task_template", "", "task_test_template.json") - reg.RegisterArtifact("test_workflow_v1", "workflow", "", "wf_test_workflow_v1.json") - reg.RegisterArtifact("test_render_config", "generic_template", "", "generic_render_config.json") - reg.RegisterArtifact("generic_user_input", "subtask_template", "", "subtask_generic_user_input.json") - reg.RegisterArtifact("generic_user_input_with_extensions", "subtask_template", "", "subtask_generic_user_input_with_extensions.json") - reg.RegisterArtifact("generic_external_review", "subtask_template", "", "subtask_generic_external_review.json") - return reg + m[artifact.ManifestPath(1)] = []byte(`{"artifacts": [ + {"id": "test_template", "kind": "task_template", "path": "task_test_template.json"}, + {"id": "test_workflow_v1", "kind": "workflow", "path": "wf_test_workflow_v1.json"}, + {"id": "test_render_config", "kind": "generic_template", "path": "generic_render_config.json"}, + {"id": "generic_user_input", "kind": "subtask_template", "path": "subtask_generic_user_input.json"}, + {"id": "generic_user_input_with_extensions", "kind": "subtask_template", "path": "subtask_generic_user_input_with_extensions.json"}, + {"id": "generic_external_review", "kind": "subtask_template", "path": "subtask_generic_external_review.json"} + ]}`) + set, err := artifact.LoadSet(context.Background(), m) + if err != nil { + panic(err) + } + return set } // --------------------------------------------------------------------------- @@ -297,13 +303,14 @@ func TestTaskManager_Lifecycle(t *testing.T) { // 1. StartTask payload := engine.TaskPayload{ - WorkflowID: "parent-workflow", - RunID: "parent-run", - NodeID: "node-1", - ActivationID: "step-1", - TaskTemplateID: "test_template", - Inputs: map[string]any{"userform.name": "Alice"}, - CaseID: "parent-workflow", + ManifestVersion: 1, + WorkflowID: "parent-workflow", + RunID: "parent-run", + NodeID: "node-1", + ActivationID: "step-1", + TaskTemplateID: "test_template", + Inputs: map[string]any{"userform.name": "Alice"}, + CaseID: "parent-workflow", } if _, err := tm.StartTask(context.Background(), payload); err != nil && !errors.Is(err, activity.ErrResultPending) { @@ -331,12 +338,13 @@ func TestTaskManager_Lifecycle(t *testing.T) { // 2. StartTaskStep — generic_user_input payloadTaskWF := engine.TaskPayload{ - WorkflowID: task.TaskWorkflowID, - RunID: "task-run", - NodeID: "task-node", - ActivationID: "task-step", - Seq: 1, - TaskTemplateID: "generic_user_input", + ManifestVersion: 1, + WorkflowID: task.TaskWorkflowID, + RunID: "task-run", + NodeID: "task-node", + ActivationID: "task-step", + Seq: 1, + TaskTemplateID: "generic_user_input", } if _, err := tm.StartTaskStep(context.Background(), payloadTaskWF); err != nil && !errors.Is(err, activity.ErrResultPending) { t.Fatalf("StartTaskStep failed: %v", err) @@ -408,8 +416,9 @@ func TestStartTask_UnknownTemplateID(t *testing.T) { tm := newTestTaskManager(newSafeMockTaskStore(), newTestRegistry(), &mockTemporalManager{}, noopCallback) _, err := tm.StartTask(context.Background(), engine.TaskPayload{ - WorkflowID: "parent-wf", - TaskTemplateID: "does_not_exist", + ManifestVersion: 1, + WorkflowID: "parent-wf", + TaskTemplateID: "does_not_exist", }) if err == nil { t.Fatal("expected error for unknown template ID, got nil") @@ -426,8 +435,9 @@ func TestStartTask_TaskWorkflowManagerError(t *testing.T) { tm := newTestTaskManager(newSafeMockTaskStore(), newTestRegistry(), mockTaskWF, noopCallback) _, err := tm.StartTask(context.Background(), engine.TaskPayload{ - WorkflowID: "parent-wf", - TaskTemplateID: "test_template", + ManifestVersion: 1, + WorkflowID: "parent-wf", + TaskTemplateID: "test_template", }) if err == nil { t.Fatal("expected error when task StartWorkflow fails, got nil") @@ -454,9 +464,10 @@ func TestStartTask_InitTaskErrorPreventsWorkflowStart(t *testing.T) { tm := newTestTaskManager(taskStore, newTestRegistry(), mockTaskWF, noopCallback) _, err := tm.StartTask(context.Background(), engine.TaskPayload{ - WorkflowID: "parent-wf", - ActivationID: "step-1", - TaskTemplateID: "test_template", + ManifestVersion: 1, + WorkflowID: "parent-wf", + ActivationID: "step-1", + TaskTemplateID: "test_template", }) if err == nil { t.Fatal("expected error when InitTask fails, got nil") @@ -486,12 +497,13 @@ func TestStartTask_PassesThroughCaseID(t *testing.T) { tm := newTestTaskManager(db, newTestRegistry(), &mockTemporalManager{}, noopCallback) payload := engine.TaskPayload{ - WorkflowID: tc.workflowID, - RunID: "run-1", - NodeID: "node-1", - ActivationID: "step-1", - TaskTemplateID: "test_template", - CaseID: tc.caseID, + ManifestVersion: 1, + WorkflowID: tc.workflowID, + RunID: "run-1", + NodeID: "node-1", + ActivationID: "step-1", + TaskTemplateID: "test_template", + CaseID: tc.caseID, } if _, err := tm.StartTask(context.Background(), payload); err != nil && !errors.Is(err, activity.ErrResultPending) { t.Fatalf("StartTask failed: %v", err) @@ -516,8 +528,9 @@ func TestStartTaskStep_UnknownWorkflowID(t *testing.T) { tm := newTestTaskManager(newSafeMockTaskStore(), newTestRegistry(), &mockTemporalManager{}, noopCallback) _, err := tm.StartTaskStep(context.Background(), engine.TaskPayload{ - WorkflowID: "workflow-that-was-never-registered", - TaskTemplateID: "generic_user_input", + ManifestVersion: 1, + WorkflowID: "workflow-that-was-never-registered", + TaskTemplateID: "generic_user_input", }) if err == nil { t.Fatal("expected error for unknown workflow ID, got nil") @@ -527,17 +540,19 @@ func TestStartTaskStep_UnknownWorkflowID(t *testing.T) { func TestStartTaskStep_UnknownTaskTemplateID(t *testing.T) { db := newSafeMockTaskStore() db.InitTask(context.Background(), store.TaskRecord{ - TaskID: "task-1", - TaskWorkflowID: "task-workflow-1", - State: "STARTING", - Data: map[string]any{}, + ManifestVersion: 1, + TaskID: "task-1", + TaskWorkflowID: "task-workflow-1", + State: "STARTING", + Data: map[string]any{}, }) tm := newTestTaskManager(db, newTestRegistry(), &mockTemporalManager{}, noopCallback) _, err := tm.StartTaskStep(context.Background(), engine.TaskPayload{ - WorkflowID: "task-workflow-1", - TaskTemplateID: "not_a_real_template", + ManifestVersion: 1, + WorkflowID: "task-workflow-1", + TaskTemplateID: "not_a_real_template", }) if err == nil { t.Fatal("expected error for unknown task_template_id in StartTaskStep, got nil") @@ -551,10 +566,11 @@ func TestStartTaskStep_ExternalReviewPath(t *testing.T) { ) db := newSafeMockTaskStore() db.InitTask(context.Background(), store.TaskRecord{ - TaskID: taskID, - TaskWorkflowID: "task-ext-workflow", - State: "STARTING", - Data: map[string]any{}, + ManifestVersion: 1, + TaskID: taskID, + TaskWorkflowID: "task-ext-workflow", + State: "STARTING", + Data: map[string]any{}, }) // The dispatcher receives the callback token for the step, not the bare task ID. @@ -567,12 +583,13 @@ func TestStartTaskStep_ExternalReviewPath(t *testing.T) { tm := NewTaskManager(db, newTestRegistry(), pr, nil, &mockTemporalManager{}, noopCallback, noopRenderer{}) _, err := tm.StartTaskStep(context.Background(), engine.TaskPayload{ - WorkflowID: "task-ext-workflow", - RunID: "run-1", - NodeID: "node-ext", - ActivationID: stepID, - Seq: 1, - TaskTemplateID: "generic_external_review", + ManifestVersion: 1, + WorkflowID: "task-ext-workflow", + RunID: "run-1", + NodeID: "node-ext", + ActivationID: stepID, + Seq: 1, + TaskTemplateID: "generic_external_review", }) if err != nil && !errors.Is(err, activity.ErrResultPending) { t.Fatalf("StartTaskStep for generic_external_review failed: %v", err) @@ -608,9 +625,10 @@ func TestCompleteTaskStep_UnknownTaskID(t *testing.T) { func TestCompleteTaskStep_AlreadyCompleted(t *testing.T) { db := newSafeMockTaskStore() db.InitTask(context.Background(), store.TaskRecord{ - TaskID: "task-done", - State: "COMPLETED", - Data: map[string]any{}, + ManifestVersion: 1, + TaskID: "task-done", + State: "COMPLETED", + Data: map[string]any{}, }) tm := newTestTaskManager(db, newTestRegistry(), &mockTemporalManager{}, noopCallback) @@ -624,9 +642,10 @@ func TestCompleteTaskStep_AlreadyCompleted(t *testing.T) { func TestCompleteTaskStep_NoActiveStep(t *testing.T) { db := newSafeMockTaskStore() db.InitTask(context.Background(), store.TaskRecord{ - TaskID: "task-starting", - State: "STARTING", - Data: map[string]any{}, + ManifestVersion: 1, + TaskID: "task-starting", + State: "STARTING", + Data: map[string]any{}, }) tm := newTestTaskManager(db, newTestRegistry(), &mockTemporalManager{}, noopCallback) @@ -779,6 +798,7 @@ func TestTaskManager_ExtensionsPipeline(t *testing.T) { // Setup a task record with active step configuration record := store.TaskRecord{ + ManifestVersion: 1, TaskID: "test-task-ext", TaskType: "TEST", State: "PENDING_USER", @@ -853,7 +873,7 @@ func TestTaskManager_ExtensionsPipeline(t *testing.T) { // The task view and the list carry the step to act on and the version. func TestTaskViews_CarryStepIDAndVersion(t *testing.T) { db := newSafeMockTaskStore() - db.InitTask(context.Background(), store.TaskRecord{TaskID: "task-v", TaskType: "TEST", State: "PENDING_USER", ActiveStepID: "step-9", Seq: 9, Data: map[string]any{}}) + db.InitTask(context.Background(), store.TaskRecord{ManifestVersion: 1, TaskID: "task-v", TaskType: "TEST", State: "PENDING_USER", ActiveStepID: "step-9", Seq: 9, Data: map[string]any{}}) tm := newTestTaskManager(db, newTestRegistry(), &mockTemporalManager{}, noopCallback) view, err := tm.GetTaskRenderInfo(context.Background(), "task-v") diff --git a/taskflow/orchestrator/manifest_version_test.go b/taskflow/orchestrator/manifest_version_test.go new file mode 100644 index 00000000..51e861f6 --- /dev/null +++ b/taskflow/orchestrator/manifest_version_test.go @@ -0,0 +1,246 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package orchestrator + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/temporal" + + "github.com/OpenNSW/core/artifact" + "github.com/OpenNSW/core/artifact/testutil" + "github.com/OpenNSW/core/taskflow/extensions" + "github.com/OpenNSW/core/taskflow/renderer" + "github.com/OpenNSW/core/taskflow/store" + engine "github.com/OpenNSW/core/workflow" +) + +// twoManifestSet has manifest v1 and v2 for the same IDs. v2 changes the task template's type +// and the user input step's output namespace, so a test can tell which version a call used. +// v2 is current. +func twoManifestSet(t *testing.T) *artifact.Set { + t.Helper() + m := testutil.MemLoader{ + "task_v1.json": []byte(`{"id":"apply","type":"TYPE_V1","workflow_id":"apply-flow","render_config_id":"apply-render"}`), + "task_v2.json": []byte(`{"id":"apply","type":"TYPE_V2","workflow_id":"apply-flow","render_config_id":"apply-render"}`), + "flow.json": []byte(`{"id":"apply-flow","nodes":[]}`), + "render.json": []byte(`{}`), + "step_v1.json": []byte(`{"id":"apply-form","task_type":"USER_INPUT","output_namespace":"form_v1","plugin_properties":{},"extensions":[{"id":"probe","phase":"PRE_RESUME"}]}`), + "step_v2.json": []byte(`{"id":"apply-form","task_type":"USER_INPUT","output_namespace":"form_v2","plugin_properties":{},"extensions":[{"id":"probe","phase":"PRE_RESUME"}]}`), + "manifest_v1.json": []byte(`{"artifacts":[ + {"id":"apply","kind":"task_template","path":"task_v1.json"}, + {"id":"apply-flow","kind":"workflow","path":"flow.json"}, + {"id":"apply-render","kind":"generic_template","path":"render.json"}, + {"id":"apply-form","kind":"subtask_template","path":"step_v1.json"}]}`), + "manifest_v2.json": []byte(`{"artifacts":[ + {"id":"apply","kind":"task_template","path":"task_v2.json"}, + {"id":"apply-flow","kind":"workflow","path":"flow.json"}, + {"id":"apply-render","kind":"generic_template","path":"render.json"}, + {"id":"apply-form","kind":"subtask_template","path":"step_v2.json"}]}`), + } + set, err := artifact.LoadSet(context.Background(), m) + if err != nil { + t.Fatalf("LoadSet: %v", err) + } + if set.Current() != 2 { + t.Fatalf("current = %d, want 2", set.Current()) + } + return set +} + +// A task records the version its parent workflow carries, uses that version's templates, and +// passes the version on to its own workflow. +func TestStartTask_PinsTheParentsManifestVersion(t *testing.T) { + db := newSafeMockTaskStore() + var started map[string]any + wf := &mockTemporalManager{startWorkflowFunc: func(_ context.Context, _ string, _ engine.WorkflowDefinition, vars map[string]any) error { + started = vars + return nil + }} + tm := newTestTaskManager(db, twoManifestSet(t), wf, noopCallback) + + _, err := tm.StartTask(context.Background(), engine.TaskPayload{ + ManifestVersion: 1, + WorkflowID: "parent", + ActivationID: "6f1d2c3b-4a59-4e8d-9c7b-1a2b3c4d5e6f", + TaskTemplateID: "apply", + }) + if !errors.Is(err, activity.ErrResultPending) { + t.Fatalf("StartTask: %v", err) + } + rec, ok := db.GetTask(context.Background(), "6f1d2c3b-4a59-4e8d-9c7b-1a2b3c4d5e6f") + if !ok { + t.Fatal("task record not created") + } + if rec.ManifestVersion != 1 { + t.Errorf("record ManifestVersion = %d, want 1", rec.ManifestVersion) + } + if rec.TaskType != "TYPE_V1" { + t.Errorf("task type = %q, want TYPE_V1 from manifest v1 (v2 is current)", rec.TaskType) + } + if started[engine.VarManifestVersion] != 1 { + t.Errorf("task workflow started with %s = %v, want 1", engine.VarManifestVersion, started[engine.VarManifestVersion]) + } +} + +// A task without a manifest version is refused before anything is written, and the error is not +// retryable: no retry can supply the version. +func TestStartTask_WithoutManifestVersionFailsWithoutRetry(t *testing.T) { + db := newSafeMockTaskStore() + startedWorkflow := false + wf := &mockTemporalManager{startWorkflowFunc: func(context.Context, string, engine.WorkflowDefinition, map[string]any) error { + startedWorkflow = true + return nil + }} + tm := newTestTaskManager(db, twoManifestSet(t), wf, noopCallback) + + _, err := tm.StartTask(context.Background(), engine.TaskPayload{ + WorkflowID: "parent", + ActivationID: "7a1d2c3b-4a59-4e8d-9c7b-1a2b3c4d5e6f", + TaskTemplateID: "apply", + }) + if !errors.Is(err, artifact.ErrUnknownManifestVersion) { + t.Fatalf("expected ErrUnknownManifestVersion, got %v", err) + } + // Temporal reads retryability off the returned error itself, so it must not be wrapped. + appErr, ok := err.(*temporal.ApplicationError) + if !ok || !appErr.NonRetryable() { + t.Errorf("expected an unwrapped non-retryable application error, got %T: %v", err, err) + } + if _, ok := db.GetTask(context.Background(), "7a1d2c3b-4a59-4e8d-9c7b-1a2b3c4d5e6f"); ok { + t.Error("a task record was written") + } + if startedWorkflow { + t.Error("the task workflow was started") + } +} + +// A version this replica has not loaded may still be published, so that error stays retryable. +func TestStartTask_UnpublishedVersionStaysRetryable(t *testing.T) { + tm := newTestTaskManager(newSafeMockTaskStore(), twoManifestSet(t), &mockTemporalManager{}, noopCallback) + _, err := tm.StartTask(context.Background(), engine.TaskPayload{ + ManifestVersion: 9, + WorkflowID: "parent", + ActivationID: "8b1d2c3b-4a59-4e8d-9c7b-1a2b3c4d5e6f", + TaskTemplateID: "apply", + }) + if !errors.Is(err, artifact.ErrUnknownManifestVersion) { + t.Fatalf("expected ErrUnknownManifestVersion, got %v", err) + } + var appErr *temporal.ApplicationError + if errors.As(err, &appErr) && appErr.NonRetryable() { + t.Errorf("an unpublished version should be retryable, got %v", err) + } +} + +// Steps resolve through the task record's version, even when the step payload and the current +// manifest say otherwise, and the version reaches extensions through the context. +func TestSteps_UseTheTaskRecordsManifestVersion(t *testing.T) { + const ( + taskID = "1c2d3e4f-5a6b-4c7d-8e9f-0a1b2c3d4e5f" + stepID = "2d3e4f5a-6b7c-4d8e-9f0a-1b2c3d4e5f6a" + ) + ctx := context.Background() + db := newSafeMockTaskStore() + db.InitTask(ctx, store.TaskRecord{ + ManifestVersion: 1, + TaskID: taskID, + TaskWorkflowID: "task-wf", + State: "STARTING", + Data: map[string]any{}, + }) + + var extensionVersion int + var extensionSawVersion bool + extReg := extensions.NewRegistry() + extReg.Register("probe", &mockExtension{executeFunc: func(ctx context.Context, _ *store.TaskRecord, _ map[string]any, _ json.RawMessage) error { + extensionVersion, extensionSawVersion = artifact.ManifestVersionFrom(ctx) + return nil + }}) + tm := NewTaskManager(db, twoManifestSet(t), newTestPluginsRegistry(), extReg, &mockTemporalManager{}, noopCallback, noopRenderer{}) + + _, err := tm.StartTaskStep(ctx, engine.TaskPayload{ + ManifestVersion: 2, // ignored: the record is authoritative + WorkflowID: "task-wf", + ActivationID: stepID, + Seq: 1, + TaskTemplateID: "apply-form", + }) + if err != nil && !errors.Is(err, activity.ErrResultPending) { + t.Fatalf("StartTaskStep: %v", err) + } + + if err := tm.CompleteTaskStep(ctx, taskID, stepID, map[string]any{"name": "Ann"}); err != nil { + t.Fatalf("CompleteTaskStep: %v", err) + } + rec, _ := db.GetTask(ctx, taskID) + if _, ok := rec.Data["form_v1"]; !ok { + t.Errorf("submission stored under %v, want the v1 step's namespace form_v1", keys(rec.Data)) + } + if !extensionSawVersion || extensionVersion != 1 { + t.Errorf("extension saw manifest version %d (present %v), want 1", extensionVersion, extensionSawVersion) + } +} + +type versionRecordingRenderer struct { + version int + ok bool +} + +func (r *versionRecordingRenderer) Render(ctx context.Context, _ json.RawMessage, _ renderer.Facts) (json.RawMessage, error) { + r.version, r.ok = artifact.ManifestVersionFrom(ctx) + return json.RawMessage(`{}`), nil +} + +// Rendering a task hands the renderer the task's version, so its template provider resolves the +// render config's templates through it. +func TestGetTaskRenderInfo_PassesTheTasksManifestVersion(t *testing.T) { + ctx := context.Background() + db := newSafeMockTaskStore() + db.InitTask(ctx, store.TaskRecord{ManifestVersion: 1, TaskID: "task-r", State: "PENDING_USER", Data: map[string]any{}}) + r := &versionRecordingRenderer{} + tm := NewTaskManager(db, twoManifestSet(t), newTestPluginsRegistry(), nil, &mockTemporalManager{}, noopCallback, r) + + if _, err := tm.GetTaskRenderInfo(ctx, "task-r"); err != nil { + t.Fatalf("GetTaskRenderInfo: %v", err) + } + if !r.ok || r.version != 1 { + t.Errorf("renderer saw manifest version %d (present %v), want 1", r.version, r.ok) + } +} + +func keys(m map[string]any) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + return out +} + +// A step of a task recorded without a version fails without retrying, before the plugin runs. +func TestStartTaskStep_RecordWithoutManifestVersionFailsWithoutRetry(t *testing.T) { + ctx := context.Background() + db := newSafeMockTaskStore() + db.InitTask(ctx, store.TaskRecord{TaskID: "task-0", TaskWorkflowID: "task-wf-0", State: "STARTING", Data: map[string]any{}}) + tm := newTestTaskManager(db, twoManifestSet(t), &mockTemporalManager{}, noopCallback) + + _, err := tm.StartTaskStep(ctx, engine.TaskPayload{ + ManifestVersion: 2, + WorkflowID: "task-wf-0", + ActivationID: "3e4f5a6b-7c8d-4e9f-8a0b-1c2d3e4f5a6b", + Seq: 1, + TaskTemplateID: "apply-form", + }) + appErr, ok := err.(*temporal.ApplicationError) + if !ok || !appErr.NonRetryable() || !errors.Is(err, artifact.ErrUnknownManifestVersion) { + t.Fatalf("expected an unwrapped non-retryable ErrUnknownManifestVersion, got %T: %v", err, err) + } + if rec, _ := db.GetTask(ctx, "task-0"); rec.ActiveStepID != "" { + t.Errorf("the step was claimed (active step %q)", rec.ActiveStepID) + } +} diff --git a/taskflow/orchestrator/start_subtask_test.go b/taskflow/orchestrator/start_subtask_test.go index 7e05746d..4fd54a8b 100644 --- a/taskflow/orchestrator/start_subtask_test.go +++ b/taskflow/orchestrator/start_subtask_test.go @@ -47,7 +47,8 @@ func startTaskStepFixture(t *testing.T) (*TaskManager, *safeMockTaskStore, *prob t.Helper() db := newSafeMockTaskStore() db.InitTask(context.Background(), store.TaskRecord{ - TaskID: "task-1", TaskType: "TEST", State: "STARTING", TaskWorkflowID: "task-wf", + ManifestVersion: 1, + TaskID: "task-1", TaskType: "TEST", State: "STARTING", TaskWorkflowID: "task-wf", Data: map[string]any{"old": "value"}, }) return startTaskStepFixtureOver(t, db) @@ -67,7 +68,8 @@ func startTaskStepFixtureOver(t *testing.T, db *safeMockTaskStore) (*TaskManager func stepPayload(step string, seq int64, inputs map[string]any) engine.TaskPayload { return engine.TaskPayload{ - WorkflowID: "task-wf", RunID: "run-1", NodeID: "form", ActivationID: step, Seq: seq, + ManifestVersion: 1, + WorkflowID: "task-wf", RunID: "run-1", NodeID: "form", ActivationID: step, Seq: seq, TaskTemplateID: "generic_user_input", Inputs: inputs, } } diff --git a/taskflow/renderer/zoneview/zone_assembler.go b/taskflow/renderer/zoneview/zone_assembler.go index e808f9be..3c790cb9 100644 --- a/taskflow/renderer/zoneview/zone_assembler.go +++ b/taskflow/renderer/zoneview/zone_assembler.go @@ -7,6 +7,7 @@ import ( "context" "fmt" + "github.com/OpenNSW/core/artifact" tfrenderer "github.com/OpenNSW/core/taskflow/renderer" "github.com/OpenNSW/core/taskflow/store" ) @@ -29,7 +30,11 @@ func NewZoneViewAssembler(inner *TaskRenderer) *ZoneViewAssembler { // own. Pass nil when the render config gates nothing on a claim. Because a // hidden section is absent from the projector's output, its handles are dropped // with it — the renderer only decorates sections the projector actually emitted. +// Templates resolve through record.ManifestVersion, which Assemble puts on ctx. func (a *ZoneViewAssembler) Assemble(ctx context.Context, record store.TaskRecord, claims map[string]bool) (ZoneView, error) { + // The template provider resolves the render config's templates through the task's manifest + // version (artifact.ManifestVersionFrom), so a form does not change under an open task. + ctx = artifact.WithManifestVersion(ctx, record.ManifestVersion) view, err := a.inner.Render(ctx, record.RenderConfig, tfrenderer.Facts{ State: record.State, Data: record.Data, diff --git a/taskflow/renderer/zoneview/zone_assembler_test.go b/taskflow/renderer/zoneview/zone_assembler_test.go index 206ae147..0f91f97c 100644 --- a/taskflow/renderer/zoneview/zone_assembler_test.go +++ b/taskflow/renderer/zoneview/zone_assembler_test.go @@ -9,7 +9,9 @@ import ( "strings" "testing" + "github.com/OpenNSW/core/artifact" "github.com/OpenNSW/core/taskflow/store" + "github.com/OpenNSW/core/uiprojector" ) func newTestAssembler(t *testing.T) *ZoneViewAssembler { @@ -170,3 +172,34 @@ func TestAssemble_CarriesStepAndVersionAndOffersNoActionsWhileAdvancing(t *testi } } } + +// versionProbe records the manifest version on the context of each template fetch. +type versionProbe struct{ seen []int } + +func (p *versionProbe) GetTemplate(ctx context.Context, _ string) ([]byte, error) { + v, _ := artifact.ManifestVersionFrom(ctx) + p.seen = append(p.seen, v) + return []byte(`"text"`), nil +} + +// The renderer's template provider sees the task's manifest version. +func TestAssemble_ResolvesTemplatesThroughTheTasksManifestVersion(t *testing.T) { + probe := &versionProbe{} + asm, err := uiprojector.NewAssembler(probe, uiprojector.DefaultProjectors()) + if err != nil { + t.Fatal(err) + } + record := pendingRecord(claimGatedConfig) + record.ManifestVersion = 7 + if _, err := NewZoneViewAssembler(NewTaskRenderer(asm)).Assemble(context.Background(), record, map[string]bool{"role:trader": true, "role:cha": true}); err != nil { + t.Fatalf("Assemble: %v", err) + } + if len(probe.seen) == 0 { + t.Fatal("no template was fetched") + } + for _, v := range probe.seen { + if v != 7 { + t.Errorf("template fetched with manifest version %d, want 7", v) + } + } +} diff --git a/taskflow/store/db.go b/taskflow/store/db.go index d0214f28..03e2dfd1 100644 --- a/taskflow/store/db.go +++ b/taskflow/store/db.go @@ -31,6 +31,11 @@ type TaskRecord struct { // lists a case's tasks by it. CaseID string `json:"case_id"` + // ManifestVersion is the artifact manifest version the task resolves its templates through + // (see workflow.VarManifestVersion): the parent workflow's at the time the task started. Every + // step of the task, and every render of it, uses it. + ManifestVersion int `json:"manifest_version"` + // Active step execution coordinates — used to resume/wake the currently active step via the API. // WARNING: Since the store only holds a single set of coordinates, only one step can be active at any given time // (strictly sequential execution). Parallel/concurrent steps inside a single Task Workflow are not supported. diff --git a/taskflow/store/gorm/model.go b/taskflow/store/gorm/model.go index 483d39b4..cf98d4f8 100644 --- a/taskflow/store/gorm/model.go +++ b/taskflow/store/gorm/model.go @@ -19,6 +19,7 @@ type TaskRecordModel struct { RenderConfig json.RawMessage `gorm:"column:render_config;type:jsonb;serializer:json"` ParentWorkflowID string `gorm:"column:parent_workflow_id;type:text;index"` CaseID string `gorm:"column:case_id;type:text;not null;default:''"` + ManifestVersion int `gorm:"column:manifest_version;not null;default:0"` ParentStepID string `gorm:"column:parent_step_id;type:text"` TaskWorkflowID string `gorm:"column:task_workflow_id;type:text;index"` ActiveTaskTemplateID string `gorm:"column:active_task_template_id;type:text"` @@ -59,6 +60,7 @@ func (m TaskRecordModel) ToDomain() store.TaskRecord { ParentWorkflowID: m.ParentWorkflowID, ParentStepID: m.ParentStepID, CaseID: m.CaseID, + ManifestVersion: m.ManifestVersion, TaskWorkflowID: m.TaskWorkflowID, ActiveTaskTemplateID: m.ActiveTaskTemplateID, ActiveStepID: activeStepID, @@ -88,6 +90,7 @@ func FromDomain(r store.TaskRecord) TaskRecordModel { RenderConfig: r.RenderConfig, ParentWorkflowID: r.ParentWorkflowID, CaseID: r.CaseID, + ManifestVersion: r.ManifestVersion, ParentStepID: r.ParentStepID, TaskWorkflowID: r.TaskWorkflowID, ActiveTaskTemplateID: r.ActiveTaskTemplateID, diff --git a/taskflow/store/gorm/model_test.go b/taskflow/store/gorm/model_test.go index 703cdc79..1d3aff43 100644 --- a/taskflow/store/gorm/model_test.go +++ b/taskflow/store/gorm/model_test.go @@ -29,6 +29,7 @@ func TestRoundTrip_PreservesFields(t *testing.T) { ParentWorkflowID: "consignment-9--node-3--branch-1", ParentStepID: "step-3", CaseID: "consignment-9", + ManifestVersion: 3, TaskWorkflowID: "task-wf-1", ActiveTaskTemplateID: "tmpl-1", ActiveStepID: "0a0a0a0a-0000-4000-8000-00000000000a", @@ -41,6 +42,9 @@ func TestRoundTrip_PreservesFields(t *testing.T) { if got.CaseID != original.CaseID { t.Errorf("CaseID not preserved: got %q, want %q", got.CaseID, original.CaseID) } + if got.ManifestVersion != original.ManifestVersion { + t.Errorf("ManifestVersion: got %d, want %d", got.ManifestVersion, original.ManifestVersion) + } if got.TaskID != original.TaskID { t.Errorf("TaskID: got %q, want %q", got.TaskID, original.TaskID) } diff --git a/taskflow/store/gorm/store_test.go b/taskflow/store/gorm/store_test.go index f20cdd71..b142e1cc 100644 --- a/taskflow/store/gorm/store_test.go +++ b/taskflow/store/gorm/store_test.go @@ -43,6 +43,7 @@ func newTestStore(t *testing.T) (*TaskStore, context.Context) { parent_workflow_id TEXT, parent_step_id TEXT, task_workflow_id TEXT, active_task_template_id TEXT, case_id TEXT NOT NULL DEFAULT '', + manifest_version INTEGER NOT NULL DEFAULT 0, active_step_id TEXT NULL, seq INTEGER NOT NULL DEFAULT 0, data TEXT, created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL)`).Error; err != nil { t.Fatalf("create table: %v", err) diff --git a/workflow/README.md b/workflow/README.md index 331e60f3..16d39f01 100644 --- a/workflow/README.md +++ b/workflow/README.md @@ -91,7 +91,7 @@ Each partition runs as a child workflow, and the join merges its items back by I * **`input_mapping`** (parent path → child path): each partition starts with only the mapped variables, plus the engine's context variables (names starting with `_`, such as - `_case_id`). A missing source parks the split unless its key ends in `?`. A split + `_case_id` and `_manifest_version`). A missing source parks the split unless its key ends in `?`. A split without an `input_mapping` gives its partitions no workflow variables besides their items. * **`output_mapping`** (child path → parent path): the parent path is replaced with a flat list of the unique values the partitions hold at the child path, in partition order. A @@ -197,10 +197,20 @@ manager := engine.NewTemporalManager( completionHandler, // WorkflowCompletionHandler ) -// Register sub-workflow definition loader (required if using SPLIT_TASK nodes) -manager.RegisterDefinitionHandler(func(templateID string) (engine.WorkflowDefinition, error) { - // Retrieve definition from database or local files - return loadDefinition(templateID), nil +// Register sub-workflow definition loader (required if using SPLIT_TASK nodes). It resolves +// templateID in the manifest version the calling workflow carries (VarManifestVersion). +manager.RegisterDefinitionHandler(func(manifestVersion int, templateID string) (engine.WorkflowDefinition, error) { + reg, err := artifacts.Registry(ctx, manifestVersion) + if err != nil { + return engine.WorkflowDefinition{}, err + } + return workflowdef.Load(ctx, reg, templateID) +}) + +// Start a top-level workflow on the current manifest version. The engine carries it to every +// child workflow and reports it on every task activation (TaskPayload.ManifestVersion). +err := manager.StartWorkflow(ctx, id, def, map[string]any{ + engine.VarManifestVersion: artifacts.Current(), }) // Start the internal worker to begin execution diff --git a/workflow/activities.go b/workflow/activities.go index 4adb2a8c..17dd830d 100644 --- a/workflow/activities.go +++ b/workflow/activities.go @@ -27,19 +27,21 @@ type Activities struct { // FetchWorkflowDefinitionHandler is invoked to dynamically retrieve the workflow definition structure // for a given template ID during SPLIT_TASK execution. - FetchWorkflowDefinitionHandler func(templateID string) (WorkflowDefinition, error) + FetchWorkflowDefinitionHandler DefinitionHandler // AdminParkHandler is invoked whenever a node parks for admin intervention. See the // AdminParkHandler type doc for what a nil handler means. AdminParkHandler AdminParkHandler } -// FetchWorkflowDefinitionActivity is a Temporal activity that retrieves the workflow definition for a template ID. -func (a *Activities) FetchWorkflowDefinitionActivity(_ context.Context, templateID string) (WorkflowDefinition, error) { +// FetchWorkflowDefinitionActivity is a Temporal activity that retrieves the workflow definition for +// a template ID, from the artifact manifest version the calling workflow resolves through (see +// VarManifestVersion). +func (a *Activities) FetchWorkflowDefinitionActivity(_ context.Context, templateID string, manifestVersion int) (WorkflowDefinition, error) { if a.FetchWorkflowDefinitionHandler == nil { return WorkflowDefinition{}, fmt.Errorf("FetchWorkflowDefinitionHandler is not initialized on Activities wrapper") } - return a.FetchWorkflowDefinitionHandler(templateID) + return a.FetchWorkflowDefinitionHandler(manifestVersion, templateID) } // ActivationRef is what the workflow tells ExecuteTaskActivity about the run it is starting, beyond what @@ -49,20 +51,23 @@ type ActivationRef struct { NodeID string `json:"node_id"` // Seq is the workflow-wide step counter value for this run. Seq int64 `json:"seq"` + // ManifestVersion is the workflow's artifact manifest version (see VarManifestVersion). + ManifestVersion int `json:"manifest_version,omitempty"` } // ExecuteTaskActivity pushes the task to your application and sleeps waiting for it or completes synchronously func (a *Activities) ExecuteTaskActivity(ctx context.Context, taskTemplateID string, inputs map[string]any, caseID string, step ActivationRef) (map[string]any, error) { info := activity.GetInfo(ctx) payload := TaskPayload{ - WorkflowID: info.WorkflowExecution.ID, - RunID: info.WorkflowExecution.RunID, - NodeID: step.NodeID, - ActivationID: info.ActivityID, // the step ID the workflow passed as the ActivityID - Seq: step.Seq, - TaskTemplateID: taskTemplateID, - Inputs: inputs, - CaseID: caseID, + WorkflowID: info.WorkflowExecution.ID, + RunID: info.WorkflowExecution.RunID, + NodeID: step.NodeID, + ActivationID: info.ActivityID, // the step ID the workflow passed as the ActivityID + Seq: step.Seq, + TaskTemplateID: taskTemplateID, + ManifestVersion: step.ManifestVersion, + Inputs: inputs, + CaseID: caseID, } // Trigger custom code block. ExecuteTaskActivityHandler can return error ErrResultPending to pause the workflow diff --git a/workflow/admin_recovery.go b/workflow/admin_recovery.go index f6c0311e..572ea22b 100644 --- a/workflow/admin_recovery.go +++ b/workflow/admin_recovery.go @@ -273,6 +273,7 @@ func (g *graphInterpreter) notifyAdminPark(ctx workflow.Context, node *Node, nod WorkflowID: workflow.GetInfo(ctx).WorkflowExecution.ID, RunID: workflow.GetInfo(ctx).WorkflowExecution.RunID, CaseID: g.caseID(), + ManifestVersion: g.manifestVersion(), NodeID: node.ID, ActivationID: nodeInfo.ActivationID, NodeType: string(node.Type), diff --git a/workflow/admin_recovery_test.go b/workflow/admin_recovery_test.go index 83571508..0247bc85 100644 --- a/workflow/admin_recovery_test.go +++ b/workflow/admin_recovery_test.go @@ -210,12 +210,14 @@ func TestAdminParkNotificationCarriesCategoryAndMappings(t *testing.T) { env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{ "global_user_email": "user@example.com", + VarManifestVersion: 5, }) require.True(t, env.IsWorkflowCompleted()) require.NoError(t, env.GetWorkflowError()) require.Equal(t, ParkCategoryInputMapping, captured.ParkCategory) + require.Equal(t, 5, captured.ManifestVersion, "AdminParkPayload carries the workflow's manifest version") require.Equal(t, map[string]string{"missing_global_var": "local_key"}, captured.InputMapping) require.Empty(t, captured.OutputMapping) require.Equal(t, parkedID, captured.ActivationID, "AdminParkPayload.ActivationID matches NodeInfo's") diff --git a/workflow/dsl.go b/workflow/dsl.go index 83c84753..37de52fd 100644 --- a/workflow/dsl.go +++ b/workflow/dsl.go @@ -83,6 +83,13 @@ const ( // so a descendant at any depth can read it directly without walking a parent-to-parent // chain. VarCaseID = "_case_id" + // VarManifestVersion holds the artifact manifest version the workflow resolves its templates + // through (see artifact.Set). A host sets it when it starts a top-level workflow, from the + // version current at that moment. It is carried unchanged to every child, and reaches the + // host on every task activation (TaskPayload.ManifestVersion) and definition fetch, so a + // deploy never changes the templates of a workflow already running. The engine itself never + // resolves artifacts; a workflow started without it reports 0. + VarManifestVersion = "_manifest_version" // VarBranchID contains the unique branch ID assigned to the specific child workflow branch. VarBranchID = "_branch_id" diff --git a/workflow/dynamic_split.go b/workflow/dynamic_split.go index bad4f8d2..bf05e836 100644 --- a/workflow/dynamic_split.go +++ b/workflow/dynamic_split.go @@ -191,7 +191,7 @@ func (g *graphInterpreter) spawnChildWorkflows( for idx, templateID := range batch { futures[idx] = workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, workflow.ActivityOptions{ StartToCloseTimeout: 15 * time.Second, - }), "FetchWorkflowDefinitionActivity", templateID) + }), "FetchWorkflowDefinitionActivity", templateID, g.manifestVersion()) } for idx, templateID := range batch { @@ -210,6 +210,7 @@ func (g *graphInterpreter) spawnChildWorkflows( childVars := map[string]any{ VarParentWorkflowID: parentInfo.WorkflowExecution.ID, VarCaseID: g.caseID(), + VarManifestVersion: g.manifestVersion(), VarSplitNodeID: node.ID, VarBranchID: p.BranchID, iterKey: map[string]any{ diff --git a/workflow/dynamic_split_test.go b/workflow/dynamic_split_test.go index edcbaf71..b35d43c5 100644 --- a/workflow/dynamic_split_test.go +++ b/workflow/dynamic_split_test.go @@ -102,15 +102,18 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithDifferentTemplates() { } // 4. Mock Definition Hydration Activity Provider - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "oga_phyto_workflow").Return(phytoDef, nil) - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "oga_health_workflow").Return(healthDef, nil) - - // Mock Task Activity Processing Handlers - env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_phyto_inspection", mock.Anything, mock.Anything, mock.Anything).Return(map[string]any{ + // The definitions come from the manifest version the master workflow was started with. + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "oga_phyto_workflow", 7).Return(phytoDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "oga_health_workflow", 7).Return(healthDef, nil) + + // Mock Task Activity Processing Handlers. Each branch carries the master's manifest version + // on its task activations. + pinned := mock.MatchedBy(func(ref ActivationRef) bool { return ref.ManifestVersion == 7 }) + env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_phyto_inspection", mock.Anything, mock.Anything, pinned).Return(map[string]any{ "inspection_status": "APPROVED_PHYTO", }, nil) - env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_health_inspection", mock.Anything, mock.Anything, mock.Anything).Return(map[string]any{ + env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_health_inspection", mock.Anything, mock.Anything, pinned).Return(map[string]any{ "inspection_status": "APPROVED_HEALTH", }, nil) @@ -121,6 +124,7 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithDifferentTemplates() { // 5. Establish initial runtime variables simulating dynamic lookup results initialVars := map[string]any{ + VarManifestVersion: 7, "active_oga_requirements": []map[string]any{ { "template_id": "oga_phyto_workflow", @@ -230,7 +234,7 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithSameTemplateMode() { } // 3. Mock Definition Hydration Activity Provider - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "simple_child_workflow").Return(childDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "simple_child_workflow", mock.Anything).Return(childDef, nil) // Mock Task Activity Processing Handlers env.OnActivity("ExecuteTaskActivity", mock.Anything, "process_item", mock.Anything, mock.Anything, mock.Anything).Return(map[string]any{ @@ -330,7 +334,7 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithCollectAllFailures() { } // Mock Definition Hydration Activity Provider - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "failing_child_workflow").Return(childDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "failing_child_workflow", mock.Anything).Return(childDef, nil) // Mock Task Activity to fail env.OnActivity("ExecuteTaskActivity", mock.Anything, "fail_task", mock.Anything, mock.Anything, mock.Anything).Return(nil, errors.New("task failed intentionally")) @@ -515,8 +519,8 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithCrossBranchBroadcast() { } // 4. Mock Definition Hydration Activity Provider - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "oga_phyto_workflow").Return(phytoDef, nil) - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "oga_health_workflow").Return(healthDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "oga_phyto_workflow", mock.Anything).Return(phytoDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "oga_health_workflow", mock.Anything).Return(healthDef, nil) // Mock Task Activity Processing Handlers env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_phyto_inspection", mock.Anything, mock.Anything, mock.Anything).Return(map[string]any{ @@ -635,8 +639,8 @@ func (s *NSWEngineTestSuite) TestConcurrentSplitTasksDoNotCrossTalkBroadcast() { }, } - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "signal_emitter").Return(emitDef, nil) - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "signal_waiter").Return(waitDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "signal_emitter", mock.Anything).Return(emitDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "signal_waiter", mock.Anything).Return(waitDef, nil) // Master: PARALLEL_SPLIT fans out into two independent SPLIT_TASK groups (A and B), each // running its own emit/wait pair concurrently, then PARALLEL_JOIN converges before END. @@ -788,7 +792,7 @@ func (s *NSWEngineTestSuite) TestChildBranchEndNodeDoesNotFireCompletionHook() { }, Edges: []Edge{{ID: "ce1", SourceID: "c_start", TargetID: "c_end"}}, } - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "child_wf").Return(childDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "child_wf", mock.Anything).Return(childDef, nil) // Guard against concurrent appends under -race if multiple branches fire the hook concurrently. var mu sync.Mutex @@ -835,7 +839,7 @@ func (s *NSWEngineTestSuite) TestChildBranchCompletionHandlerErrorDoesNotHang() }, Edges: []Edge{{ID: "ce1", SourceID: "c_start", TargetID: "c_end"}}, } - env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "child_wf").Return(childDef, nil) + env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "child_wf", mock.Anything).Return(childDef, nil) // The host recognizes only the top-level workflow; synthetic child IDs (which contain "--") // return an error, as a real DB-backed completion handler keyed by workflow ID would. diff --git a/workflow/manager.go b/workflow/manager.go index c7fe90da..1db78905 100644 --- a/workflow/manager.go +++ b/workflow/manager.go @@ -71,8 +71,16 @@ type TaskPayload struct { // of nesting, so it names the case of a task in a deeply nested child, which WorkflowID // alone does not. CaseID string + // ManifestVersion is the artifact manifest version to resolve TaskTemplateID and everything + // it leads to through (see VarManifestVersion). 0 if the host started the workflow without + // one. + ManifestVersion int } +// DefinitionHandler returns the workflow definition templateID names in artifact manifest +// manifestVersion (see VarManifestVersion). The engine calls it when a SPLIT_TASK node fans out. +type DefinitionHandler func(manifestVersion int, templateID string) (WorkflowDefinition, error) + // AdminParkPayload carries the context of a node parking in NodeStatusAwaitingAdmin, delivered // to an AdminParkHandler so the host application can decide how to surface the event — log it, // emit a metric, persist it, page someone, etc. The engine itself takes no position on that; see @@ -84,6 +92,10 @@ type AdminParkPayload struct { RunID string // CaseID is the case the parked workflow belongs to (see VarCaseID). CaseID string + // ManifestVersion is the artifact manifest version the parked workflow resolves its + // templates through (see VarManifestVersion), so a handler that shows or checks the + // parked node's template uses the flow's own version. 0 if the host set none. + ManifestVersion int // NodeID is the ID of the graph node that parked. It names the definition, not this one run of // it; see ActivationID for a run. NodeID string @@ -289,7 +301,7 @@ type TemporalManager interface { Manager // RegisterDefinitionHandler registers the handler function for fetching sub-workflow definitions. - RegisterDefinitionHandler(handler func(templateID string) (WorkflowDefinition, error)) + RegisterDefinitionHandler(handler DefinitionHandler) // RegisterAdminParkHandler registers the handler invoked whenever a node parks for admin // intervention. Optional — see AdminParkHandler's doc for what a nil/unregistered handler @@ -355,7 +367,7 @@ func NewTemporalManager( return m } -func (m *temporalManagerImpl) RegisterDefinitionHandler(handler func(templateID string) (WorkflowDefinition, error)) { +func (m *temporalManagerImpl) RegisterDefinitionHandler(handler DefinitionHandler) { m.activities.FetchWorkflowDefinitionHandler = handler } diff --git a/workflow/manifest_version_test.go b/workflow/manifest_version_test.go new file mode 100644 index 00000000..a1b04135 --- /dev/null +++ b/workflow/manifest_version_test.go @@ -0,0 +1,146 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package engine + +import ( + "context" + "encoding/json" + "sync" + "testing" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/testsuite" + "go.temporal.io/sdk/workflow" +) + +// batchManifestWorkflow: start -> load_items -> BATCH_SPLIT -> (food | goods) -> BATCH_JOIN -> end. +func batchManifestWorkflow() WorkflowDefinition { + return WorkflowDefinition{ + ID: "batch_manifest_test", + Nodes: []Node{ + {ID: "start", Type: NodeTypeStart}, + {ID: "load_items", Type: NodeTypeTask, TaskTemplateID: "LOAD_ITEMS", + OutputMapping: map[string]string{"items": "commodities"}}, + {ID: "gw", Type: NodeTypeGateway, GatewayType: GatewayTypeBatchSplit, + BatchGateway: &BatchGatewayConfig{ItemsVariable: "commodities", IDField: "id"}}, + {ID: "food", Type: NodeTypeTask, TaskTemplateID: "FOOD_INSPECTION"}, + {ID: "goods", Type: NodeTypeTask, TaskTemplateID: "GOODS_INSPECTION"}, + {ID: "join", Type: NodeTypeGateway, GatewayType: GatewayTypeBatchJoin, + BatchJoin: &BatchJoinConfig{GatewayNodeID: "gw", ItemsVariable: "commodities", IDField: "id"}}, + {ID: "end", Type: NodeTypeEnd}, + }, + Edges: []Edge{ + {ID: "e1", SourceID: "start", TargetID: "load_items"}, + {ID: "e2", SourceID: "load_items", TargetID: "gw"}, + {ID: "e3", SourceID: "gw", TargetID: "food", Condition: `item.type == "food"`}, + {ID: "e4", SourceID: "gw", TargetID: "goods", Condition: `item.type == "goods"`}, + {ID: "e5", SourceID: "food", TargetID: "join"}, + {ID: "e6", SourceID: "goods", TargetID: "join"}, + {ID: "e7", SourceID: "join", TargetID: "end"}, + }, + } +} + +// runRecordingActivations runs def with vars and returns the manifest version of every task +// activation, keyed by task template ID. +func runRecordingActivations(t *testing.T, def WorkflowDefinition, vars map[string]any) map[string][]int { + t.Helper() + env := (&testsuite.WorkflowTestSuite{}).NewTestWorkflowEnvironment() + acts := &Activities{} + env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"}) + env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) + env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"}) + env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"}) + mockWorkflowCompletedIgnoringChildren(env) + + var mu sync.Mutex + seen := make(map[string][]int) + env.OnActivity("ExecuteTaskActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return( + func(_ context.Context, templateID string, inputs map[string]any, _ string, ref ActivationRef) (map[string]any, error) { + mu.Lock() + seen[templateID] = append(seen[templateID], ref.ManifestVersion) + mu.Unlock() + if templateID == "LOAD_ITEMS" { + return map[string]any{"items": []any{ + map[string]any{"id": "i1", "type": "food"}, + map[string]any{"id": "i2", "type": "goods"}, + }}, nil + } + if c, ok := inputs["commodities"]; ok { + return map[string]any{"commodities": c}, nil + } + return map[string]any{}, nil + }) + + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "manifest-test"}) + env.ExecuteWorkflow(GraphInterpreterWorkflow, def, vars) + require.True(t, env.IsWorkflowCompleted()) + require.NoError(t, env.GetWorkflowError()) + return seen +} + +// Every task activation, in the top-level workflow and in every child a gateway starts, carries +// the manifest version the host started the workflow with. +func TestManifestVersionReachesEveryTaskActivation(t *testing.T) { + for name, def := range map[string]WorkflowDefinition{ + "BATCH_SPLIT": batchManifestWorkflow(), + "PARALLEL_SPLIT": buildParallelMergeWorkflow(map[string]string{"commodities": "id"}), + } { + t.Run(name, func(t *testing.T) { + seen := runRecordingActivations(t, def, map[string]any{VarManifestVersion: 3}) + require.NotEmpty(t, seen) + children := 0 + for template, versions := range seen { + for _, v := range versions { + require.Equal(t, 3, v, "activation of %s", template) + } + if template != "LOAD_ITEMS" { + children += len(versions) + } + } + require.Equal(t, 2, children, "both child branches should have run a task") + }) + } + + t.Run("a workflow started without a version reports 0", func(t *testing.T) { + seen := runRecordingActivations(t, batchManifestWorkflow(), map[string]any{}) + for template, versions := range seen { + for _, v := range versions { + require.Equal(t, 0, v, "activation of %s", template) + } + } + }) +} + +func TestManifestVersionValueForms(t *testing.T) { + for name, tc := range map[string]struct { + value any + want int + }{ + "int": {4, 4}, + "int64": {int64(4), 4}, + "float64 from JSON": {float64(4), 4}, + "json.Number": {json.Number("4"), 4}, + "absent": {nil, 0}, + "zero": {0, 0}, + "negative int": {-2, 0}, + "negative float": {float64(-2), 0}, + "fractional": {4.5, 0}, + "string": {"4", 0}, + "non-integer Number": {json.Number("4.5"), 0}, + "negative json.Number": {json.Number("-1"), 0}, + } { + t.Run(name, func(t *testing.T) { + vars := map[string]any{} + if tc.value != nil { + vars[VarManifestVersion] = tc.value + } + g := &graphInterpreter{instance: &WorkflowInstance{WorkflowVariables: vars}} + require.Equal(t, tc.want, g.manifestVersion()) + }) + } +} diff --git a/workflow/workflow.go b/workflow/workflow.go index 511b500d..c1681bce 100644 --- a/workflow/workflow.go +++ b/workflow/workflow.go @@ -4,6 +4,7 @@ package engine import ( + "encoding/json" "fmt" "maps" "slices" @@ -429,6 +430,28 @@ func (g *graphInterpreter) mapTaskOutputs(workflowVars map[string]any, outputMap return nil } +// manifestVersion returns the artifact manifest version this execution resolves templates +// through (see VarManifestVersion), or 0 if the host set none. The variable passes through +// Temporal's JSON payloads, so a whole number arrives as a float64; a value that is not a +// positive whole number counts as none. +func (g *graphInterpreter) manifestVersion() int { + switch v := g.instance.WorkflowVariables[VarManifestVersion].(type) { + case int: + return max(v, 0) + case int64: + return int(max(v, 0)) + case float64: + if v > 0 && v == float64(int(v)) { + return int(v) + } + case json.Number: + if n, err := v.Int64(); err == nil && n > 0 { + return int(n) + } + } + return 0 +} + // caseID returns the case this execution belongs to (see VarCaseID). func (g *graphInterpreter) caseID() string { if v, ok := g.instance.WorkflowVariables[VarCaseID].(string); ok && v != "" { @@ -454,7 +477,7 @@ func (g *graphInterpreter) handleTaskNode(ctx workflow.Context, nodeInfo *NodeIn }) err = workflow.ExecuteActivity(nodeCtx, "ExecuteTaskActivity", node.TaskTemplateID, inputs, g.caseID(), - ActivationRef{NodeID: node.ID, Seq: nodeInfo.Seq}).Get(ctx, &result) + ActivationRef{NodeID: node.ID, Seq: nodeInfo.Seq, ManifestVersion: g.manifestVersion()}).Get(ctx, &result) if err != nil { return withCategory(ParkCategoryTaskFailure, err) }