diff --git a/README.md b/README.md
index 649f956..8ccb19d 100644
--- a/README.md
+++ b/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)
diff --git a/taskflow/README.md b/taskflow/README.md
index 952b272..9af3bbf 100644
--- a/taskflow/README.md
+++ b/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
diff --git a/taskflow/docs/integration-guide.md b/taskflow/docs/integration-guide.md
index 5f4d5f0..dccd4b4 100644
--- a/taskflow/docs/integration-guide.md
+++ b/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
 
diff --git a/workflow/activation_identity_test.go b/workflow/activation_identity_test.go
index edc1746..681c45d 100644
--- a/workflow/activation_identity_test.go
+++ b/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 {
diff --git a/workflow/activities.go b/workflow/activities.go
index 17dd830..33f5ee0 100644
--- a/workflow/activities.go
+++ b/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
diff --git a/workflow/admin_recovery_test.go b/workflow/admin_recovery_test.go
index 0247bc8..2a54e73 100644
--- a/workflow/admin_recovery_test.go
+++ b/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")
 
diff --git a/workflow/batch_gateway_test.go b/workflow/batch_gateway_test.go
index 2d55a68..40616f1 100644
--- a/workflow/batch_gateway_test.go
+++ b/workflow/batch_gateway_test.go
@@ -79,25 +79,25 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_Basic_TwoPartitions() {
 
 	// load_items returns 3 items.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"items": []any{
 				map[string]any{"commodity_id": "C1", "type": "food", "description": "Dried fish"},
 				map[string]any{"commodity_id": "C2", "type": "goods", "description": "Textiles"},
 				map[string]any{"commodity_id": "C3", "type": "food", "description": "Spices"},
 			},
-		}, nil).Once()
+		}}, nil).Once()
 
 	// food_inspect: called in the "food" child workflow, processes 2 items.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "FOOD_INSPECTION", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	// goods_inspect: called in the "goods" child workflow, processes 1 item.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "GOODS_INSPECTION", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	// issue_cert: called in the parent after join, sees all 3 items.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "ISSUE_CERTIFICATE", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "batch-test-1"})
@@ -164,7 +164,7 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_SingleItem() {
 		})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "single-item-1"})
@@ -461,9 +461,9 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_DefaultEdgeCatchAll() {
 		})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "FOOD_PROCESS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "OTHER_PROCESS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "default-edge-1"})
@@ -524,7 +524,7 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_EmptyItems_SkipsToJoin() {
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "FINAL", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "empty-items-1"})
@@ -917,7 +917,7 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_Depth2_PhytoConsignment() {
 
 	// load_consignment: 5 items — 3 visual, 2 lab.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_CONSIGNMENT", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"items": []any{
 				map[string]any{"id": "item-1", "requiresLabTest": false, "type": "visual"},
 				map[string]any{"id": "item-2", "requiresLabTest": true, "type": "lab"},
@@ -925,36 +925,36 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_Depth2_PhytoConsignment() {
 				map[string]any{"id": "item-4", "requiresLabTest": true, "type": "lab"},
 				map[string]any{"id": "item-5", "requiresLabTest": false, "type": "visual"},
 			},
-		}, nil).Once()
+		}}, nil).Once()
 
 	// visual_inspection: runs for the 3 visual items.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_INSPECTION", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	// lab_sampling: simulates a task that tests each item and writes labResult back.
 	// The activity returns "tested_items" which output_mapping writes to "_items".
 	// item-2 fails, item-4 passes.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_SAMPLING", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"tested_items": []any{
 				map[string]any{"id": "item-2", "requiresLabTest": true, "type": "lab", "labResult": "fail"},
 				map[string]any{"id": "item-4", "requiresLabTest": true, "type": "lab", "labResult": "pass"},
 			},
-		}, nil)
+		}}, nil)
 
 	// pass_through and supervisor_review at depth-2.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "PASS_THROUGH", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "SUPERVISOR_REVIEW", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"reviewed_items": []any{
 				map[string]any{"id": "item-2", "requiresLabTest": true, "type": "lab", "labResult": "fail", "supervisorVerdict": "approved_with_conditions"},
 			},
-		}, nil)
+		}}, nil)
 
 	// issue_certificate: final step after all items merge.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "ISSUE_CERTIFICATE", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "phyto-1"})
@@ -1046,9 +1046,9 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_EdgeConditionSeesWorkflowVars() {
 		})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "HEAVY", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LIGHT", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "wf-vars-1"})
@@ -1118,9 +1118,9 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ReplayDeterminism() {
 		})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS_A", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS_B", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "replay-batch-1"})
@@ -1152,8 +1152,8 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ReplayDeterminism() {
 			}
 			return nil
 		})
-	env2.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS_A", mock.Anything, mock.Anything, mock.Anything).Return(map[string]any{}, nil)
-	env2.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS_B", mock.Anything, mock.Anything, mock.Anything).Return(map[string]any{}, nil)
+	env2.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS_A", mock.Anything, mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{}}, nil)
+	env2.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS_B", mock.Anything, mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{}}, nil)
 	env2.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env2.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "replay-batch-2"})
 
