From 1f0aa1da8fb35bb4f972b4b5d1e7e4d3ebf0c54a Mon Sep 17 00:00:00 2001 From: Alan Zhang Date: Mon, 20 Jul 2026 00:24:28 -0700 Subject: [PATCH 1/2] [python] Propagate config model reconstruction errors in Action deserialize --- python/flink_agents/plan/actions/action.py | 4 +-- python/flink_agents/plan/tests/test_action.py | 25 +++++++++++++++++++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/python/flink_agents/plan/actions/action.py b/python/flink_agents/plan/actions/action.py index c3ad31457..a1c4a360a 100644 --- a/python/flink_agents/plan/actions/action.py +++ b/python/flink_agents/plan/actions/action.py @@ -95,11 +95,11 @@ def __custom_deserialize(self) -> "Action": self["config"][name] = value["value"] return self for name, value in config.items(): - try: + if isinstance(value, list | tuple): module = importlib.import_module(value[0]) clazz = getattr(module, value[1]) self["config"][name] = clazz.model_validate(value[2]) - except Exception: # noqa : PERF203 + else: self["config"][name] = value return self diff --git a/python/flink_agents/plan/tests/test_action.py b/python/flink_agents/plan/tests/test_action.py index f93ee4531..f95491089 100644 --- a/python/flink_agents/plan/tests/test_action.py +++ b/python/flink_agents/plan/tests/test_action.py @@ -120,3 +120,28 @@ def test_action_deserialize_java_shape_config_unwraps_primitives() -> None: "rate": 1.5, "label": "fast", } + + +def test_action_deserialize_python_config_propagates_reconstruction_error() -> None: + """A serialized model whose module is missing must fail loudly. + + The list-shaped entry is treated as a serialized model, so importing a + non-existent module raises instead of being silently swallowed. + """ + json_str = json.dumps( + { + "name": "legal", + "exec": { + "func_type": "PythonFunction", + "module": "flink_agents.plan.tests.test_action", + "qualname": "legal_signature", + }, + "trigger_conditions": ["_input_event"], + "config": { + "__config_type__": "python", + "broken": ["nonexistent_module_xyz", "SomeClass", {}], + }, + } + ) + with pytest.raises(ModuleNotFoundError): + Action.model_validate_json(json_str) From baf357ccb996d0e4af8793bc741f7dde290faf2c Mon Sep 17 00:00:00 2001 From: Alan Zhang Date: Wed, 22 Jul 2026 10:45:09 -0700 Subject: [PATCH 2/2] [python] Tag serialized config models with a marker to avoid misreading plain lists --- python/flink_agents/plan/actions/action.py | 26 +++++++++----- .../plan/tests/resources/action.json | 11 +++--- python/flink_agents/plan/tests/test_action.py | 34 ++++++++++++++++--- 3 files changed, 53 insertions(+), 18 deletions(-) diff --git a/python/flink_agents/plan/actions/action.py b/python/flink_agents/plan/actions/action.py index a1c4a360a..9d57ae7c9 100644 --- a/python/flink_agents/plan/actions/action.py +++ b/python/flink_agents/plan/actions/action.py @@ -26,6 +26,9 @@ from flink_agents.plan.function import Function, JavaFunction, PythonFunction _CONFIG_TYPE = "__config_type__" +# Tags a config entry that we serialized from a pydantic model, so on the way +# back we only rebuild the ones we tagged and leave plain values alone. +_PYDANTIC_MODEL_MARKER = "__pydantic_model__" class Action(BaseModel): @@ -70,11 +73,12 @@ def __serialize_config(self, config: Dict[str, Any]) -> Dict[str, Any] | None: data[_CONFIG_TYPE] = "python" for name, value in config.items(): if isinstance(value, BaseModel): - data[name] = ( - inspect.getmodule(value).__name__, - value.__class__.__name__, - value, - ) + data[name] = { + _PYDANTIC_MODEL_MARKER: True, + "module": inspect.getmodule(value).__name__, + "class": value.__class__.__name__, + "value": value, + } else: data[name] = value return data @@ -95,10 +99,14 @@ def __custom_deserialize(self) -> "Action": self["config"][name] = value["value"] return self for name, value in config.items(): - if isinstance(value, list | tuple): - module = importlib.import_module(value[0]) - clazz = getattr(module, value[1]) - self["config"][name] = clazz.model_validate(value[2]) + # Rebuild only entries with our marker, so a plain list or dict + # value is not treated as a model by mistake. Do not catch errors: + # if the class cannot be imported (e.g. missing on the worker), + # fail here with the real cause instead of a confusing error later. + if isinstance(value, dict) and value.get(_PYDANTIC_MODEL_MARKER): + module = importlib.import_module(value["module"]) + clazz = getattr(module, value["class"]) + self["config"][name] = clazz.model_validate(value["value"]) else: self["config"][name] = value return self diff --git a/python/flink_agents/plan/tests/resources/action.json b/python/flink_agents/plan/tests/resources/action.json index 09393e475..08cdad3fb 100644 --- a/python/flink_agents/plan/tests/resources/action.json +++ b/python/flink_agents/plan/tests/resources/action.json @@ -10,10 +10,11 @@ ], "config": { "__config_type__": "python", - "output_schema": [ - "flink_agents.api.agents.types", - "OutputSchema", - { + "output_schema": { + "__pydantic_model__": true, + "module": "flink_agents.api.agents.types", + "class": "OutputSchema", + "value": { "output_schema": { "names": [ "result" @@ -23,6 +24,6 @@ ] } } - ] + } } } \ No newline at end of file diff --git a/python/flink_agents/plan/tests/test_action.py b/python/flink_agents/plan/tests/test_action.py index f95491089..08f0d5dd2 100644 --- a/python/flink_agents/plan/tests/test_action.py +++ b/python/flink_agents/plan/tests/test_action.py @@ -123,10 +123,10 @@ def test_action_deserialize_java_shape_config_unwraps_primitives() -> None: def test_action_deserialize_python_config_propagates_reconstruction_error() -> None: - """A serialized model whose module is missing must fail loudly. + """A tagged model whose module is missing must fail loudly. - The list-shaped entry is treated as a serialized model, so importing a - non-existent module raises instead of being silently swallowed. + The entry is marked as a serialized model, so importing a non-existent + module raises instead of being silently swallowed. """ json_str = json.dumps( { @@ -139,9 +139,35 @@ def test_action_deserialize_python_config_propagates_reconstruction_error() -> N "trigger_conditions": ["_input_event"], "config": { "__config_type__": "python", - "broken": ["nonexistent_module_xyz", "SomeClass", {}], + "broken": { + "__pydantic_model__": True, + "module": "nonexistent_module_xyz", + "class": "SomeClass", + "value": {}, + }, }, } ) with pytest.raises(ModuleNotFoundError): Action.model_validate_json(json_str) + + +def test_action_deserialize_python_config_preserves_plain_list() -> None: + """A plain user list must survive untouched, not be mistaken for a model.""" + json_str = json.dumps( + { + "name": "legal", + "exec": { + "func_type": "PythonFunction", + "module": "flink_agents.plan.tests.test_action", + "qualname": "legal_signature", + }, + "trigger_conditions": ["_input_event"], + "config": { + "__config_type__": "python", + "hosts": ["host-a", "host-b", "host-c"], + }, + } + ) + action = Action.model_validate_json(json_str) + assert action.config == {"hosts": ["host-a", "host-b", "host-c"]}