feat(workflow)!: wait for a step's completion outside its Activity

README.md

@@ -142,8 +142,8 @@ func Build(ctx context.Context, cfg *Config) (*App, error) {
     var tm *orchestrator.TaskManager
     microRunner := workflow.NewTemporalManager(
         temporalClient, cfg.Temporal.Namespace, "MICRO_WORKFLOW_QUEUE",
-        func(p workflow.TaskPayload) (map[string]any, error) { return tm.StartTaskStep(ctx, p) },
-        func(c workflow.WorkflowCompletion) error { return tm.HandleTaskCompletion(ctx, c) },
+        func(ctx context.Context, p workflow.TaskPayload) (map[string]any, error) { return tm.StartTaskStep(ctx, p) },
+        func(ctx context.Context, c workflow.WorkflowCompletion) error { return tm.HandleTaskCompletion(ctx, c) },
     )
     onTaskCompleted := func(parentWorkflowID, parentStepID string, vars map[string]any) error {
         return consignmentService.HandleTaskCompletion(ctx, parentWorkflowID, parentStepID, vars)

taskflow/README.md

@@ -73,11 +73,11 @@ workflowRunner := workflow.NewTemporalManager(
     temporalClient,
     "default", // must match the namespace temporalClient was dialed with
     "MICRO_WORKFLOW_QUEUE",
-    func(payload workflow.TaskPayload) (map[string]any, error) {
-        return tm.StartTaskStep(context.Background(), payload)
+    func(ctx context.Context, payload workflow.TaskPayload) (map[string]any, error) {
+        return tm.StartTaskStep(ctx, payload)
     },
-    func(c workflow.WorkflowCompletion) error {
-        return tm.HandleTaskCompletion(context.Background(), c)
+    func(ctx context.Context, c workflow.WorkflowCompletion) error {
+        return tm.HandleTaskCompletion(ctx, c)
     },
 )
 
@@ -138,7 +138,7 @@ Register it: `pluginRegistry.Register("MY_PLUGIN", &MyPlugin{remoteManager: rm})
 - **The task row is written through guarded statements.** `ActiveStepID` and `Seq` change only via `ClaimStep`, `WriteRenderState`, `PersistSubmission` and `CompleteTask`, each one atomic conditional write whose 0-rows result means "stale, dropped". `WriteDraft` (from `SaveDraft`) is guarded the same way and writes only the active step's data.
 - **taskflow stores claims; `CompleteTaskStep` does not check them.** `ClaimedBy` and `ClaimedAt` change only via `ClaimTask` and `ReleaseTask`, never via the step writes, so a claim lasts for the whole task. For this MVP a claim is UI guidance so two people can avoid working the same task (see [`docs/integration-guide.md`](docs/integration-guide.md#7-task-claims)).
 - **Plugins suspend with `plugins.ErrSuspended`.** Synchronous plugins return `nil`; the workflow advances without waiting.
-- **`StartTask` returns `activity.ErrResultPending`** on the happy path. The parent activity suspends until `onTaskCompleted` fires.
+- **`StartTask` returns `activity.ErrResultPending`** on the happy path. The parent's activity ends, and its node waits in the workflow until `onTaskCompleted` completes it.
 - **Submission payloads are scoped** to the active step's `OutputNamespace` in `TaskRecord.Data`. Callers send a raw object; the server stamps the correct key.
 
 ## Task-level workflow constraint: no parallel paths

taskflow/docs/integration-guide.md

@@ -169,16 +169,16 @@ taskWorkflowManager := engine.NewTemporalManager(
 The handlers wire into the orchestrator:
 
 ```go
-parentTaskHandler := func(p engine.TaskPayload) (map[string]any, error) {
-    return tm.StartTask(p)
+parentTaskHandler := func(ctx context.Context, p engine.TaskPayload) (map[string]any, error) {
+    return tm.StartTask(ctx, p)
 }
 
-taskHandler := func(p engine.TaskPayload) (map[string]any, error) {
-    return tm.StartTaskStep(p)
+taskHandler := func(ctx context.Context, p engine.TaskPayload) (map[string]any, error) {
+    return tm.StartTaskStep(ctx, p)
 }
 
-taskCompletionHandler := func(c engine.WorkflowCompletion) error {
-    return tm.HandleTaskCompletion(context.Background(), c)
+taskCompletionHandler := func(ctx context.Context, c engine.WorkflowCompletion) error {
+    return tm.HandleTaskCompletion(ctx, c)
 }
 ```
 
@@ -191,11 +191,11 @@ taskCompletionHandler := func(c engine.WorkflowCompletion) error {
 ```go
 var tm *orchestrator.TaskManager
 
-taskHandler := func(p engine.TaskPayload) (map[string]any, error) {
+taskHandler := func(ctx context.Context, p engine.TaskPayload) (map[string]any, error) {
     if tm == nil {
         return nil, fmt.Errorf("task manager not initialised")
     }
-    return tm.StartTaskStep(p)
+    return tm.StartTaskStep(ctx, p)
 }
 
 // ... construct managers ...
@@ -289,7 +289,7 @@ The orchestrator uses two sentinel errors to signal "this isn't really an error,
 
 | Sentinel                                                         | Returned by                                        | Meaning                                                                                  |
 |------------------------------------------------------------------|----------------------------------------------------|------------------------------------------------------------------------------------------|
-| `activity.ErrResultPending` (from `go.temporal.io/sdk/activity`) | `StartTask`, `StartTaskStep` (when plugin suspends) | The Temporal activity should park indefinitely; the orchestrator will resume it later.   |
+| `activity.ErrResultPending` (from `go.temporal.io/sdk/activity`) | `StartTask`, `StartTaskStep` (when plugin suspends) | The step waits for a completion: its activity ends, and its node waits in the workflow, with no timeout, until `CompleteActivation` resumes it. |
 | `plugins.ErrSuspended`                                           | A plugin's `Execute` method                        | "I dispatched the work; don't advance the workflow until something external resumes me." |
 
 `StartTaskStep` translates `ErrSuspended` from a plugin into `ErrResultPending` for Temporal. Both are normal happy-path values, not failures.
@@ -301,18 +301,18 @@ A real error from `StartTask` / `StartTaskStep` (template not found, plugin fail
 ## What `StartTask` does, step by step
 
 ```go
-func (tm *TaskManager) StartTask(payload engine.TaskPayload) (map[string]any, error) {
+func (tm *TaskManager) StartTask(ctx context.Context, payload engine.TaskPayload) (map[string]any, error) {
     // 1. Resolve TaskTemplate → workflow definition → render config
     // 2. Reject if the child workflow has parallel gateways
     // 3. taskID := payload.ActivationID (the parent's step ID); taskWorkflowID := "task-wf-" + taskID
     // 4. Build the initial Data map from payload.Inputs (setNestedKey for dotted keys)
     // 5. Persist TaskRecord (state=STARTING, parent coords, render config snapshot)
     // 6. Start the child task workflow on the task queue
-    // 7. Return activity.ErrResultPending — the parent activity parks
+    // 7. Return activity.ErrResultPending — the parent's node waits for the task to complete
 }
 ```
 
-Note that **the parent activity returns `ErrResultPending` even on success**. The parent workflow doesn't get its result until `onTaskCompleted` wakes it later.
+Note that **`StartTask` returns `ErrResultPending` even on success**. The parent's activity ends there, and the parent workflow doesn't get the task's result until `onTaskCompleted` completes its node later.
 
 ## What `StartTaskStep` does
 

workflow/activation_identity_test.go

@@ -44,12 +44,12 @@ func runRecordingSteps(t *testing.T, defJSON string, result func(template string
 		seen []stepSeen
 	)
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
-		Return(func(ctx context.Context, template string, _ map[string]any, _ string, step ActivationRef) (map[string]any, error) {
+		Return(taskOutput(func(ctx context.Context, template string, _ map[string]any, _ string, step ActivationRef) (map[string]any, error) {
 			mu.Lock()
 			defer mu.Unlock()
 			seen = append(seen, stepSeen{activityID: activity.GetInfo(ctx).ActivityID, template: template, step: step})
 			return result(template, len(seen)), nil
-		})
+		}))
 	var completion WorkflowCompletion
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(func(_ context.Context, workflowID string, vars map[string]any, seq int64) error {

workflow/activities.go

@@ -6,6 +6,7 @@ package engine
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"go.temporal.io/sdk/activity"
@@ -14,16 +15,14 @@ import (
 // Activities encapsulates the Temporal activity implementations utilized by the workflow engine.
 // It maps the activity execution flow to custom callback handlers provided by the host application.
 type Activities struct {
-	// ExecuteTaskActivityHandler is invoked when the workflow engine encounters a task node.
-	// - For synchronous execution, it should return a nil error with a map containing the results.
-	// - For asynchronous execution, it should return a nil map and an ErrResultPending error,
-	//   which pauses the workflow activity until an external handler triggers CompleteActivation.
-	ExecuteTaskActivityHandler func(TaskPayload) (map[string]any, error)
+	// ExecuteTaskActivityHandler is invoked when the workflow engine encounters a task node. See
+	// TaskActivationHandler.
+	ExecuteTaskActivityHandler TaskActivationHandler
 
 	// WorkflowCompletedActivityHandler is invoked when the overall workflow execution succeeds and reaches
 	// an End node. It receives the workflow ID and the final accumulated workflow variables, allowing the
 	// host application to run any necessary completion triggers, notify listeners, or persist final state.
-	WorkflowCompletedActivityHandler func(WorkflowCompletion) error
+	WorkflowCompletedActivityHandler WorkflowCompletionHandler
 
 	// FetchWorkflowDefinitionHandler is invoked to dynamically retrieve the workflow definition structure
 	// for a given template ID during SPLIT_TASK execution.
@@ -55,8 +54,20 @@ type ActivationRef struct {
 	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) {
+// TaskResult is what ExecuteTaskActivity reports to the workflow.
+type TaskResult struct {
+	// Suspended is true when the step is waiting for a completion (the handler returned
+	// ErrResultPending). The workflow then waits for CompleteActivation, outside the Activity.
+	Suspended bool `json:"suspended,omitempty"`
+	// Output is the step's result when it completed synchronously.
+	Output map[string]any `json:"output,omitempty"`
+}
+
+// ExecuteTaskActivity hands the task to the host application. It returns the step's output when the
+// handler completes it synchronously, and Suspended when the handler returns ErrResultPending: the
+// Activity ends either way, and a suspended step's completion reaches the workflow through
+// CompleteActivation, not through this Activity.
+func (a *Activities) ExecuteTaskActivity(ctx context.Context, taskTemplateID string, inputs map[string]any, caseID string, step ActivationRef) (TaskResult, error) {
 	info := activity.GetInfo(ctx)
 	payload := TaskPayload{
 		WorkflowID:      info.WorkflowExecution.ID,
@@ -70,20 +81,19 @@ func (a *Activities) ExecuteTaskActivity(ctx context.Context, taskTemplateID str
 		CaseID:          caseID,
 	}
 
-	// Trigger custom code block. ExecuteTaskActivityHandler can return error ErrResultPending to pause the workflow
-	// or return a nil error with the outputs for the next step to consume (synchronous execution)
-	res, err := a.ExecuteTaskActivityHandler(payload)
+	res, err := a.ExecuteTaskActivityHandler(ctx, payload)
+	if errors.Is(err, activity.ErrResultPending) {
+		return TaskResult{Suspended: true}, nil
+	}
 	if err != nil {
-		return nil, err
+		return TaskResult{}, err
 	}
-
-	// Return result immediately for synchronous steps
-	return res, nil
+	return TaskResult{Output: res}, nil
 }
 
 // WorkflowCompletedActivity is a Temporal activity that executes when a workflow completes successfully.
-func (a *Activities) WorkflowCompletedActivity(_ context.Context, workflowID string, finalContext map[string]any, seq int64) error {
-	return a.WorkflowCompletedActivityHandler(WorkflowCompletion{WorkflowID: workflowID, Seq: seq, FinalVariables: finalContext})
+func (a *Activities) WorkflowCompletedActivity(ctx context.Context, workflowID string, finalContext map[string]any, seq int64) error {
+	return a.WorkflowCompletedActivityHandler(ctx, WorkflowCompletion{WorkflowID: workflowID, Seq: seq, FinalVariables: finalContext})
 }
 
 // AdminParkActivity is a Temporal activity that notifies the host application whenever a node

workflow/admin_recovery_test.go

@@ -296,7 +296,7 @@ func TestAdminRetryResolvesInputMappingError(t *testing.T) {
 	env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"})
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_WITH_MISSING_INPUT", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil).Once()
+		Return(TaskResult{Output: map[string]any{}}, nil).Once()
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
 
 	env.RegisterDelayedCallback(func() {
@@ -339,7 +339,7 @@ func TestAdminCompleteResolvesOutputMappingErrorWithoutReinvokingActivity(t *tes
 	env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"})
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_MISSING_REQUIRED_OUTPUT", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil).Once()
+		Return(TaskResult{Output: map[string]any{}}, nil).Once()
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
 
 	// Confirm the parked LastError explicitly warns that the Activity already ran, before
@@ -401,9 +401,9 @@ func TestAdminRetryRefreshesCachedTaskResultWithoutStaleData(t *testing.T) {
 	env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"})
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_MISSING_REQUIRED_OUTPUT", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"attempt": "first"}, nil).Once()
+		Return(TaskResult{Output: map[string]any{"attempt": "first"}}, nil).Once()
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_MISSING_REQUIRED_OUTPUT", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"attempt": "second"}, nil).Once()
+		Return(TaskResult{Output: map[string]any{"attempt": "second"}}, nil).Once()
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
 
 	// Retry without fixing anything — the Activity runs again, output mapping fails again.
@@ -645,7 +645,7 @@ func TestAdminCompleteRejectedForParkedGatewayNode(t *testing.T) {
 	env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"})
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_PASS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil).Once()
+		Return(TaskResult{Output: map[string]any{}}, nil).Once()
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
 
 	// Rejecting a GATEWAY's Complete just continues parkNodeForAdmin's loop without re-minting, so
@@ -728,9 +728,9 @@ func TestAdminParkingIsolatesParallelBranches(t *testing.T) {
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_A", mock.Anything, mock.Anything, mock.Anything).
-		Return(nil, temporal.NewNonRetryableApplicationError("boom", "TaskFailure", nil)).Once()
+		Return(TaskResult{}, temporal.NewNonRetryableApplicationError("boom", "TaskFailure", nil)).Once()
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_B", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil).Once()
+		Return(TaskResult{Output: map[string]any{}}, nil).Once()
 
 	branchWorkflowID := FormatChildWorkflowID("default-test-workflow-id", "default-test-workflow-id", "split", "e2")