@@ -1275,7 +1275,7 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ChildReturnsInvalidItemsType_Fail
 
 	// CORRUPT_TASK outputs a string instead of a slice for _items
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "CORRUPT_TASK", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"bad_items": "not-a-slice"}, nil)
+		Return(TaskResult{Output: map[string]any{"bad_items": "not-a-slice"}}, nil)
 
 	env.RegisterDelayedCallback(func() {
 		env.SignalWorkflow("AdminResolutionSignal", AdminResolutionSignal{
@@ -1343,7 +1343,7 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ChildReturnsItemMissingID_Fails()
 
 	// STRIP_ID_TASK outputs an item without an id field
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "STRIP_ID_TASK", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"items": []any{map[string]any{"name": "foo"}}}, nil)
+		Return(TaskResult{Output: map[string]any{"items": []any{map[string]any{"name": "foo"}}}}, nil)
 
 	env.RegisterDelayedCallback(func() {
 		env.SignalWorkflow("AdminResolutionSignal", AdminResolutionSignal{
@@ -1411,7 +1411,7 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ChildReturnsNonStringID_Fails() {
 
 	// NUMERIC_ID_TASK rewrites the item with a numeric id.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "NUMERIC_ID_TASK", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"items": []any{map[string]any{"id": 1000000, "name": "foo"}}}, nil)
+		Return(TaskResult{Output: map[string]any{"items": []any{map[string]any{"id": 1000000, "name": "foo"}}}}, nil)
 
 	env.RegisterDelayedCallback(func() {
 		env.SignalWorkflow("AdminResolutionSignal", AdminResolutionSignal{
@@ -1481,11 +1481,11 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ChildReturnsDuplicateItemIDAcross
 		})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_A", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	// TASK_B returns an item with id "item-a", which already belongs to partition A
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_B", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"items": []any{map[string]any{"id": "item-a", "type": "b"}}}, nil)
+		Return(TaskResult{Output: map[string]any{"items": []any{map[string]any{"id": "item-a", "type": "b"}}}}, nil)
 
 	env.RegisterDelayedCallback(func() {
 		env.SignalWorkflow("AdminResolutionSignal", AdminResolutionSignal{
@@ -1546,7 +1546,7 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ChildTaskAdminAbort_PropagatesWit
 	}
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS", mock.Anything, mock.Anything, mock.Anything).
-		Return(nil, temporal.NewNonRetryableApplicationError("inspection boom", "TaskFailure", nil)).Once()
+		Return(TaskResult{}, temporal.NewNonRetryableApplicationError("inspection boom", "TaskFailure", nil)).Once()
 
 	parentWorkflowID := "batch-child-abort-1"
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
diff --git a/workflow/batch_mapping_test.go b/workflow/batch_mapping_test.go
index d393663..3b5cb35 100644
--- a/workflow/batch_mapping_test.go
+++ b/workflow/batch_mapping_test.go
@@ -41,7 +41,7 @@ func (s *BatchGatewayTestSuite) newBatchMappingTestEnv() *testsuite.TestWorkflow
 
 func (s *BatchGatewayTestSuite) onTask(env *testsuite.TestWorkflowEnvironment, templateID string, output map[string]any) {
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, templateID, mock.Anything, mock.Anything, mock.Anything).
-		Return(output, nil)
+		Return(TaskResult{Output: output}, nil)
 }
 
 func (s *BatchGatewayTestSuite) runBatchMappingWorkflow(env *testsuite.TestWorkflowEnvironment, def WorkflowDefinition, vars map[string]any) map[string]any {
@@ -209,13 +209,13 @@ func (s *BatchGatewayTestSuite) recordTaskInputs(env *testsuite.TestWorkflowEnvi
 	var mu sync.Mutex
 	var seen []map[string]any
 	for _, tmpl := range []string{"FOOD_TASK", "GOODS_TASK"} {
-		env.OnActivity("ExecuteTaskActivity", mock.Anything, tmpl, mock.Anything, mock.Anything, mock.Anything).Return(
+		env.OnActivity("ExecuteTaskActivity", mock.Anything, tmpl, mock.Anything, mock.Anything, mock.Anything).Return(taskOutput(
 			func(_ context.Context, _ string, inputs map[string]any, _ string, _ ActivationRef) (map[string]any, error) {
 				mu.Lock()
 				defer mu.Unlock()
 				seen = append(seen, inputs)
 				return map[string]any{}, nil
-			})
+			}))
 	}
 	return func() []map[string]any { mu.Lock(); defer mu.Unlock(); return seen }
 }
diff --git a/workflow/complete_activation_integration_test.go b/workflow/complete_activation_integration_test.go
index ee98d32..9a51b7b 100644
--- a/workflow/complete_activation_integration_test.go
+++ b/workflow/complete_activation_integration_test.go
@@ -57,13 +57,13 @@ func startPendingLoop(t *testing.T) (TemporalManager, string, <-chan TaskPayload
 	runs := make(chan TaskPayload, 8)
 	var mu sync.Mutex
 	mgr := NewTemporalManager(c, "default", "task-done-it-"+uuid.NewString(),
-		func(p TaskPayload) (map[string]any, error) {
+		func(_ context.Context, p TaskPayload) (map[string]any, error) {
 			mu.Lock()
 			defer mu.Unlock()
 			runs <- p
 			return nil, activity.ErrResultPending
 		},
-		func(WorkflowCompletion) error { return nil })
+		func(context.Context, WorkflowCompletion) error { return nil })
 	require.NoError(t, mgr.StartWorker())
 	t.Cleanup(mgr.StopWorker)
 
@@ -92,13 +92,12 @@ func TestCompleteStepRejectsAnEarlierRunOfALoopingNode(t *testing.T) {
 	ctx := context.Background()
 
 	first := nextRun(t, runs)
-	require.Equal(t, int64(1), first.Seq)
 	require.NoError(t, mgr.CompleteActivation(ctx, workflowID, "", first.ActivationID, map[string]any{"delivered": false}))
 
 	second := nextRun(t, runs) // the loop came back to the same node
 	require.Equal(t, first.NodeID, second.NodeID, "same node in the definition")
 	require.NotEqual(t, first.ActivationID, second.ActivationID, "but a new step")
-	require.Equal(t, int64(2), second.Seq)
+	require.Greater(t, second.Seq, first.Seq, "a later run of the node has a higher Seq")
 
 	// A late or duplicate completion for the first pass is rejected and must not touch the second.
 	err := mgr.CompleteActivation(ctx, workflowID, "", first.ActivationID, map[string]any{"delivered": true})
@@ -136,3 +135,33 @@ func TestCompleteStepAfterTheWorkflowEndedIsNotPending(t *testing.T) {
 	require.Error(t, err)
 	require.True(t, errors.Is(err, ErrActivationNotPending), "got %v", err)
 }
+
+// Of two completions for the same step sent at once, exactly one is accepted and the other is
+// ErrActivationNotPending. CompleteTaskStep relies on this to let only one submission write.
+func TestCompleteStepConcurrentCompletionsOneWins(t *testing.T) {
+	mgr, workflowID, runs := startPendingLoop(t)
+	first := nextRun(t, runs)
+
+	const callers = 5
+	errs := make(chan error, callers)
+	var start sync.WaitGroup
+	start.Add(1)
+	for range callers {
+		go func() {
+			start.Wait()
+			errs <- mgr.CompleteActivation(context.Background(), workflowID, "", first.ActivationID, map[string]any{"delivered": true})
+		}()
+	}
+	start.Done()
+
+	accepted := 0
+	for range callers {
+		err := <-errs
+		if err == nil {
+			accepted++
+			continue
+		}
+		require.ErrorIs(t, err, ErrActivationNotPending)
+	}
+	require.Equal(t, 1, accepted)
+}
diff --git a/workflow/completion.go b/workflow/completion.go
new file mode 100644
index 0000000..9cf0a94
--- /dev/null
+++ b/workflow/completion.go
@@ -0,0 +1,69 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) 2026 Lanka Software Foundation
+
+package engine
+
+import (
+	"fmt"
+
+	"go.temporal.io/sdk/temporal"
+	"go.temporal.io/sdk/workflow"
+)
+
+// completeActivationUpdate is the Workflow Update that delivers a suspended step's output to the
+// node waiting on it. Manager.CompleteActivation sends it.
+const completeActivationUpdate = "CompleteActivation"
+
+// activationNotPendingErrorType is the type of the error the workflow rejects a completion with when
+// the step it names isn't waiting. Manager.CompleteActivation turns it into ErrActivationNotPending.
+const activationNotPendingErrorType = "ActivationNotPending"
+
+// A step whose handler returns ErrResultPending ends its Activity at once, reporting Suspended, and
+// its node waits here, in the workflow, for CompleteActivation. The wait is not an Activity, so it
+// has no timeout: a step can wait for a person for as long as it takes, and the Activity's own
+// timeout only has to cover running the plugin.
+//
+// pendingCompletions holds a Settable for each step waiting to be completed, keyed by its
+// ActivationID. A node registers its step before it starts the Activity, so a completion that
+// arrives while the Activity is still finishing is accepted, and waits for the node to read it.
+
+// registerCompletionHandler installs the CompleteActivation Update handler. Its validator rejects a
+// completion for a step that isn't waiting, and its handler takes the step out of pendingCompletions
+// before anything else runs, so of several completions for one step exactly one is accepted and the
+// rest are rejected: the same guarantee Temporal gave when completing an Activity.
+func (g *graphInterpreter) registerCompletionHandler(ctx workflow.Context) error {
+	return workflow.SetUpdateHandlerWithOptions(ctx, completeActivationUpdate,
+		func(_ workflow.Context, activationID string, output map[string]any) error {
+			settable, ok := g.pendingCompletions[activationID]
+			if !ok {
+				// The validator ran against the same state, with nothing in between.
+				return activationNotPending(activationID)
+			}
+			delete(g.pendingCompletions, activationID)
+			settable.Set(output, nil)
+			return nil
+		},
+		workflow.UpdateHandlerOptions{
+			Validator: func(_ workflow.Context, activationID string, _ map[string]any) error {
+				if _, ok := g.pendingCompletions[activationID]; !ok {
+					return activationNotPending(activationID)
+				}
+				return nil
+			},
+		},
+	)
+}
+
+func activationNotPending(activationID string) error {
+	return temporal.NewApplicationError(
+		fmt.Sprintf("step %s is not waiting to be completed", activationID), activationNotPendingErrorType)
+}
+
+// expectCompletion registers activationID as a step that may be completed, and returns the Future its
+// completion arrives on and a function that unregisters it. Call the function on every way out of the
+// node: a step that completed synchronously, failed or was cancelled must stop accepting completions.
+func (g *graphInterpreter) expectCompletion(ctx workflow.Context, activationID string) (workflow.Future, func()) {
+	future, settable := workflow.NewFuture(ctx)
+	g.pendingCompletions[activationID] = settable
+	return future, func() { delete(g.pendingCompletions, activationID) }
+}
diff --git a/workflow/completion_test.go b/workflow/completion_test.go
new file mode 100644
index 0000000..964631d
--- /dev/null
+++ b/workflow/completion_test.go
@@ -0,0 +1,224 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) 2026 Lanka Software Foundation
+
+package engine
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"go.temporal.io/sdk/activity"
+	"go.temporal.io/sdk/testsuite"
+)
+
+// suspendingWorkflowJSON is start -> form -> end, where form's output maps to the variable decision.
+const suspendingWorkflowJSON = `
+{
+  "workflow_id": "suspending",
+  "name": "Suspending",
+  "version": 1,
+  "edges":[
+    { "id": "e1", "source_id": "start", "target_id": "form" },
+    { "id": "e2", "source_id": "form", "target_id": "end" }
+  ],
+  "nodes":[
+    { "id": "start", "type": "START" },
+    { "id": "form", "type": "TASK", "task_template_id": "FORM", "output_mapping": { "decision": "decision" } },
+    { "id": "end", "type": "END" }
+  ]
+}`
+
+// newSuspendingEnv returns a test environment running suspendingWorkflowJSON, whose FORM step
+// suspends (its handler returned ErrResultPending), and a pointer to how often the step's Activity ran.
+func newSuspendingEnv(t *testing.T) (*testsuite.TestWorkflowEnvironment, WorkflowDefinition, *int) {
+	t.Helper()
+	env := (&testsuite.WorkflowTestSuite{}).NewTestWorkflowEnvironment()
+	var def WorkflowDefinition
+	require.NoError(t, json.Unmarshal([]byte(suspendingWorkflowJSON), &def))
+
+	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"})
+	runs := 0
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "FORM", mock.Anything, mock.Anything, mock.Anything).
+		Return(func(context.Context, string, map[string]any, string, ActivationRef) (TaskResult, error) {
+			runs++
+			return TaskResult{Suspended: true}, nil
+		})
+	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
+	return env, def, &runs
+}
+
+// activationOf returns the ActivationID of nodeID's current run.
+func activationOf(t *testing.T, env *testsuite.TestWorkflowEnvironment, nodeID string) string {
+	t.Helper()
+	val, err := env.QueryWorkflow("GetStatus")
+	require.NoError(t, err)
+	var instance WorkflowInstance
+	require.NoError(t, val.Get(&instance))
+	return instance.NodeInfo[nodeID].ActivationID
+}
+
+// updateOutcome is what became of one CompleteActivation Update.
+type updateOutcome struct {
+	accepted bool
+	rejected error
+}
+
+// complete sends the CompleteActivation Update. The workflow decides on it after the calling callback
+// returns, so read the outcome once the workflow has run.
+func complete(env *testsuite.TestWorkflowEnvironment, updateID, activationID string, output map[string]any) *updateOutcome {
+	outcome := &updateOutcome{}
+	env.UpdateWorkflow(completeActivationUpdate, updateID, &testsuite.TestUpdateCallback{
+		OnAccept:   func() { outcome.accepted = true },
+		OnReject:   func(err error) { outcome.rejected = err },
+		OnComplete: func(any, error) {},
+	}, activationID, output)
+	return outcome
+}
+
+func requireAccepted(t *testing.T, o *updateOutcome) {
+	t.Helper()
+	require.NoError(t, o.rejected)
+	require.True(t, o.accepted)
+}
+
+func requireRejected(t *testing.T, o *updateOutcome) {
+	t.Helper()
+	require.False(t, o.accepted)
+	require.ErrorContains(t, o.rejected, "is not waiting to be completed")
+}
+
+func finalVariables(t *testing.T, env *testsuite.TestWorkflowEnvironment) map[string]any {
+	t.Helper()
+	require.True(t, env.IsWorkflowCompleted())
+	require.NoError(t, env.GetWorkflowError())
+	var instance WorkflowInstance
+	require.NoError(t, env.GetWorkflowResult(&instance))
+	return instance.WorkflowVariables
+}
+
+// A suspended step waits outside its Activity, and its completion carries the step's output.
+func TestSuspendedStepCompletesThroughTheUpdate(t *testing.T) {
+	env, def, runs := newSuspendingEnv(t)
+
+	var done *updateOutcome
+	env.RegisterDelayedCallback(func() {
+		done = complete(env, "u1", activationOf(t, env, "form"), map[string]any{"decision": "approved"})
+	}, time.Hour)
+
+	env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{})
+
+	requireAccepted(t, done)
+	require.Equal(t, "approved", finalVariables(t, env)["decision"])
+	require.Equal(t, 1, *runs)
+}
+
+// A step can wait far longer than its Activity's timeout, and its plugin is not run again: the wait
+// is not an Activity.
+func TestSuspendedStepWaitsPastTheActivityTimeout(t *testing.T) {
+	env, def, runs := newSuspendingEnv(t)
+
+	var done *updateOutcome
+	env.RegisterDelayedCallback(func() {
+		done = complete(env, "u1", activationOf(t, env, "form"), map[string]any{"decision": "approved"})
+	}, 400*24*time.Hour)
+
+	env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{})
+
+	requireAccepted(t, done)
+	require.Equal(t, "approved", finalVariables(t, env)["decision"])
+	require.Equal(t, 1, *runs, "the step's Activity ran once")
+}
+
+// A completion for a step that isn't waiting is rejected and changes nothing; of two completions for
+// the same step, the first is accepted and the second rejected.
+func TestCompletionForAStepNotWaitingIsRejected(t *testing.T) {
+	env, def, _ := newSuspendingEnv(t)
+
+	var unknown, first, duplicate *updateOutcome
+	env.RegisterDelayedCallback(func() {
+		unknown = complete(env, "u1", "no-such-step", map[string]any{"decision": "forged"})
+		step := activationOf(t, env, "form")
+		first = complete(env, "u2", step, map[string]any{"decision": "approved"})
+		duplicate = complete(env, "u3", step, map[string]any{"decision": "again"})
+	}, time.Hour)
+
+	env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{})
+
+	requireRejected(t, unknown)
+	requireAccepted(t, first)
+	requireRejected(t, duplicate)
+	require.Equal(t, "approved", finalVariables(t, env)["decision"])
+}
+
+// A host can complete a step it suspended before the step's Activity has reported back: the
+// completion is accepted and the node takes it once the Activity returns.
+func TestCompletionBeforeTheActivityReturnsIsKept(t *testing.T) {
+	env := (&testsuite.WorkflowTestSuite{}).NewTestWorkflowEnvironment()
+	var def WorkflowDefinition
+	require.NoError(t, json.Unmarshal([]byte(suspendingWorkflowJSON), &def))
+	acts := &Activities{}
+	env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"})
+	env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"})
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "FORM", mock.Anything, mock.Anything, mock.Anything).
+		After(time.Minute).Return(TaskResult{Suspended: true}, nil)
+	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+	var early *updateOutcome
+	env.RegisterDelayedCallback(func() {
+		early = complete(env, "u1", activationOf(t, env, "form"), map[string]any{"decision": "early"})
+	}, time.Second)
+
+	env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{})
+
+	requireAccepted(t, early)
+	require.Equal(t, "early", finalVariables(t, env)["decision"])
+}
+
+// A step that completed synchronously takes no completion once the workflow has moved on.
+func TestCompletionForASynchronousStepIsRejected(t *testing.T) {
+	const twoStepJSON = `
+	{
+	  "workflow_id": "two-step", "name": "Two step", "version": 1,
+	  "edges":[
+	    { "id": "e1", "source_id": "start", "target_id": "calc" },
+	    { "id": "e2", "source_id": "calc", "target_id": "form" },
+	    { "id": "e3", "source_id": "form", "target_id": "end" }
+	  ],
+	  "nodes":[
+	    { "id": "start", "type": "START" },
+	    { "id": "calc", "type": "TASK", "task_template_id": "CALC" },
+	    { "id": "form", "type": "TASK", "task_template_id": "FORM", "output_mapping": { "decision": "decision" } },
+	    { "id": "end", "type": "END" }
+	  ]
+	}`
+	env := (&testsuite.WorkflowTestSuite{}).NewTestWorkflowEnvironment()
+	var def WorkflowDefinition
+	require.NoError(t, json.Unmarshal([]byte(twoStepJSON), &def))
+	acts := &Activities{}
+	env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"})
+	env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"})
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "CALC", mock.Anything, mock.Anything, mock.Anything).
+		Return(TaskResult{Output: map[string]any{}}, nil)
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "FORM", mock.Anything, mock.Anything, mock.Anything).
+		Return(TaskResult{Suspended: true}, nil)
+	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+	var stale, current *updateOutcome
+	env.RegisterDelayedCallback(func() {
+		stale = complete(env, "u1", activationOf(t, env, "calc"), map[string]any{"decision": "forged"})
+		current = complete(env, "u2", activationOf(t, env, "form"), map[string]any{"decision": "approved"})
+	}, time.Hour)
+
+	env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{})
+
+	requireRejected(t, stale)
+	requireAccepted(t, current)
+	require.Equal(t, "approved", finalVariables(t, env)["decision"])
+}
diff --git a/workflow/dynamic_split_test.go b/workflow/dynamic_split_test.go
index f7d6517..f1291e9 100644
--- a/workflow/dynamic_split_test.go
+++ b/workflow/dynamic_split_test.go
@@ -109,13 +109,13 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithDifferentTemplates() {
 	// 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{
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_phyto_inspection", mock.Anything, mock.Anything, pinned).Return(TaskResult{Output: map[string]any{
 		"inspection_status": "APPROVED_PHYTO",
-	}, nil)
+	}}, nil)
 
