Small fixes for throughput stress with python worker (#233)

## What was changed
Two small fixes to allow `throughput_stress` to run on the Python
worker:
- Provide a default `BackoffCoefficient` for the `DefaultLocalActivity`
action. The Go SDK provides default values for local activity retry
policy fields (if missing), this is not the case in other SDKs. As such,
we provide a default backoff coefficient when we create the action
- Pass search attributes through to child workflows on the python
worker. Child workflows on the python worker were not including the
throughput stress search attribute, causing the visibility count check
to fail (and consequently, the scenario).

## Why?
We want the python worker to be able to run the throughput stress
scenario successfully

2. How was this tested:
Previously:
```
go run ./cmd run-scenario-with-worker \
    --scenario throughput_stress \
    --language python \
    --iterations 10 \
    --option internal-iterations=10
```
failed. With these fixes it succeeds.

Author: THardy98

cmd/dev/util.go

@@ -19,14 +19,14 @@ var (
 		"dotnet", "go", "java", "node", "protoc", "python", "rust",
 	}
 	toolDependencies = map[string][]string{
-		"python":     {"python", "uv", "poe"},
+		"python":     {"python3", "uv", "poe"},
 		"typescript": {"node"},
 		"protoc":     {"protoc-gen-go"},
 	}
 	toolVersionCommands = map[string][]string{
 		"go":            {"go", "version"},
 		"java":          {"java", "-version"},
-		"python":        {"python", "--version"},
+		"python3":       {"python3", "--version"},
 		"node":          {"node", "--version"},
 		"dotnet":        {"dotnet", "--version"},
 		"cargo":         {"cargo", "--version"},

loadgen/kitchensink/helpers.go

@@ -155,8 +155,9 @@ func DefaultLocalActivity(activity *ExecuteActivityAction) *Action {
 		IsLocal: &emptypb.Empty{},
 	}
 	activity.RetryPolicy = &common.RetryPolicy{
-		InitialInterval: durationpb.New(10 * time.Millisecond),
-		MaximumAttempts: 10,
+		InitialInterval:    durationpb.New(10 * time.Millisecond),
+		MaximumAttempts:    10,
+		BackoffCoefficient: 2.0,
 	}
 	return &Action{
 		Variant: &Action_ExecActivity{

scenarios/throughput_stress.go

@@ -287,6 +287,7 @@ func (t *tpsExecutor) Run(ctx context.Context, info loadgen.ScenarioInfo) error
 
 	var sb strings.Builder
 	sb.WriteString("[Scenario completion summary] ")
+	sb.WriteString(fmt.Sprintf("Run ID: %s, ", info.RunID))
 	sb.WriteString(fmt.Sprintf("Total iterations completed: %d, ", completedIterations))
 	sb.WriteString(fmt.Sprintf("Total child workflows: %d (%d per iteration), ", completedChildWorkflows, t.config.InternalIterations))
 	sb.WriteString(fmt.Sprintf("Total continue-as-new workflows: %d (%d per iteration), ", continueAsNewWorkflows, continueAsNewPerIter))

workers/python/kitchen_sink.py

@@ -14,6 +14,7 @@
     SearchAttributeKey,
     SearchAttributeUpdate,
 )
+from temporalio.converter import DefaultPayloadConverter
 from temporalio.workflow import ActivityHandle, ChildWorkflowHandle
 
 from protos.kitchen_sink_pb2 import (
@@ -129,9 +130,15 @@ async def handle_action(self, action: Action) -> Optional[Payload]:
             child_action = action.exec_child_workflow
             child = child_action.workflow_type or "kitchenSink"
             args = [RawValue(i) for i in child_action.input]
+
             await handle_awaitable_choice(
                 workflow.start_child_workflow(
-                    child, id=child_action.workflow_id, args=args
+                    child,
+                    id=child_action.workflow_id,
+                    args=args,
+                    search_attributes=decode_search_attrs(
+                        child_action.search_attributes, DefaultPayloadConverter()
+                    ),
                 ),
                 child_action.awaitable_choice,
                 after_started_fn=wait_task_complete,
@@ -317,3 +324,10 @@ def convert_act_cancel_type(
         return temporalio.workflow.ActivityCancellationType.ABANDON
     else:
         raise NotImplementedError("Unknown cancellation type " + str(ctype))
+
+
+def decode_search_attrs(msg_map, converter):
+    return {
+        k: v if isinstance(v := converter.from_payload(p), list) else [v]
+        for k, p in msg_map.items()
+    }