feat(workflow)!: name the root workflow ID a case ID

taskflow/orchestrator/manager.go

@@ -171,18 +171,14 @@ func (tm *TaskManager) StartTask(ctx context.Context, payload engine.TaskPayload
 	}
 	initialData["_task_id"] = taskID
 
-	// root_workflow_id is the top-level workflow's ID, propagated by the engine through every
-	// level of nesting (see engine.VarRootWorkflowID).
-	rootWorkflowID := payload.RootWorkflowID
-
 	record := store.TaskRecord{
 		TaskID:           taskID,
 		TaskType:         template.Type,
 		State:            "STARTING",
 		RenderConfig:     renderConfig,
 		ParentWorkflowID: payload.WorkflowID,
 		ParentStepID:     payload.ActivationID, // the parent's Activity to complete
-		RootWorkflowID:   rootWorkflowID,
+		CaseID:           payload.CaseID,       // the same at every level of nesting (see engine.VarCaseID)
 		TaskWorkflowID:   taskWorkflowID,
 		Data:             initialData,
 		CreatedAt:        time.Now(),
@@ -560,10 +556,10 @@ func (tm *TaskManager) GetTaskRenderInfo(context context.Context, taskID string)
 // field is intentionally not populated — callers should use GetTaskRenderInfo to fetch
 // the full rendered view for a specific task.
 //
-// If parentWorkflowID is non-empty, the result is narrowed to tasks spawned by that
-// parent workflow; an empty string returns all tasks.
-func (tm *TaskManager) GetAllTasks(ctx context.Context, parentWorkflowID string) []TaskView {
-	records := tm.db.GetAllTasks(ctx, parentWorkflowID)
+// If caseID is non-empty, the result is narrowed to the case's tasks (TaskRecord.CaseID),
+// across every workflow of the case; an empty string returns all tasks.
+func (tm *TaskManager) GetAllTasks(ctx context.Context, caseID string) []TaskView {
+	records := tm.db.GetAllTasks(ctx, caseID)
 	resList := make([]TaskView, 0, len(records))
 	for _, r := range records {
 		resList = append(resList, TaskView{

taskflow/orchestrator/manager_test.go

@@ -303,7 +303,7 @@ func TestTaskManager_Lifecycle(t *testing.T) {
 		ActivationID:   "step-1",
 		TaskTemplateID: "test_template",
 		Inputs:         map[string]any{"userform.name": "Alice"},
-		RootWorkflowID: "parent-workflow",
+		CaseID:         "parent-workflow",
 	}
 
 	if _, err := tm.StartTask(context.Background(), payload); err != nil && !errors.Is(err, activity.ErrResultPending) {
@@ -324,10 +324,9 @@ func TestTaskManager_Lifecycle(t *testing.T) {
 	if task.ParentWorkflowID != "parent-workflow" {
 		t.Errorf("expected parent workflow 'parent-workflow', got '%s'", task.ParentWorkflowID)
 	}
-	// A top-level workflow ID has no "--" separator, so the root consignment ID
-	// is the workflow ID itself.
-	if task.RootWorkflowID != "parent-workflow" {
-		t.Errorf("expected root workflow 'parent-workflow', got '%s'", task.RootWorkflowID)
+	// The task carries the case the engine passed in the payload.
+	if task.CaseID != "parent-workflow" {
+		t.Errorf("expected case 'parent-workflow', got '%s'", task.CaseID)
 	}
 
 	// 2. StartTaskStep — generic_user_input
@@ -467,19 +466,18 @@ func TestStartTask_InitTaskErrorPreventsWorkflowStart(t *testing.T) {
 	}
 }
 
-// TestStartTask_PassesThroughRootWorkflowID verifies that RootWorkflowID on the
-// stored TaskRecord comes straight from payload.RootWorkflowID, as propagated by
-// the engine (see engine.VarRootWorkflowID) — StartTask no longer derives it by
-// parsing payload.WorkflowID.
-func TestStartTask_PassesThroughRootWorkflowID(t *testing.T) {
+// TestStartTask_PassesThroughCaseID verifies that CaseID on the stored
+// TaskRecord comes straight from payload.CaseID, as propagated by the engine
+// (see engine.VarCaseID), rather than from parsing payload.WorkflowID.
+func TestStartTask_PassesThroughCaseID(t *testing.T) {
 	cases := []struct {
-		name           string
-		workflowID     string
-		rootWorkflowID string
+		name       string
+		workflowID string
+		caseID     string
 	}{
 		{"top-level workflow", "consignment-123", "consignment-123"},
 		{"nested child workflow", "consignment-123--a1b2c3d4e5f60789", "consignment-123"},
-		{"empty root workflow id", "consignment-123", ""},
+		{"empty case id", "consignment-123", ""},
 	}
 
 	for _, tc := range cases {
@@ -493,7 +491,7 @@ func TestStartTask_PassesThroughRootWorkflowID(t *testing.T) {
 				NodeID:         "node-1",
 				ActivationID:   "step-1",
 				TaskTemplateID: "test_template",
-				RootWorkflowID: tc.rootWorkflowID,
+				CaseID:         tc.caseID,
 			}
 			if _, err := tm.StartTask(context.Background(), payload); err != nil && !errors.Is(err, activity.ErrResultPending) {
 				t.Fatalf("StartTask failed: %v", err)
@@ -503,8 +501,8 @@ func TestStartTask_PassesThroughRootWorkflowID(t *testing.T) {
 			if !ok {
 				t.Fatal("expected task record to be saved")
 			}
-			if task.RootWorkflowID != tc.rootWorkflowID {
-				t.Errorf("RootWorkflowID = %q, want %q", task.RootWorkflowID, tc.rootWorkflowID)
+			if task.CaseID != tc.caseID {
+				t.Errorf("CaseID = %q, want %q", task.CaseID, tc.caseID)
 			}
 		})
 	}

taskflow/store/db.go

@@ -27,7 +27,9 @@ type TaskRecord struct {
 	ParentWorkflowID string `json:"parent_workflow_id"`
 	ParentStepID     string `json:"parent_step_id"` // the parent's step ID: the Activity to complete
 
-	RootWorkflowID string `json:"root_workflow_id"`
+	// CaseID is the case the task's workflow belongs to (see workflow.VarCaseID); GetAllTasks
+	// lists a case's tasks by it.
+	CaseID string `json:"case_id"`
 
 	// 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
@@ -112,7 +114,9 @@ type TaskStore interface {
 	InitTask(context context.Context, record TaskRecord) error
 	GetTask(context context.Context, taskID string) (TaskRecord, bool)
 	GetTaskByWorkflowID(context context.Context, workflowID string) (TaskRecord, bool)
-	GetAllTasks(context context.Context, parentWorkflowID string) []TaskRecord
+	// GetAllTasks returns the tasks of a case (TaskRecord.CaseID), or every task when caseID is
+	// empty.
+	GetAllTasks(context context.Context, caseID string) []TaskRecord
 
 	// ClaimStep makes claim.StepID the active step. Guard: the stored seq is <= claim.Seq, so a
 	// retry of the same step matches again and an older step's late attempt does not.

taskflow/store/db_test.go

@@ -33,10 +33,10 @@ func (t *testStore) GetTaskByWorkflowID(_ context.Context, workflowID string) (T
 	return TaskRecord{}, false
 }
 
-func (t *testStore) GetAllTasks(_ context.Context, parentWorkflowID string) []TaskRecord {
+func (t *testStore) GetAllTasks(_ context.Context, caseID string) []TaskRecord {
 	var list []TaskRecord
 	for _, record := range t.tasks {
-		if parentWorkflowID != "" && record.ParentWorkflowID != parentWorkflowID {
+		if caseID != "" && record.CaseID != caseID {
 			continue
 		}
 		list = append(list, record)

taskflow/store/gorm/model.go

@@ -18,7 +18,7 @@ type TaskRecordModel struct {
 	State                string          `gorm:"column:state;type:text"`
 	RenderConfig         json.RawMessage `gorm:"column:render_config;type:jsonb;serializer:json"`
 	ParentWorkflowID     string          `gorm:"column:parent_workflow_id;type:text;index"`
-	RootWorkflowID       string          `gorm:"column:root_workflow_id;type:text;not null;default:''"`
+	CaseID               string          `gorm:"column:case_id;type:text;not null;default:''"`
 	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"`
@@ -58,7 +58,7 @@ func (m TaskRecordModel) ToDomain() store.TaskRecord {
 		RenderConfig:         m.RenderConfig,
 		ParentWorkflowID:     m.ParentWorkflowID,
 		ParentStepID:         m.ParentStepID,
-		RootWorkflowID:       m.RootWorkflowID,
+		CaseID:               m.CaseID,
 		TaskWorkflowID:       m.TaskWorkflowID,
 		ActiveTaskTemplateID: m.ActiveTaskTemplateID,
 		ActiveStepID:         activeStepID,
@@ -87,7 +87,7 @@ func FromDomain(r store.TaskRecord) TaskRecordModel {
 		State:                r.State,
 		RenderConfig:         r.RenderConfig,
 		ParentWorkflowID:     r.ParentWorkflowID,
-		RootWorkflowID:       r.RootWorkflowID,
+		CaseID:               r.CaseID,
 		ParentStepID:         r.ParentStepID,
 		TaskWorkflowID:       r.TaskWorkflowID,
 		ActiveTaskTemplateID: r.ActiveTaskTemplateID,

taskflow/store/gorm/model_test.go

@@ -12,11 +12,11 @@ import (
 )
 
 // TestRoundTrip_PreservesFields verifies that a domain TaskRecord survives a
-// FromDomain -> ToDomain round trip unchanged. RootWorkflowID is the field that
-// regressed: it is derived once at task creation (in the orchestrator) and the
-// GORM mappers must persist and read it back verbatim. FromDomain/ToDomain are
-// deliberately pure pass-throughs — neither re-derives RootWorkflowID — so this
-// test also guards against re-introducing derivation logic in FromDomain.
+// FromDomain -> ToDomain round trip unchanged. CaseID is set once at task
+// creation (in the orchestrator), and the GORM mappers must persist and read
+// it back verbatim. FromDomain/ToDomain are deliberately pure pass-throughs —
+// neither derives CaseID — so this test also guards against derivation logic
+// in FromDomain.
 //
 // CreatedAt/UpdatedAt are intentionally excluded: GORM owns them via
 // autoCreateTime/autoUpdateTime, so FromDomain never copies them.
@@ -28,7 +28,7 @@ func TestRoundTrip_PreservesFields(t *testing.T) {
 		RenderConfig:         json.RawMessage(`{"layout":"form"}`),
 		ParentWorkflowID:     "consignment-9--node-3--branch-1",
 		ParentStepID:         "step-3",
-		RootWorkflowID:       "consignment-9",
+		CaseID:               "consignment-9",
 		TaskWorkflowID:       "task-wf-1",
 		ActiveTaskTemplateID: "tmpl-1",
 		ActiveStepID:         "0a0a0a0a-0000-4000-8000-00000000000a",
@@ -38,8 +38,8 @@ func TestRoundTrip_PreservesFields(t *testing.T) {
 
 	got := FromDomain(original).ToDomain()
 
-	if got.RootWorkflowID != original.RootWorkflowID {
-		t.Errorf("RootWorkflowID not preserved: got %q, want %q", got.RootWorkflowID, original.RootWorkflowID)
+	if got.CaseID != original.CaseID {
+		t.Errorf("CaseID not preserved: got %q, want %q", got.CaseID, original.CaseID)
 	}
 	if got.TaskID != original.TaskID {
 		t.Errorf("TaskID: got %q, want %q", got.TaskID, original.TaskID)
@@ -64,20 +64,19 @@ func TestRoundTrip_PreservesFields(t *testing.T) {
 	}
 }
 
-// TestFromDomain_DoesNotDeriveRootWorkflowID locks in that FromDomain is a pure
-// pass-through for RootWorkflowID. A previous iteration derived it here from
-// ParentWorkflowID; the source of truth now lives in the orchestrator, so
-// FromDomain must persist exactly what the caller set — even when it diverges
-// from ParentWorkflowID's prefix.
-func TestFromDomain_DoesNotDeriveRootWorkflowID(t *testing.T) {
+// TestFromDomain_DoesNotDeriveCaseID locks in that FromDomain is a pure
+// pass-through for CaseID: the orchestrator sets it, so FromDomain must
+// persist exactly what the caller set — even when it diverges from
+// ParentWorkflowID's prefix.
+func TestFromDomain_DoesNotDeriveCaseID(t *testing.T) {
 	r := store.TaskRecord{
 		TaskID:           "task-2",
 		ParentWorkflowID: "some-other-wf--node--branch",
-		RootWorkflowID:   "explicit-root",
+		CaseID:           "explicit-case",
 	}
 
-	if got := FromDomain(r).RootWorkflowID; got != "explicit-root" {
-		t.Errorf("FromDomain re-derived RootWorkflowID: got %q, want %q", got, "explicit-root")
+	if got := FromDomain(r).CaseID; got != "explicit-case" {
+		t.Errorf("FromDomain re-derived CaseID: got %q, want %q", got, "explicit-case")
 	}
 }
 

taskflow/store/gorm/store.go

@@ -68,14 +68,14 @@ func (s *TaskStore) GetTaskByWorkflowID(ctx context.Context, workflowID string)
 	return model.ToDomain(), true
 }
 
-func (s *TaskStore) GetAllTasks(ctx context.Context, parentWorkflowID string) []store.TaskRecord {
+func (s *TaskStore) GetAllTasks(ctx context.Context, caseID string) []store.TaskRecord {
 	var models []TaskRecordModel
 	query := s.db.WithContext(ctx)
-	if parentWorkflowID != "" {
-		query = query.Where("root_workflow_id = ?", parentWorkflowID)
+	if caseID != "" {
+		query = query.Where("case_id = ?", caseID)
 	}
 	if err := query.Find(&models).Error; err != nil {
-		slog.ErrorContext(ctx, "taskflow gorm store: GetAllTasks db error", "parent_workflow_id", parentWorkflowID, "error", err)
+		slog.ErrorContext(ctx, "taskflow gorm store: GetAllTasks db error", "case_id", caseID, "error", err)
 		return nil
 	}
 

taskflow/store/gorm/store_test.go

@@ -42,7 +42,7 @@ func newTestStore(t *testing.T) (*TaskStore, context.Context) {
 		task_id TEXT PRIMARY KEY, task_type TEXT, state TEXT, render_config TEXT,
 		parent_workflow_id TEXT, parent_step_id TEXT,
 		task_workflow_id TEXT,
-		active_task_template_id TEXT, root_workflow_id TEXT NOT NULL DEFAULT '',
+		active_task_template_id TEXT, case_id TEXT NOT NULL DEFAULT '',
 		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)
@@ -235,3 +235,27 @@ func TestStepLifecycle_StaleWritesAreDropped(t *testing.T) {
 	wantRows(t, n, err, 1)
 	wantRow(t, mustGet(t, s, ctx), stepB, 6, store.StateCompleted)
 }
+
+// GetAllTasks lists a case's tasks across every workflow of the case: a host that runs several
+// top-level workflows for one case finds all their tasks by the case ID.
+func TestGetAllTasksListsACaseAcrossWorkflows(t *testing.T) {
+	s, ctx := newTestStore(t)
+	for _, r := range []store.TaskRecord{
+		{TaskID: "export-task", ParentWorkflowID: "C1:export:u1", CaseID: "C1"},
+		{TaskID: "payment-task", ParentWorkflowID: "C1:payment:u2", CaseID: "C1"},
+		{TaskID: "other-case-task", ParentWorkflowID: "C2:export:u3", CaseID: "C2"},
+	} {
+		r.TaskType, r.State, r.Data = "TEST", "STARTING", map[string]any{}
+		if err := s.InitTask(ctx, r); err != nil {
+			t.Fatalf("InitTask %s: %v", r.TaskID, err)
+		}
+	}
+
+	got := map[string]bool{}
+	for _, r := range s.GetAllTasks(ctx, "C1") {
+		got[r.TaskID] = true
+	}
+	if len(got) != 2 || !got["export-task"] || !got["payment-task"] {
+		t.Errorf("GetAllTasks(C1) = %v, want export-task and payment-task", got)
+	}
+}

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
-  `_root_workflow_id`). A missing source parks the split unless its key ends in `?`. A split
+  `_case_id`). 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

workflow/activities.go

@@ -52,7 +52,7 @@ type ActivationRef struct {
 }
 
 // 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, rootWorkflowID string, step ActivationRef) (map[string]any, error) {
+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,
@@ -62,7 +62,7 @@ func (a *Activities) ExecuteTaskActivity(ctx context.Context, taskTemplateID str
 		Seq:            step.Seq,
 		TaskTemplateID: taskTemplateID,
 		Inputs:         inputs,
-		RootWorkflowID: rootWorkflowID,
+		CaseID:         caseID,
 	}
 
 	// Trigger custom code block. ExecuteTaskActivityHandler can return error ErrResultPending to pause the workflow

workflow/admin_recovery.go

@@ -272,7 +272,7 @@ func (g *graphInterpreter) notifyAdminPark(ctx workflow.Context, node *Node, nod
 	payload := AdminParkPayload{
 		WorkflowID:       workflow.GetInfo(ctx).WorkflowExecution.ID,
 		RunID:            workflow.GetInfo(ctx).WorkflowExecution.RunID,
-		RootWorkflowID:   g.rootWorkflowID(),
+		CaseID:           g.caseID(),
 		NodeID:           node.ID,
 		ActivationID:     nodeInfo.ActivationID,
 		NodeType:         string(node.Type),

workflow/batch_gateway.go

@@ -242,7 +242,7 @@ func (g *graphInterpreter) spawnBatchChildren(
 		maputil.SetNestedKey(childVars, itemsVar, toAnySlice(p.Items))
 		childVars[VarScopePath] = childScopePath
 
-		childWorkflowID := FormatChildWorkflowID(g.rootWorkflowID(), parentInfo.WorkflowExecution.ID, nodeID, edgeID)
+		childWorkflowID := FormatChildWorkflowID(g.caseID(), parentInfo.WorkflowExecution.ID, nodeID, edgeID)
 		childCtx := workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{
 			WorkflowID: childWorkflowID,
 		})

workflow/batch_mapping_test.go

@@ -227,7 +227,7 @@ func withTaskInputs(def WorkflowDefinition) WorkflowDefinition {
 		"npqs.reference_number?": "parent_ref",
 		"secret?":                "secret",
 		"commodities?":           "items",
-		"_root_workflow_id?":     "root",
+		"_case_id?":              "root",
 	}
 	for i := range def.Nodes {
 		if def.Nodes[i].Type == NodeTypeTask {
@@ -316,7 +316,7 @@ func (s *BatchGatewayTestSuite) TestBatchValidation_SplitMappings() {
 		{"input: writes the items variable", map[string]string{"npqs.items": "commodities"}, nil, "cannot read or write the items variable"},
 		{"input: reads the items variable", map[string]string{"commodities": "all_items"}, nil, "cannot read or write the items variable"},
 		{"input: reads the items variable, optional", map[string]string{"commodities?": "all_items"}, nil, "cannot read or write the items variable"},
-		{"input: writes an engine variable", map[string]string{"npqs.root": "_root_workflow_id"}, nil, "reserved for the engine"},
+		{"input: writes an engine variable", map[string]string{"npqs.root": "_case_id"}, nil, "reserved for the engine"},
 	}
 	for _, tc := range cases {
 		s.Run(tc.name, func() {