-	env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_health_inspection", mock.Anything, mock.Anything, pinned).Return(map[string]any{
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_health_inspection", mock.Anything, mock.Anything, pinned).Return(TaskResult{Output: map[string]any{
 		"inspection_status": "APPROVED_HEALTH",
-	}, nil)
+	}}, nil)
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
@@ -237,9 +237,9 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithSameTemplateMode() {
 	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{
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "process_item", mock.Anything, mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{
 		"processed_status": "DONE_SUCCESS",
-	}, nil)
+	}}, nil)
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
@@ -337,7 +337,7 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithCollectAllFailures() {
 	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"))
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "fail_task", mock.Anything, mock.Anything, mock.Anything).Return(TaskResult{}, errors.New("task failed intentionally"))
 
 	// Register nested sub-workflow runtime interpreter engine
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
@@ -523,13 +523,13 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithCrossBranchBroadcast() {
 	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{
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "run_phyto_inspection", mock.Anything, mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{
 		"inspection_status": "APPROVED_CLEAN",
-	}, nil)
+	}}, nil)
 
-	env.OnActivity("ExecuteTaskActivity", mock.Anything, "verify_cross_border_docs", mock.Anything, mock.Anything, mock.Anything).Return(map[string]any{
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "verify_cross_border_docs", mock.Anything, mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{
 		"health_clearance": "PASSED_SECURE",
-	}, nil)
+	}}, nil)
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
diff --git a/workflow/engine_test.go b/workflow/engine_test.go
index c6d6c2d..8ddc928 100644
--- a/workflow/engine_test.go
+++ b/workflow/engine_test.go
@@ -225,25 +225,25 @@ func TestCustomsExportLCLFlow(t *testing.T) {
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "SUBMIT_CUSDEC", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"consignment_type": "LCL"}, nil).Once()
+		Return(TaskResult{Output: map[string]any{"consignment_type": "LCL"}}, nil).Once()
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "PAY_DUTIES", mock.Anything, mock.Anything, mock.Anything).
-		Return(emptyMap, nil).Once()
+		Return(TaskResult{Output: emptyMap}, nil).Once()
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "CREATE_LCL_CDN", mock.Anything, mock.Anything, mock.Anything).
-		Return(emptyMap, nil).Once()
+		Return(TaskResult{Output: emptyMap}, nil).Once()
 
 	// CREATE_FCL_CDN should NEVER be called since the LCL path was evaluated.
 	env.AssertNotCalled(t, "ExecuteTaskActivity", mock.Anything, "CREATE_FCL_CDN", mock.Anything, mock.Anything, mock.Anything)
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "ACK_CDNS", mock.Anything, mock.Anything, mock.Anything).
-		Return(emptyMap, nil).Once()
+		Return(TaskResult{Output: emptyMap}, nil).Once()
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "CREATE_BOAT_NOTE", mock.Anything, mock.Anything, mock.Anything).
-		Return(emptyMap, nil).Once()
+		Return(TaskResult{Output: emptyMap}, nil).Once()
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "APPROVE_BOAT_NOTE", mock.Anything, mock.Anything, mock.Anything).
-		Return(emptyMap, nil).Once()
+		Return(TaskResult{Output: emptyMap}, nil).Once()
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
@@ -279,14 +279,14 @@ func TestParallelJoinFlow(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(emptyMap, nil).Once()
+		Return(TaskResult{Output: emptyMap}, nil).Once()
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_B", mock.Anything, mock.Anything, mock.Anything).
-		Return(emptyMap, nil).Once()
+		Return(TaskResult{Output: emptyMap}, nil).Once()
 
 	// TASK_C must only be called ONCE to prove join synchronization works
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_C", mock.Anything, mock.Anything, mock.Anything).
-		Return(emptyMap, nil).Once()
+		Return(TaskResult{Output: emptyMap}, nil).Once()
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
@@ -329,7 +329,7 @@ func TestTaskNodeAppliesInputMapping(t *testing.T) {
 		}
 		value, exists := inputs["local_email"]
 		return exists && value == "user@example.com"
-	}), mock.Anything, mock.Anything).Return(map[string]any{}, nil).Once()
+	}), mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{}}, nil).Once()
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
@@ -361,7 +361,7 @@ func TestTaskNodeWithEmptyInputMappingPassesNoInputs(t *testing.T) {
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_EMPTY_INPUTS", mock.MatchedBy(func(inputs map[string]any) bool {
 		return len(inputs) == 0
-	}), mock.Anything, mock.Anything).Return(map[string]any{}, nil).Once()
+	}), mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{}}, nil).Once()
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
@@ -421,10 +421,10 @@ func TestNodeOutputFlowsIntoSubsetInputMapping(t *testing.T) {
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "NODE1_TASK", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"task_email": "user@example.com",
 			"task_phone": "+123456789",
-		}, nil).Once()
+		}}, nil).Once()
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "NODE2_TASK", mock.MatchedBy(func(inputs map[string]any) bool {
 		if len(inputs) != 1 {
@@ -432,7 +432,7 @@ func TestNodeOutputFlowsIntoSubsetInputMapping(t *testing.T) {
 		}
 		value, exists := inputs["local_email"]
 		return exists && value == "user@example.com"
-	}), mock.Anything, mock.Anything).Return(map[string]any{}, nil).Once()
+	}), mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{}}, nil).Once()
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
@@ -474,7 +474,7 @@ func TestTaskNodeSkipsOptionalInputWhenMissing(t *testing.T) {
 		}
 		value, exists := inputs["local_email"]
 		return exists && value == "user@example.com"
