feat(workflow,taskflow)!: resolve a flow's templates through its manifest version

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)

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 {

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)
 ```
 
 ---

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
 }

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 {

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)
 	}