[Bug] Child workflow search attributes aren't decrypted in 1.18.0 #1129
Description
Describe the bug
As of 1.18.0, launching a child workflow with search_attributes fails when using a custom payload_codec.
On 1.17.0 this completes succesfully, but on 1.18.0 we get the following error:
2025-09-29T20:18:32.022484Z WARN temporal_sdk_core::worker::workflow: Error while completing workflow activation error=status: InvalidArgument, message: "invalid SearchAttributes on StartChildWorkflowCommand: invalid value for search attribute show_name of type Keyword: value from <metadata:{key:\"encoding\" value:\"binary/encrypted\"} metadata:{key:\"encryption-key-id\" value:\"test-key-id\"} data:\"\\xf9H\\x99+\\xb6\\xee\\xffo\\xd4}\\xde\\xe9\\x7fg\\xb4\\x89\\xa5\\xb4\\x10\\x0e\\xefT\\xd2a\\xf2\\xb9\\t\\xb2t\\xbd\\xa0\\xae\\x8d<\\xf7\\xd2?\\xe4\\xca6X\\xfb\\x83\\xf2\\xb0\\x1aՔ\\x0e\\xbbZ\\xe4\\x94ZU\\x81d\\xf3e\\xc4Zx\\x98\\xf0h\\xc2\\x10Ӑ\\x84\\xfdm\\x9b\\x15S¡8\\x8fI\\xf2R\">. WorkflowId=child-Temporal WorkflowType=ChildWorkflow Namespace=default", details: [], metadata: MetadataMap { headers: {"content-type": "application/grpc"} }
Minimal Reproduction
Based off the sample https://github.com/temporalio/samples-python/tree/main/encryption
import asyncio
import dataclasses
import temporalio.converter
from temporalio import workflow
from temporalio.client import Client
from temporalio.worker import Worker
import os
from typing import Iterable, List
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from temporalio.api.common.v1 import Payload
from temporalio.converter import PayloadCodec
from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes
default_key = b"test-key-test-key-test-key-test!"
default_key_id = "test-key-id"
class EncryptionCodec(PayloadCodec):
def __init__(self, key_id: str = default_key_id, key: bytes = default_key) -> None:
super().__init__()
self.key_id = key_id
# We are using direct AESGCM to be compatible with samples from
# TypeScript and Go. Pure Python samples may prefer the higher-level,
# safer APIs.
self.encryptor = AESGCM(key)
async def encode(self, payloads: Iterable[Payload]) -> List[Payload]:
# We blindly encode all payloads with the key and set the metadata
# saying which key we used
return [
Payload(
metadata={
"encoding": b"binary/encrypted",
"encryption-key-id": self.key_id.encode(),
},
data=self.encrypt(p.SerializeToString()),
)
for p in payloads
]
async def decode(self, payloads: Iterable[Payload]) -> List[Payload]:
ret: List[Payload] = []
for p in payloads:
# Ignore ones w/out our expected encoding
if p.metadata.get("encoding", b"").decode() != "binary/encrypted":
ret.append(p)
continue
# Confirm our key ID is the same
key_id = p.metadata.get("encryption-key-id", b"").decode()
if key_id != self.key_id:
raise ValueError(f"Unrecognized key ID {key_id}. Current key ID is {self.key_id}.")
# Decrypt and append
ret.append(Payload.FromString(self.decrypt(p.data)))
return ret
def encrypt(self, data: bytes) -> bytes:
nonce = os.urandom(12)
return nonce + self.encryptor.encrypt(nonce, data, None)
def decrypt(self, data: bytes) -> bytes:
return self.encryptor.decrypt(data[:12], data[12:], None)
@workflow.defn(name="Workflow")
class GreetingWorkflow:
@workflow.run
async def run(self, name: str) -> str:
print(
await workflow.execute_child_workflow(
workflow=ChildWorkflow.run,
arg=name,
id=f"child-{name}",
search_attributes=workflow.info().typed_search_attributes,
)
)
return f"Hello, {name}"
@workflow.defn(name="ChildWorkflow")
class ChildWorkflow:
@workflow.run
async def run(self, name: str) -> str:
return f"Hello from child, {name}"
async def main():
# Connect client
client = await Client.connect(
"localhost:7233",
# Use the default converter, but change the codec
data_converter=dataclasses.replace(temporalio.converter.default(), payload_codec=EncryptionCodec()),
)
# Run a worker for the workflow
async with Worker(
client,
task_queue="encryption-task-queue",
workflows=[GreetingWorkflow, ChildWorkflow],
):
# Run workflow
result = await client.execute_workflow(
GreetingWorkflow.run,
"Temporal",
id=f"encryption-workflow-id",
task_queue="encryption-task-queue",
search_attributes=TypedSearchAttributes(
[SearchAttributePair(SearchAttributeKey.for_keyword("show_name"), "test_show")]
),
)
print(f"Workflow result: {result}")
if __name__ == "__main__":
asyncio.run(main())Environment/Versions
- OS and processor: Windows 11
- Temporal Version: Python SDK 1.18.0
- Occurs on both the local temporal CLI and temporal cloud.