-	}), mock.Anything, mock.Anything).Return(map[string]any{}, nil).Once()
+	}), mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{}}, nil).Once()
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
@@ -511,7 +511,7 @@ func TestTaskNodeAppliesOptionalInputWhenPresent(t *testing.T) {
 		email, emailOk := inputs["local_email"]
 		phone, phoneOk := inputs["local_phone"]
 		return emailOk && email == "user@example.com" && phoneOk && phone == "+123456789"
-	}), mock.Anything, mock.Anything).Return(map[string]any{}, nil).Once()
+	}), mock.Anything, mock.Anything).Return(TaskResult{Output: map[string]any{}}, nil).Once()
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
@@ -537,7 +537,7 @@ func TestTaskNodeSkipsOptionalOutputWhenMissing(t *testing.T) {
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_OPTIONAL_OUTPUTS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"task_email": "user@example.com"}, nil).Once()
+		Return(TaskResult{Output: map[string]any{"task_email": "user@example.com"}}, nil).Once()
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
@@ -571,7 +571,7 @@ func TestTaskNodeFailsWhenRequiredOutputMissing(t *testing.T) {
 	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()
 
 	// The output mapping error parks the node for admin intervention. Abort it so the workflow fails.
 	env.RegisterDelayedCallback(func() {
@@ -605,11 +605,11 @@ func TestEdgesAreReturnedInWorkflowInstance(t *testing.T) {
 
 	// Critical: ensure condition variable exists
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "SUBMIT_CUSDEC", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"consignment_type": "LCL"}, nil)
+		Return(TaskResult{Output: map[string]any{"consignment_type": "LCL"}}, nil)
 
 	// fallback
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil)
@@ -651,7 +651,7 @@ func TestEdgesReferenceValidNodeInstanceIDs(t *testing.T) {
 	env.RegisterActivityWithOptions(acts.AdminParkActivity, activity.RegisterOptions{Name: "AdminParkActivity"})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{}, nil)
+		Return(TaskResult{Output: map[string]any{}}, nil)
 
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil)
@@ -954,9 +954,9 @@ func TestTimerNodeLoopsUntilDelivered(t *testing.T) {
 
 	// Not delivered twice, then delivered on the third poll.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "POLL_STATUS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"delivered": false}, nil).Twice()
+		Return(TaskResult{Output: map[string]any{"delivered": false}}, nil).Twice()
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "POLL_STATUS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"delivered": true}, nil).Once()
+		Return(TaskResult{Output: map[string]any{"delivered": true}}, nil).Once()
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
 
@@ -995,7 +995,7 @@ func TestTimerNodeStopsAtAttemptCap(t *testing.T) {
 
 	// Never delivered: the cap must stop the loop at exactly 3 polls.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "POLL_STATUS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"delivered": false}, nil).Times(3)
+		Return(TaskResult{Output: map[string]any{"delivered": false}}, nil).Times(3)
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil).Once()
 
diff --git a/workflow/manager.go b/workflow/manager.go
index 1db7890..3463e5e 100644
--- a/workflow/manager.go
+++ b/workflow/manager.go
@@ -14,6 +14,7 @@ import (
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/sdk/activity"
 	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/worker"
 	"go.temporal.io/sdk/workflow"
 )
@@ -222,9 +223,12 @@ type UpdateEvent struct {
 // 2. Asynchronous Execution:
 //   - If the work is long-running (e.g. awaits external API callback, human UI interaction, etc.),
 //     return a nil map and an ErrResultPending error.
-//   - The workflow activity pauses and awaits completion. The host application must eventually resume
-//     it by calling Manager.CompleteActivation() with the matching workflow, run, and step IDs.
-type TaskActivationHandler func(payload TaskPayload) (map[string]any, error)
+//   - The Activity ends, and the node waits in the workflow, for as long as it takes, until the host
+//     application calls Manager.CompleteActivation() with the matching workflow, run, and step IDs.
+//
+// ctx is the Activity's: it carries the attempt's deadline (taskActivityTimeout), and is cancelled
+// when the attempt times out, so a call made with it doesn't outlive the attempt.
+type TaskActivationHandler func(ctx context.Context, payload TaskPayload) (map[string]any, error)
 
 // WorkflowCompletion is what the engine reports when a top-level workflow reaches its END node.
 type WorkflowCompletion struct {
@@ -240,8 +244,9 @@ type WorkflowCompletion struct {
 
 // WorkflowCompletionHandler is invoked when the generic DAG workflow successfully reaches an "End" node,
 // providing the final, accumulated state of the workflow variables. It may be invoked more than once
-// for the same workflow (Temporal retries the Activity), so it must be idempotent.
-type WorkflowCompletionHandler func(WorkflowCompletion) error
+// for the same workflow (Temporal retries the Activity), so it must be idempotent. ctx is the
+// Activity's, with the attempt's deadline.
+type WorkflowCompletionHandler func(ctx context.Context, completion WorkflowCompletion) error
 
 // AdminParkHandler is invoked once every time a node parks in NodeStatusAwaitingAdmin —
 // including a re-park after a failed AdminActionRetry, since that's newly actionable
@@ -265,9 +270,10 @@ type Manager interface {
 	// CompleteActivation is called by the external system to resume a paused workflow node.
 	// It routes the output data back into the specific workflow's WorkflowVariables using the
 	// workflowID and activationID (TaskPayload.ActivationID) that were originally emitted via the
-	// TaskActivationHandler. runID may be empty to address the workflow's current run. It fails if
-	// that step is not pending, so a call for a step that has already completed, or that never
-	// existed, is rejected rather than applied to whichever step is active now.
+	// TaskActivationHandler. runID may be empty to address the workflow's current run. It fails with
+	// ErrActivationNotPending if that step is not waiting, so a call for a step that has already
+	// completed, or that never existed, is rejected rather than applied to whichever step is active
+	// now. Of several calls for the same step, exactly one succeeds.
 	CompleteActivation(ctx context.Context, workflowID, runID, activationID string, output map[string]any) error
 
 	// TaskUpdate is used to send an update about the task to the workflow.
@@ -327,8 +333,7 @@ type temporalManagerImpl struct {
 
 // NewTemporalManager creates a new instance of TemporalManager.
 //
-// namespace must be the namespace c was dialed with. client.Client does not expose it, and
-// some calls (e.g. CompleteActivityByID) require it explicitly.
+// namespace must be the namespace c was dialed with. client.Client does not expose it.
 func NewTemporalManager(
 	c client.Client,
 	namespace string,
@@ -394,20 +399,41 @@ func (m *temporalManagerImpl) StartWorkflow(ctx context.Context, ID string, def
 	return nil
 }
 
-// CompleteActivation is invoked by the external application to complete a dormant asynchronous Temporal Activity.
-// WorkflowID is the ID of the workflow
-// runID is the ID of the run
-// activationID is the ID of the step (TaskPayload.ActivationID), which is the Activity's ID
-// output is the key valye pairs that should be added to the global context
+// CompleteActivation delivers a suspended step's output to the workflow waiting on it, as the
+// workflow's CompleteActivation Update (see registerCompletionHandler), and waits for the workflow
+// to accept it. A step that isn't waiting, or a workflow that has ended or doesn't exist, is
+// ErrActivationNotPending.
+//
+// The Update gets no ID of its own: Temporal would answer a second Update with the same ID with the
+// first one's result, so a duplicate completion would succeed instead of being rejected.
 func (m *temporalManagerImpl) CompleteActivation(ctx context.Context, workflowID, runID, activationID string, output map[string]any) error {
-	err := m.temporalClient.CompleteActivityByID(ctx, m.namespace, workflowID, runID, activationID, output, nil)
-	var notFound *serviceerror.NotFound
-	if errors.As(err, &notFound) {
+	handle, err := m.temporalClient.UpdateWorkflow(ctx, client.UpdateWorkflowOptions{
+		WorkflowID:   workflowID,
+		RunID:        runID,
+		UpdateName:   completeActivationUpdate,
+		Args:         []any{activationID, output},
+		WaitForStage: client.WorkflowUpdateStageCompleted,
+	})
+	if err == nil {
+		err = handle.Get(ctx, nil)
+	}
+	if isActivationNotPending(err) {
 		return fmt.Errorf("%w: %w", ErrActivationNotPending, err)
 	}
 	return err
 }
 
+// isActivationNotPending reports whether err is the workflow rejecting a completion for a step that
+// isn't waiting, or the workflow having ended or never existed.
+func isActivationNotPending(err error) bool {
+	var appErr *temporal.ApplicationError
+	if errors.As(err, &appErr) && appErr.Type() == activationNotPendingErrorType {
+		return true
+	}
+	var notFound *serviceerror.NotFound
+	return errors.As(err, &notFound)
+}
+
 func (m *temporalManagerImpl) TaskUpdate(ctx context.Context, workflowID, runID string, event UpdateEvent) error {
 	return m.temporalClient.SignalWorkflow(ctx, workflowID, runID, "TaskUpdateSignal", event)
 }
diff --git a/workflow/manager_test.go b/workflow/manager_test.go
index 900833d..78883e8 100644
--- a/workflow/manager_test.go
+++ b/workflow/manager_test.go
@@ -13,25 +13,66 @@ import (
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/sdk/client"
 	"go.temporal.io/sdk/mocks"
+	"go.temporal.io/sdk/temporal"
 )
 
-// TestTemporalManagerImpl_TaskDone_UsesConfiguredNamespace guards against CompleteActivation completing the
-// activity in a namespace other than the one the manager was configured with.
-func TestTemporalManagerImpl_TaskDone_UsesConfiguredNamespace(t *testing.T) {
+// CompleteActivation sends the step's output as the workflow's CompleteActivation Update, with no
+// Update ID of its own, and waits for the workflow to accept it.
+func TestTemporalManagerImpl_CompleteActivation_SendsTheUpdate(t *testing.T) {
 	mockClient := &mocks.Client{}
+	handle := &mocks.WorkflowUpdateHandle{}
 	output := map[string]any{"k": "v"}
-	mockClient.On("CompleteActivityByID", mock.Anything, "staging", "wf-1", "run-1", "node-1", output, nil).
-		Return(nil)
+	mockClient.On("UpdateWorkflow", mock.Anything, client.UpdateWorkflowOptions{
+		WorkflowID:   "wf-1",
+		RunID:        "run-1",
+		UpdateName:   completeActivationUpdate,
+		Args:         []any{"step-1", output},
+		WaitForStage: client.WorkflowUpdateStageCompleted,
+	}).Return(handle, nil)
+	handle.On("Get", mock.Anything, nil).Return(nil)
 
-	// Go through the constructor so the test also covers it storing the namespace. worker.New
-	// rejects mocks, so construct with a lazy (non-connecting) client, then swap in the mock.
-	lazyClient, err := client.NewLazyClient(client.Options{Namespace: "staging"})
-	require.NoError(t, err)
-	m := NewTemporalManager(lazyClient, "staging", "some-queue", nil, nil).(*temporalManagerImpl)
-	m.temporalClient = mockClient
-
-	require.NoError(t, m.CompleteActivation(context.Background(), "wf-1", "run-1", "node-1", output))
+	m := &temporalManagerImpl{temporalClient: mockClient}
+	require.NoError(t, m.CompleteActivation(context.Background(), "wf-1", "run-1", "step-1", output))
 	mockClient.AssertExpectations(t)
+	handle.AssertExpectations(t)
+}
+
+func TestTemporalManagerImpl_CompleteActivation_NotPending(t *testing.T) {
+	rejected := temporal.NewApplicationError("step step-1 is not waiting to be completed", activationNotPendingErrorType)
+
+	for name, tc := range map[string]struct {
+		updateErr, getErr error
+	}{
+		"the workflow rejects it":           {getErr: rejected},
+		"the workflow has ended or is gone": {updateErr: serviceerror.NewNotFound("workflow execution already completed")},
+	} {
+		t.Run(name, func(t *testing.T) {
+			mockClient := &mocks.Client{}
+			handle := &mocks.WorkflowUpdateHandle{}
+			if tc.updateErr != nil {
+				mockClient.On("UpdateWorkflow", mock.Anything, mock.Anything).Return(nil, tc.updateErr)
+			} else {
+				mockClient.On("UpdateWorkflow", mock.Anything, mock.Anything).Return(handle, nil)
+				handle.On("Get", mock.Anything, nil).Return(tc.getErr)
+			}
+
+			m := &temporalManagerImpl{temporalClient: mockClient}
+			err := m.CompleteActivation(context.Background(), "wf-1", "", "step-1", nil)
+			require.ErrorIs(t, err, ErrActivationNotPending)
+		})
+	}
+}
+
+// Any other failure is returned as it is, not as ErrActivationNotPending: the caller may retry it.
+func TestTemporalManagerImpl_CompleteActivation_OtherErrorPassesThrough(t *testing.T) {
+	mockClient := &mocks.Client{}
+	underlying := serviceerror.NewUnavailable("temporal unavailable")
+	mockClient.On("UpdateWorkflow", mock.Anything, mock.Anything).Return(nil, underlying)
+
+	m := &temporalManagerImpl{temporalClient: mockClient}
+	err := m.CompleteActivation(context.Background(), "wf-1", "", "step-1", nil)
+	require.ErrorIs(t, err, underlying)
+	require.False(t, errors.Is(err, ErrActivationNotPending))
 }
 
 func TestTemporalManagerImpl_GetStatus_NotFound(t *testing.T) {
diff --git a/workflow/manifest_version_test.go b/workflow/manifest_version_test.go
index a1b0413..0996f0e 100644
--- a/workflow/manifest_version_test.go
+++ b/workflow/manifest_version_test.go
@@ -60,20 +60,20 @@ func runRecordingActivations(t *testing.T, def WorkflowDefinition, vars map[stri
 	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) {
+		func(_ context.Context, templateID string, inputs map[string]any, _ string, ref ActivationRef) (TaskResult, error) {
 			mu.Lock()
 			seen[templateID] = append(seen[templateID], ref.ManifestVersion)
 			mu.Unlock()
 			if templateID == "LOAD_ITEMS" {
-				return map[string]any{"items": []any{
+				return TaskResult{Output: map[string]any{"items": []any{
 					map[string]any{"id": "i1", "type": "food"},
 					map[string]any{"id": "i2", "type": "goods"},
-				}}, nil
+				}}}, nil
 			}
 			if c, ok := inputs["commodities"]; ok {
-				return map[string]any{"commodities": c}, nil
+				return TaskResult{Output: map[string]any{"commodities": c}}, nil
 			}
-			return map[string]any{}, nil
+			return TaskResult{Output: map[string]any{}}, nil
 		})
 
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "manifest-test"})
diff --git a/workflow/parallel_gateway_test.go b/workflow/parallel_gateway_test.go
index 7788666..f54389a 100644
--- a/workflow/parallel_gateway_test.go
+++ b/workflow/parallel_gateway_test.go
@@ -88,16 +88,16 @@ func (s *ParallelGatewayTestSuite) TestParallelSplit_ItemFieldMerge_BothBranches
 	def := buildParallelMergeWorkflow(map[string]string{"commodities": "id"})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"items": []any{
 				map[string]any{"id": "item-1", "name": "Cut Flowers"},
 				map[string]any{"id": "item-2", "name": "Timber"},
 			},
-		}, nil).Once()
+		}}, nil).Once()
 
 	// lab_task: annotates every item with lab_result, echoing the identity fields it saw.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything, mock.Anything, mock.Anything).
-		Return(func(_ context.Context, _ string, inputs map[string]any, _ string, _ ActivationRef) (map[string]any, error) {
+		Return(taskOutput(func(_ context.Context, _ string, inputs map[string]any, _ string, _ ActivationRef) (map[string]any, error) {
 			in, _ := inputs["commodities"].([]any)
 			out := make([]any, 0, len(in))
 			for _, raw := range in {
@@ -106,11 +106,11 @@ func (s *ParallelGatewayTestSuite) TestParallelSplit_ItemFieldMerge_BothBranches
 				out = append(out, updated)
 			}
 			return map[string]any{"commodities": out}, nil
-		}).Once()
+		})).Once()
 
 	// visual_task: annotates every item with visual_result, independently of lab_task.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything, mock.Anything, mock.Anything).
-		Return(func(_ context.Context, _ string, inputs map[string]any, _ string, _ ActivationRef) (map[string]any, error) {
+		Return(taskOutput(func(_ context.Context, _ string, inputs map[string]any, _ string, _ ActivationRef) (map[string]any, error) {
 			in, _ := inputs["commodities"].([]any)
 			out := make([]any, 0, len(in))
 			for _, raw := range in {
@@ -119,7 +119,7 @@ func (s *ParallelGatewayTestSuite) TestParallelSplit_ItemFieldMerge_BothBranches
 				out = append(out, updated)
 			}
 			return map[string]any{"commodities": out}, nil
-		}).Once()
+		})).Once()
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "parallel-merge-test-1"})
@@ -193,9 +193,9 @@ func (s *ParallelGatewayTestSuite) TestParallelSplit_GenericMapMerge_DisjointFie
 	}
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TRACK_A", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"value": true}, nil).Once()
+		Return(TaskResult{Output: map[string]any{"value": true}}, nil).Once()
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "TRACK_B", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{"value": true}, nil).Once()
+		Return(TaskResult{Output: map[string]any{"value": true}}, nil).Once()
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "parallel-map-merge-test-1"})
@@ -232,19 +232,19 @@ func (s *ParallelGatewayTestSuite) TestParallelSplit_SameFieldConflict_LastBranc
 	def := buildParallelMergeWorkflow(map[string]string{"commodities": "id"})
 
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"items": []any{map[string]any{"id": "item-1"}},
-		}, nil).Once()
+		}}, nil).Once()
 
 	// Both branches write the SAME field ("failure_reason") with DIFFERENT values.
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"commodities": []any{map[string]any{"id": "item-1", "failure_reason": "lab reason"}},
-		}, nil).Once()
+		}}, nil).Once()
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"commodities": []any{map[string]any{"id": "item-1", "failure_reason": "visual reason"}},
-		}, nil).Once()
+		}}, nil).Once()
 
 	env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"})
 	env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "parallel-conflict-test-1"})
