From afada6778ed78f726081958f0c8893ff66e3564f Mon Sep 17 00:00:00 2001 From: "samnayana@gmail.com" Date: Sun, 11 Oct 2026 16:12:21 +0530 Subject: [PATCH 1/2] feat(workflow)!: name the root workflow ID a case ID The ID every workflow of a tree carries was named after the top-level workflow, but it is a grouping key a host may set: a host that runs several top-level workflows for one real-world matter (a consignment, an agency case) starts each with the same ID, so their tasks are grouped together. It is named for that. - VarRootWorkflowID ("_root_workflow_id") becomes VarCaseID ("_case_id"). A top-level workflow started without one is its own case, under its own workflow ID, as before. - TaskPayload.RootWorkflowID and AdminParkPayload.RootWorkflowID become CaseID; store.TaskRecord.RootWorkflowID becomes CaseID (JSON "case_id"), stored in column case_id. GetAllTasks lists a case's tasks. - Child workflow IDs keep the form "--"; the hash folds in the parent's workflow ID, so children of two workflows of one case never collide. BREAKING CHANGE: hosts set and read VarCaseID / CaseID instead of VarRootWorkflowID / RootWorkflowID, and must rename the task records' root_workflow_id column to case_id. Co-Authored-By: Claude Opus 5.5 --- taskflow/orchestrator/manager.go | 6 +--- taskflow/orchestrator/manager_test.go | 32 ++++++++--------- taskflow/store/db.go | 8 +++-- taskflow/store/gorm/model.go | 6 ++-- taskflow/store/gorm/model_test.go | 33 +++++++++-------- taskflow/store/gorm/store.go | 8 ++--- taskflow/store/gorm/store_test.go | 2 +- workflow/README.md | 2 +- workflow/activities.go | 4 +-- workflow/admin_recovery.go | 2 +- workflow/batch_gateway.go | 2 +- workflow/batch_mapping_test.go | 4 +-- workflow/case_id_test.go | 51 +++++++++++++++++++++++++++ workflow/dsl.go | 13 ++++--- workflow/dynamic_split.go | 6 ++-- workflow/manager.go | 12 +++---- workflow/parallel_gateway.go | 2 +- workflow/utils.go | 30 ++++++++-------- workflow/workflow.go | 19 +++++----- 19 files changed, 147 insertions(+), 95 deletions(-) create mode 100644 workflow/case_id_test.go diff --git a/taskflow/orchestrator/manager.go b/taskflow/orchestrator/manager.go index 07fe1a29..fc746b57 100644 --- a/taskflow/orchestrator/manager.go +++ b/taskflow/orchestrator/manager.go @@ -171,10 +171,6 @@ 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, @@ -182,7 +178,7 @@ func (tm *TaskManager) StartTask(ctx context.Context, payload engine.TaskPayload 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(), diff --git a/taskflow/orchestrator/manager_test.go b/taskflow/orchestrator/manager_test.go index 2eaa6182..859a953b 100644 --- a/taskflow/orchestrator/manager_test.go +++ b/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) } }) } diff --git a/taskflow/store/db.go b/taskflow/store/db.go index be10af3a..d0214f28 100644 --- a/taskflow/store/db.go +++ b/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. diff --git a/taskflow/store/gorm/model.go b/taskflow/store/gorm/model.go index df549193..483d39b4 100644 --- a/taskflow/store/gorm/model.go +++ b/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, diff --git a/taskflow/store/gorm/model_test.go b/taskflow/store/gorm/model_test.go index 2deb5f38..703cdc79 100644 --- a/taskflow/store/gorm/model_test.go +++ b/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") } } diff --git a/taskflow/store/gorm/store.go b/taskflow/store/gorm/store.go index 6091d933..5b96f31e 100644 --- a/taskflow/store/gorm/store.go +++ b/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 } diff --git a/taskflow/store/gorm/store_test.go b/taskflow/store/gorm/store_test.go index 99eab163..fbf7d9a1 100644 --- a/taskflow/store/gorm/store_test.go +++ b/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) diff --git a/workflow/README.md b/workflow/README.md index a9acd929..331e60f3 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 - `_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 diff --git a/workflow/activities.go b/workflow/activities.go index 7dd398ec..4adb2a8c 100644 --- a/workflow/activities.go +++ b/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 diff --git a/workflow/admin_recovery.go b/workflow/admin_recovery.go index 0a8baf80..f6c0311e 100644 --- a/workflow/admin_recovery.go +++ b/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), diff --git a/workflow/batch_gateway.go b/workflow/batch_gateway.go index a0a5b7ff..28e67cd0 100644 --- a/workflow/batch_gateway.go +++ b/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, }) diff --git a/workflow/batch_mapping_test.go b/workflow/batch_mapping_test.go index 8f896df3..d3936630 100644 --- a/workflow/batch_mapping_test.go +++ b/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() { diff --git a/workflow/case_id_test.go b/workflow/case_id_test.go new file mode 100644 index 00000000..7cfa5bd7 --- /dev/null +++ b/workflow/case_id_test.go @@ -0,0 +1,51 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package engine + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/testsuite" +) + +const startEndWorkflowJSON = `{ + "id": "start_end", + "nodes": [{"id": "start", "type": "START"}, {"id": "end", "type": "END"}], + "edges": [{"id": "e1", "source_id": "start", "target_id": "end"}] +}` + +// runStartEnd runs a START -> END workflow with vars and returns its final variables. +func runStartEnd(t *testing.T, vars map[string]any) map[string]any { + t.Helper() + env := (&testsuite.WorkflowTestSuite{}).NewTestWorkflowEnvironment() + var def WorkflowDefinition + require.NoError(t, json.Unmarshal([]byte(startEndWorkflowJSON), &def)) + acts := &Activities{} + env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) + env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) + + env.ExecuteWorkflow(GraphInterpreterWorkflow, def, vars) + require.True(t, env.IsWorkflowCompleted()) + require.NoError(t, env.GetWorkflowError()) + var instance WorkflowInstance + require.NoError(t, env.GetWorkflowResult(&instance)) + return instance.WorkflowVariables +} + +// A host that runs several top-level workflows for one case starts each with the case ID; +// the engine keeps it. +func TestCaseIDFromHostIsKept(t *testing.T) { + vars := runStartEnd(t, map[string]any{VarCaseID: "consignment-1"}) + require.Equal(t, "consignment-1", vars[VarCaseID]) +} + +// A top-level workflow started without a case is its own case. +func TestCaseIDDefaultsToTopLevelWorkflowID(t *testing.T) { + vars := runStartEnd(t, map[string]any{}) + require.Equal(t, "default-test-workflow-id", vars[VarCaseID]) +} diff --git a/workflow/dsl.go b/workflow/dsl.go index 665c41d5..83c84753 100644 --- a/workflow/dsl.go +++ b/workflow/dsl.go @@ -75,11 +75,14 @@ const ( VarSplitNodeID = "_split_node_id" // VarParentWorkflowID contains the workflow ID of the parent/orchestrator workflow. VarParentWorkflowID = "_parent_workflow_id" - // VarRootWorkflowID contains the workflow ID of the top-level (root) execution. It is seeded - // once when the root GraphInterpreterWorkflow starts and carried unchanged through every - // level of nesting, so a descendant at any depth can read it directly without walking a - // parent-to-parent chain. - VarRootWorkflowID = "_root_workflow_id" + // VarCaseID holds the case a workflow belongs to: the key the task store groups tasks by + // (store.TaskRecord.CaseID) and the prefix of every child workflow ID. A host that runs + // several top-level workflows for one real-world matter (a consignment, an agency case) + // starts each with the same VarCaseID; a top-level workflow started without one is its own + // case, under its own workflow ID. It is carried unchanged through every level of nesting, + // so a descendant at any depth can read it directly without walking a parent-to-parent + // chain. + VarCaseID = "_case_id" // 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 863d94ea..bad4f8d2 100644 --- a/workflow/dynamic_split.go +++ b/workflow/dynamic_split.go @@ -209,7 +209,7 @@ func (g *graphInterpreter) spawnChildWorkflows( childVars := map[string]any{ VarParentWorkflowID: parentInfo.WorkflowExecution.ID, - VarRootWorkflowID: g.rootWorkflowID(), + VarCaseID: g.caseID(), VarSplitNodeID: node.ID, VarBranchID: p.BranchID, iterKey: map[string]any{ @@ -219,7 +219,7 @@ func (g *graphInterpreter) spawnChildWorkflows( }, } - deterministicChildID := FormatChildWorkflowID(g.rootWorkflowID(), parentInfo.WorkflowExecution.ID, node.ID, p.BranchID) + deterministicChildID := FormatChildWorkflowID(g.caseID(), parentInfo.WorkflowExecution.ID, node.ID, p.BranchID) childCtx := workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{ WorkflowID: deterministicChildID, }) @@ -240,7 +240,7 @@ func (g *graphInterpreter) spawnChildWorkflows( // can still record every branch that was actually spawned. var startErrors []error for _, p := range prepared { - childID := FormatChildWorkflowID(g.rootWorkflowID(), parentInfo.WorkflowExecution.ID, node.ID, p.BranchID) + childID := FormatChildWorkflowID(g.caseID(), parentInfo.WorkflowExecution.ID, node.ID, p.BranchID) var childExec workflow.Execution if err := activeBranches[childID].Future.GetChildWorkflowExecution().Get(ctx, &childExec); err != nil { startErrors = append(startErrors, fmt.Errorf("failed to start child workflow %s: %w", childID, err)) diff --git a/workflow/manager.go b/workflow/manager.go index bf664803..c7fe90da 100644 --- a/workflow/manager.go +++ b/workflow/manager.go @@ -67,10 +67,10 @@ type TaskPayload struct { TaskTemplateID string // Inputs contains the specific subset of WorkflowVariables mapped to this task's requirements. Inputs map[string]any - // RootWorkflowID is the workflow ID of the top-level execution. WorkflowID alone doesn't - // identify the root for a deeply nested child, so the engine propagates this explicitly from - // the root through every level of nesting (see VarRootWorkflowID). - RootWorkflowID string + // CaseID is the case the workflow belongs to (see VarCaseID). It is the same at every level + // of nesting, so it names the case of a task in a deeply nested child, which WorkflowID + // alone does not. + CaseID string } // AdminParkPayload carries the context of a node parking in NodeStatusAwaitingAdmin, delivered @@ -82,8 +82,8 @@ type AdminParkPayload struct { WorkflowID string // RunID is the unique identifier for this specific execution attempt. RunID string - // RootWorkflowID is the workflow ID of the top-level execution (see TaskPayload.RootWorkflowID). - RootWorkflowID string + // CaseID is the case the parked workflow belongs to (see VarCaseID). + CaseID string // 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 diff --git a/workflow/parallel_gateway.go b/workflow/parallel_gateway.go index f03f6848..d2e5ff69 100644 --- a/workflow/parallel_gateway.go +++ b/workflow/parallel_gateway.go @@ -70,7 +70,7 @@ func (g *graphInterpreter) handleParallelSplitGateway(ctx workflow.Context, node subDef := extractSubGraph(g.def, e.TargetID, joinNodeID) childVars := deepcopy.Map(baseVars) - childWorkflowID := FormatChildWorkflowID(g.rootWorkflowID(), parentInfo.WorkflowExecution.ID, node.ID, edgeID) + childWorkflowID := FormatChildWorkflowID(g.caseID(), parentInfo.WorkflowExecution.ID, node.ID, edgeID) childCtx := workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{ WorkflowID: childWorkflowID, }) diff --git a/workflow/utils.go b/workflow/utils.go index 518dd5bf..47e6ef29 100644 --- a/workflow/utils.go +++ b/workflow/utils.go @@ -45,42 +45,44 @@ const ( // current_executions.workflow_id (varchar(255)). maxWorkflowIDLen = 255 // childIDSuffixLen is the fixed width of the "--<16-hex-char hash>" suffix - // FormatChildWorkflowID appends to the root. + // FormatChildWorkflowID appends to the case ID. childIDSuffixLen = 2 + 16 - // maxRootForChildID is the longest root FormatChildWorkflowID can prefix without the + // maxPrefixForChildID is the longest case ID FormatChildWorkflowID can prefix without the // result exceeding maxWorkflowIDLen. - maxRootForChildID = maxWorkflowIDLen - childIDSuffixLen + maxPrefixForChildID = maxWorkflowIDLen - childIDSuffixLen ) -// FormatChildWorkflowID constructs a deterministic child workflow ID from the root workflow ID, +// FormatChildWorkflowID constructs a deterministic child workflow ID from the case ID, // the parent's own workflow ID, the gateway node ID, and the branch ID. // // The result is always "--<16-hex-char hash>": each level's hash folds in the parent's full // workflow ID, not just this level's nodeID/branchID, so the ID uniquely commits to the full // ancestor chain no matter how deep the nesting goes. Fixed-width output keeps the ID within // Temporal's varchar(255) current_executions.workflow_id column regardless of branch depth — an -// oversized root is capped via boundRootForChildID first, so the guarantee holds independent of -// how long the caller's root workflow ID is. +// oversized case ID is capped via boundPrefixForChildID first, so the guarantee holds independent +// of how long the caller's case ID is. Two workflows of the same case (a host's several +// top-level workflows) share the prefix; the hash, which folds in the parent's workflow ID, keeps +// their children apart. // -// rootWorkflowID is taken as given rather than parsed out of parentWorkflowID — callers already -// have it via VarRootWorkflowID, which the engine propagates explicitly to every level of nesting. -func FormatChildWorkflowID(rootWorkflowID, parentWorkflowID, nodeID, branchID string) string { - root := boundRootForChildID(rootWorkflowID) +// caseID is taken as given rather than parsed out of parentWorkflowID — callers already have it +// via VarCaseID, which the engine propagates explicitly to every level of nesting. +func FormatChildWorkflowID(caseID, parentWorkflowID, nodeID, branchID string) string { + root := boundPrefixForChildID(caseID) sum := sha256.Sum256([]byte(parentWorkflowID + "|" + nodeID + "|" + branchID)) return root + "--" + hex.EncodeToString(sum[:8]) } -// boundRootForChildID caps a root workflow ID at maxRootForChildID so FormatChildWorkflowID's +// boundPrefixForChildID caps a case ID at maxPrefixForChildID so FormatChildWorkflowID's // output can never exceed maxWorkflowIDLen. A root within budget passes through unchanged; an // oversized one is truncated and given a short hash suffix of the full root, so two long roots // that happen to share a prefix still can't collide once capped. -func boundRootForChildID(root string) string { - if len(root) <= maxRootForChildID { +func boundPrefixForChildID(root string) string { + if len(root) <= maxPrefixForChildID { return root } sum := sha256.Sum256([]byte(root)) suffix := hex.EncodeToString(sum[:4]) // 8 hex chars - keep := maxRootForChildID - len(suffix) - 1 + keep := maxPrefixForChildID - len(suffix) - 1 if keep < 0 { keep = 0 } diff --git a/workflow/workflow.go b/workflow/workflow.go index 36a818cc..fcdcbaab 100644 --- a/workflow/workflow.go +++ b/workflow/workflow.go @@ -59,12 +59,12 @@ func GraphInterpreterWorkflow(ctx workflow.Context, def WorkflowDefinition, init Edges: make([]Edge, len(def.Edges)), } - // Only the root execution reaches this with VarRootWorkflowID unset — every nested child is + // Only a top-level execution can reach this with VarCaseID unset — every nested child is // started with it already present in initialWorkflowVariables (copied through wholesale by - // BATCH_SPLIT/PARALLEL_SPLIT, or set explicitly by SPLIT_TASK), so this seeds it exactly once - // per workflow tree, at the true root. - if _, ok := instance.WorkflowVariables[VarRootWorkflowID]; !ok { - instance.WorkflowVariables[VarRootWorkflowID] = instance.ID + // BATCH_SPLIT/PARALLEL_SPLIT, or set explicitly by SPLIT_TASK) — so a top-level workflow the + // host started without a case is its own case. + if _, ok := instance.WorkflowVariables[VarCaseID]; !ok { + instance.WorkflowVariables[VarCaseID] = instance.ID } for _, node := range def.Nodes { @@ -427,10 +427,9 @@ func (g *graphInterpreter) mapTaskOutputs(workflowVars map[string]any, outputMap return nil } -// rootWorkflowID returns this execution's root workflow ID, propagated from the top-level -// execution via VarRootWorkflowID (see GraphInterpreterWorkflow). -func (g *graphInterpreter) rootWorkflowID() string { - if v, ok := g.instance.WorkflowVariables[VarRootWorkflowID].(string); ok && v != "" { +// 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 != "" { return v } return g.instance.ID @@ -452,7 +451,7 @@ func (g *graphInterpreter) handleTaskNode(ctx workflow.Context, nodeInfo *NodeIn StartToCloseTimeout: 24 * time.Hour * 365, }) - err = workflow.ExecuteActivity(nodeCtx, "ExecuteTaskActivity", node.TaskTemplateID, inputs, g.rootWorkflowID(), + err = workflow.ExecuteActivity(nodeCtx, "ExecuteTaskActivity", node.TaskTemplateID, inputs, g.caseID(), ActivationRef{NodeID: node.ID, Seq: nodeInfo.Seq}).Get(ctx, &result) if err != nil { return withCategory(ParkCategoryTaskFailure, err) From 41029681070a284dab7de8061ec5671bdc0563fb Mon Sep 17 00:00:00 2001 From: "samnayana@gmail.com" Date: Sun, 11 Oct 2026 21:12:51 +0530 Subject: [PATCH 2/2] fix(workflow): treat an empty case ID as none, and list tasks by case - A top-level workflow started with an empty VarCaseID is its own case, as one started without it is: the check when seeding matches caseID()'s, and the workflow ID is stored so BATCH_SPLIT/PARALLEL_SPLIT children, which copy the variables, carry it rather than falling back to their own workflow IDs. - TaskManager.GetAllTasks takes a case ID, which is what the store filters on, and lists the case's tasks across all its workflows. The in-memory test store filters on CaseID to match, and a store test lists one case's tasks from two workflows. Co-Authored-By: Claude Opus 5.5 --- taskflow/orchestrator/manager.go | 8 ++++---- taskflow/store/db_test.go | 4 ++-- taskflow/store/gorm/store_test.go | 24 ++++++++++++++++++++++++ workflow/case_id_test.go | 7 +++++++ workflow/workflow.go | 10 ++++++---- 5 files changed, 43 insertions(+), 10 deletions(-) diff --git a/taskflow/orchestrator/manager.go b/taskflow/orchestrator/manager.go index fc746b57..7c5dd1a6 100644 --- a/taskflow/orchestrator/manager.go +++ b/taskflow/orchestrator/manager.go @@ -556,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{ diff --git a/taskflow/store/db_test.go b/taskflow/store/db_test.go index b7b569ec..40b8ab96 100644 --- a/taskflow/store/db_test.go +++ b/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) diff --git a/taskflow/store/gorm/store_test.go b/taskflow/store/gorm/store_test.go index fbf7d9a1..f20cdd71 100644 --- a/taskflow/store/gorm/store_test.go +++ b/taskflow/store/gorm/store_test.go @@ -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) + } +} diff --git a/workflow/case_id_test.go b/workflow/case_id_test.go index 7cfa5bd7..7afb405a 100644 --- a/workflow/case_id_test.go +++ b/workflow/case_id_test.go @@ -49,3 +49,10 @@ func TestCaseIDDefaultsToTopLevelWorkflowID(t *testing.T) { vars := runStartEnd(t, map[string]any{}) require.Equal(t, "default-test-workflow-id", vars[VarCaseID]) } + +// An empty case ID is no case: the workflow is its own case, and the ID is stored so its +// children copy it. +func TestCaseIDEmptyDefaultsToTopLevelWorkflowID(t *testing.T) { + vars := runStartEnd(t, map[string]any{VarCaseID: ""}) + require.Equal(t, "default-test-workflow-id", vars[VarCaseID]) +} diff --git a/workflow/workflow.go b/workflow/workflow.go index fcdcbaab..511b500d 100644 --- a/workflow/workflow.go +++ b/workflow/workflow.go @@ -59,11 +59,13 @@ func GraphInterpreterWorkflow(ctx workflow.Context, def WorkflowDefinition, init Edges: make([]Edge, len(def.Edges)), } - // Only a top-level execution can reach this with VarCaseID unset — every nested child is - // started with it already present in initialWorkflowVariables (copied through wholesale by + // Only a top-level execution can reach this without a case — every nested child is started + // with one already present in initialWorkflowVariables (copied through wholesale by // BATCH_SPLIT/PARALLEL_SPLIT, or set explicitly by SPLIT_TASK) — so a top-level workflow the - // host started without a case is its own case. - if _, ok := instance.WorkflowVariables[VarCaseID]; !ok { + // host started without a case, or with an empty one, is its own case. The check matches + // caseID()'s, and the ID is stored rather than only returned, so every child copies a + // non-empty case ID instead of falling back to its own workflow ID. + if v, _ := instance.WorkflowVariables[VarCaseID].(string); v == "" { instance.WorkflowVariables[VarCaseID] = instance.ID }