From 4045ebe53bd6f0fc3959c3af90dfc961e735385c Mon Sep 17 00:00:00 2001 From: "samnayana@gmail.com" Date: Mon, 12 Oct 2026 23:43:06 +0530 Subject: [PATCH 1/2] feat(workflow,taskflow)!: resolve a flow's templates through its manifest version A host starts each top-level workflow with VarManifestVersion ("_manifest_version") set to the current artifact manifest version. The engine carries it to every child workflow and reports it to the host on every task activation and definition fetch, and each task records it, so a deploy never changes the templates of a flow already running. - The engine passes the version on ActivationRef (and so TaskPayload.ManifestVersion) and as a second argument to FetchWorkflowDefinitionActivity. Dynamic-split children get it explicitly; batch and parallel children already copy the engine's underscore variables. A whole number that came back through JSON as a float64 is read as the integer. - RegisterDefinitionHandler takes a DefinitionHandler, which receives the manifest version with the template ID. AdminParkPayload carries the version too, so a park handler can show the parked node's template from the flow's own manifest. - TaskManager takes an *artifact.Set. StartTask resolves the task template, its workflow and render config through the payload's version, stores it on the task record and seeds it into the task's own workflow. StartTaskStep and CompleteTaskStep resolve the step template through the record's version, and put it on the context so plugins, extensions and the renderer's template provider can resolve through it too (artifact.ManifestVersionFrom). - A task without a version fails with a non-retryable error before anything is written: falling back to the current version would change templates under it. A version this replica has not loaded stays retryable, since Set.Registry loads newer manifests on demand. - TaskRecord gains ManifestVersion, stored in column manifest_version. BREAKING CHANGE: NewTaskManager takes an *artifact.Set instead of an *artifact.Registry, RegisterDefinitionHandler takes a DefinitionHandler, and FetchWorkflowDefinitionActivity takes the manifest version. Hosts must set VarManifestVersion when starting a top-level workflow and add a manifest_version INTEGER NOT NULL DEFAULT 0 column to the task records table. Workflows already running fail their next task. Co-Authored-By: Claude Opus 5.5 --- README.md | 8 +- taskflow/README.md | 4 +- taskflow/docs/integration-guide.md | 8 +- taskflow/orchestrator/complete_step_test.go | 7 +- taskflow/orchestrator/completion_test.go | 3 +- taskflow/orchestrator/manager.go | 66 ++++- taskflow/orchestrator/manager_test.go | 144 +++++----- .../orchestrator/manifest_version_test.go | 246 ++++++++++++++++++ taskflow/orchestrator/start_subtask_test.go | 6 +- taskflow/store/db.go | 5 + taskflow/store/gorm/model.go | 3 + taskflow/store/gorm/model_test.go | 4 + taskflow/store/gorm/store_test.go | 1 + workflow/README.md | 20 +- workflow/activities.go | 29 ++- workflow/admin_recovery.go | 1 + workflow/admin_recovery_test.go | 2 + workflow/dsl.go | 7 + workflow/dynamic_split.go | 3 +- workflow/dynamic_split_test.go | 32 ++- workflow/manager.go | 16 +- workflow/manifest_version_test.go | 146 +++++++++++ workflow/workflow.go | 25 +- 23 files changed, 666 insertions(+), 120 deletions(-) create mode 100644 taskflow/orchestrator/manifest_version_test.go create mode 100644 workflow/manifest_version_test.go 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/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) } From cd5eaaf7eeeb9ead2c24c1c21cd5ea4fb4932de5 Mon Sep 17 00:00:00 2001 From: "samnayana@gmail.com" Date: Tue, 13 Oct 2026 00:20:29 +0530 Subject: [PATCH 2/2] fix(taskflow): render a task's zone view on its manifest version ZoneViewAssembler.Assemble, which a host's GET /api/v1/tasks/{id} serves, rendered on the caller's context. A host template provider that resolves through the task's manifest version found none there, so every template-backed section failed. Assemble puts record.ManifestVersion on the context, as GetTaskRenderInfo does. Co-Authored-By: Claude Opus 5.5 --- taskflow/renderer/zoneview/zone_assembler.go | 5 +++ .../renderer/zoneview/zone_assembler_test.go | 33 +++++++++++++++++++ 2 files changed, 38 insertions(+) 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) + } + } +}