@@ -281,17 +281,17 @@ func (s *ParallelGatewayTestSuite) TestParallelSplit_MergeByID_ItemMissingID_Fai
 
 	// An item is missing the required "id" field
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"items": []any{
 				map[string]any{"name": "No ID item"},
 			},
-		}, nil).Once()
+		}}, nil).Once()
 
 	passThrough := func(_ context.Context, _ string, inputs map[string]any, _ string, _ ActivationRef) (map[string]any, error) {
 		return map[string]any{"commodities": inputs["commodities"]}, nil
 	}
-	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything, mock.Anything, mock.Anything).Return(passThrough)
-	env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything, mock.Anything, mock.Anything).Return(passThrough)
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything, mock.Anything, mock.Anything).Return(taskOutput(passThrough))
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything, mock.Anything, mock.Anything).Return(taskOutput(passThrough))
 
 	env.RegisterDelayedCallback(func() {
 		val, err := env.QueryWorkflow("GetStatus")
@@ -331,15 +331,15 @@ func (s *ParallelGatewayTestSuite) TestParallelSplit_MergeByID_InvalidItemType_F
 
 	// LOAD_ITEMS returns invalid non-slice items
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything, mock.Anything, mock.Anything).
-		Return(map[string]any{
+		Return(TaskResult{Output: map[string]any{
 			"items": "not-a-slice",
-		}, nil).Once()
+		}}, nil).Once()
 
 	passThrough := func(_ context.Context, _ string, inputs map[string]any, _ string, _ ActivationRef) (map[string]any, error) {
 		return map[string]any{"commodities": inputs["commodities"]}, nil
 	}
-	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything, mock.Anything, mock.Anything).Return(passThrough)
-	env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything, mock.Anything, mock.Anything).Return(passThrough)
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything, mock.Anything, mock.Anything).Return(taskOutput(passThrough))
+	env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything, mock.Anything, mock.Anything).Return(taskOutput(passThrough))
 
 	env.RegisterDelayedCallback(func() {
 		val, err := env.QueryWorkflow("GetStatus")
diff --git a/workflow/park_category_test.go b/workflow/park_category_test.go
index 3776ff3..445860b 100644
--- a/workflow/park_category_test.go
+++ b/workflow/park_category_test.go
@@ -229,7 +229,7 @@ func TestParkCategoryIsRecordedForEachFailureKind(t *testing.T) {
 			nodeID: "task",
 			setup: func(env *testsuite.TestWorkflowEnvironment) {
 				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()
 			},
 			category: ParkCategoryOutputMapping,
 		},
@@ -240,7 +240,7 @@ func TestParkCategoryIsRecordedForEachFailureKind(t *testing.T) {
 			nodeID: "task",
 			setup: func(env *testsuite.TestWorkflowEnvironment) {
 				env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_MISSING_REQUIRED_OUTPUT", mock.Anything, mock.Anything, mock.Anything).
-					Return(nil, temporal.NewNonRetryableApplicationError("boom", "TaskFailure", nil)).Once()
+					Return(TaskResult{}, temporal.NewNonRetryableApplicationError("boom", "TaskFailure", nil)).Once()
 			},
 			category: ParkCategoryTaskFailure,
 		},
@@ -322,9 +322,9 @@ func TestParkCategoryOfNodeInsideParallelBranch(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_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")
 
@@ -364,7 +364,7 @@ func TestParkCategoryIsClearedOnceResolved(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() {
@@ -401,7 +401,7 @@ func TestNodeInfoMappingsAreOnlySetWhileParked(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_INPUTS", 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.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{"global_user_email": "user@example.com"})
@@ -426,7 +426,7 @@ func TestNodeInfoCarriesMappingsForParkedTask(t *testing.T) {
 	def = mustParseDefinition(t, missingRequiredOutputWorkflowJSON)
 	parked = parkAndInspect(t, def, map[string]any{}, "task", func(env *testsuite.TestWorkflowEnvironment) {
 		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()
 	})
 
 	require.Equal(t, map[string]string{"task_phone": "global_user_phone"}, parked.OutputMapping)
@@ -439,7 +439,7 @@ func TestNilActivityResultWithRequiredMappingParks(t *testing.T) {
 	def := mustParseDefinition(t, missingRequiredOutputWorkflowJSON)
 	parked := parkAndInspect(t, def, map[string]any{}, "task", func(env *testsuite.TestWorkflowEnvironment) {
 		env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_MISSING_REQUIRED_OUTPUT", mock.Anything, mock.Anything, mock.Anything).
-			Return(nil, nil).Once()
+			Return(TaskResult{}, nil).Once()
 	})
 
 	require.Equal(t, NodeStatusAwaitingAdmin, parked.Status)
diff --git a/workflow/task_retry_test.go b/workflow/task_retry_test.go
index 0369615..cd61cd7 100644
--- a/workflow/task_retry_test.go
+++ b/workflow/task_retry_test.go
@@ -49,10 +49,10 @@ func newSingleTaskEnv(t *testing.T, execute func(attempt int) (map[string]any, e
 
 	attempts := 0
 	env.OnActivity("ExecuteTaskActivity", mock.Anything, "FLAKY", mock.Anything, mock.Anything, mock.Anything).
-		Return(func(context.Context, string, map[string]any, string, ActivationRef) (map[string]any, error) {
+		Return(taskOutput(func(context.Context, string, map[string]any, string, ActivationRef) (map[string]any, error) {
 			attempts++
 			return execute(attempts)
-		})
+		}))
 	env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
 	return env, def, &attempts
 }
@@ -201,3 +201,15 @@ func TestSplitParksWhenItsDefinitionCantBeFetched(t *testing.T) {
 	require.Error(t, env.GetWorkflowError())
 	require.Equal(t, taskMaxAttempts, attempts)
 }
+
+// taskOutput adapts a mock that answers ExecuteTaskActivity with the step's output to the Activity's
+// signature: the output of a step that completed synchronously, or the mock's error.
+func taskOutput(fn func(context.Context, string, map[string]any, string, ActivationRef) (map[string]any, error)) func(context.Context, string, map[string]any, string, ActivationRef) (TaskResult, error) {
+	return func(ctx context.Context, template string, inputs map[string]any, caseID string, step ActivationRef) (TaskResult, error) {
+		output, err := fn(ctx, template, inputs, caseID, step)
+		if err != nil {
+			return TaskResult{}, err
+		}
+		return TaskResult{Output: output}, nil
+	}
+}
diff --git a/workflow/workflow.go b/workflow/workflow.go
index eeeeab2..efa3d08 100644
--- a/workflow/workflow.go
+++ b/workflow/workflow.go
@@ -35,6 +35,10 @@ type graphInterpreter struct {
 	// pendingAdminResolutions holds a Settable for each node currently parked in
 	// NodeStatusAwaitingAdmin, keyed by that park's ActivationID. See admin_recovery.go.
 	pendingAdminResolutions map[string]workflow.Settable
+
+	// pendingCompletions holds a Settable for each TASK step that may be completed through
+	// CompleteActivation, keyed by its ActivationID. See completion.go.
+	pendingCompletions map[string]workflow.Settable
 }
 
 // GraphInterpreterWorkflow is the entry point for the Temporal workflow that interprets a graph definition.
@@ -99,6 +103,7 @@ func GraphInterpreterWorkflow(ctx workflow.Context, def WorkflowDefinition, init
 		instance:                instance,
 		edgeTokens:              make(map[string]int),
 		pendingAdminResolutions: make(map[string]workflow.Settable),
+		pendingCompletions:      make(map[string]workflow.Settable),
 	}
 	interp.buildIndexes()
 
@@ -131,6 +136,9 @@ func GraphInterpreterWorkflow(ctx workflow.Context, def WorkflowDefinition, init
 	})
 
 	interp.startAdminResolutionDispatcher(ctx)
+	if err := interp.registerCompletionHandler(ctx); err != nil {
+		return nil, fmt.Errorf("failed to set %s update handler: %w", completeActivationUpdate, err)
+	}
 
 	// The END node's WorkflowCompletedActivity runs with these options. Every other Activity sets its
 	// own (handleTaskNode's ExecuteTaskActivity among them).
@@ -463,10 +471,10 @@ func (g *graphInterpreter) caseID() string {
 	return g.instance.ID
 }
 
-// taskActivityTimeout bounds one attempt of a task's Activity. It is long because a step that waits
-// for a person (ErrResultPending) holds its Activity open until CompleteActivation, and that wait
-// counts against this timeout.
-const taskActivityTimeout = 24 * time.Hour * 365
+// taskActivityTimeout bounds one attempt of a task's Activity: running the step's plugin. A step that
+// waits for a person ends its Activity first (see completion.go), so the wait doesn't count against
+// it. An attempt whose worker died or whose call hangs is retried when this runs out.
+const taskActivityTimeout = 10 * time.Minute
 
 // completionActivityTimeout bounds one attempt of WorkflowCompletedActivity. The completion hook does
 // its work and returns without waiting for anyone, so a few minutes is ample, and a worker that dies
@@ -495,20 +503,34 @@ func (g *graphInterpreter) handleTaskNode(ctx workflow.Context, nodeInfo *NodeIn
 		return err
 	}
 
-	var result map[string]any
+	// The step may be completed from the moment its Activity starts: a host can complete a step it
+	// suspended before the Activity has reported back.
+	completion, unregister := g.expectCompletion(ctx, nodeInfo.ActivationID)
 
 	// nodeInfo.ActivationID was minted by the caller (mintActivation, via executeNode or an admin
-	// retry) before dispatching here; the Activity is addressed by that same ID so a stale
-	// completion from an earlier run of this node no longer matches.
+	// retry) before dispatching here; the Activity is addressed by that same ID, and so is the step's
+	// completion, so a stale completion from an earlier run of this node no longer matches.
 	nodeCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
 		ActivityID:          nodeInfo.ActivationID,
 		StartToCloseTimeout: taskActivityTimeout,
 		RetryPolicy:         taskRetryPolicy,
 	})
 
+	var taskResult TaskResult
 	err = workflow.ExecuteActivity(nodeCtx, "ExecuteTaskActivity", node.TaskTemplateID, inputs, g.caseID(),
-		ActivationRef{NodeID: node.ID, Seq: nodeInfo.Seq, ManifestVersion: g.manifestVersion()}).Get(ctx, &result)
+		ActivationRef{NodeID: node.ID, Seq: nodeInfo.Seq, ManifestVersion: g.manifestVersion()}).Get(ctx, &taskResult)
+	result := taskResult.Output
+	if err == nil && taskResult.Suspended {
+		// Wait, outside any Activity, for CompleteActivation to deliver the step's output.
+		err = completion.Get(ctx, &result)
+	}
+	// The step now takes no completion: it completed, failed or was cancelled. This must happen
+	// here, not on return, since the node returns only after the rest of the graph has run.
+	unregister()
 	if err != nil {
+		if ctx.Err() != nil {
+			return err
+		}
 		return withCategory(ParkCategoryTaskFailure, err)
 	}