From 2686d7bb725851a6eb685107e7912279f18ff207 Mon Sep 17 00:00:00 2001 From: Martin Vu <22mvu7@gmail.com> Date: Sun, 30 Aug 2026 00:44:47 -0700 Subject: [PATCH 1/4] Reject state collisions in LoopStartOperator --- amber/src/main/python/core/models/operator.py | 18 +++++++++++++----- .../test/python/core/models/test_operator.py | 19 ++++++++++++++++++- 2 files changed, 31 insertions(+), 6 deletions(-) diff --git a/amber/src/main/python/core/models/operator.py b/amber/src/main/python/core/models/operator.py index 4e65fbb2ab2..acdc75040a8 100644 --- a/amber/src/main/python/core/models/operator.py +++ b/amber/src/main/python/core/models/operator.py @@ -366,11 +366,19 @@ class LoopStartOperator(TableOperator): @overrides.final def process_state(self, state: State, port: int) -> Optional[State]: - # First-entry only: merge upstream state into self.state. The nested - # pass-through (a frame already stamped with a LoopStartId) and all - # loop_counter bookkeeping are owned by the worker runtime - # (main_loop._process_state_frame), so this operator never sees the - # counter and never mutates the State it is handed. + # First-entry only: merge non-conflicting upstream state into self.state. + # The nested pass-through (a frame already stamped with a LoopStartId) and + # all loop_counter bookkeeping are owned by the worker runtime + # (main_loop._process_state_frame), so this operator never sees the counter + # and never mutates the State it is handed. + collisions = self.state.keys() & state.keys() + + if collisions: + raise ValueError( + f"Loop state variable(s) cannot be overwritten: " + f"{', '.join(sorted(collisions))}" + ) + self.state.update(state) return None diff --git a/amber/src/test/python/core/models/test_operator.py b/amber/src/test/python/core/models/test_operator.py index bb18853350e..194ecbec45a 100644 --- a/amber/src/test/python/core/models/test_operator.py +++ b/amber/src/test/python/core/models/test_operator.py @@ -28,7 +28,7 @@ Tuple, TupleOperatorV2, ) -from core.models.operator import Operator, TableOperator +from core.models.operator import Operator, TableOperator, LoopStartOperator class _ConcreteOperator(TupleOperatorV2): @@ -161,6 +161,10 @@ def process_table(self, table, port): self.received_tables.append(table) yield None +class _ConcreteLoopStart(LoopStartOperator): + def process_table(self, table, port): + yield table + class TestPythonTemplateDecoder: def test_stdlib_decoder_decodes_str_input(self): @@ -397,6 +401,19 @@ def test_buffers_are_keyed_by_port(self): rows = list(op.received_tables[0].as_tuples()) assert rows == [Tuple({"x": 1})] +class TestLoopStartOperator: + def test_process_state_rejects_overwriting_loop_variable(self): + op = _ConcreteLoopStart() + op.state = State({"i": 0}) + + with pytest.raises( + ValueError, + match=r"Loop state variable\(s\) cannot be overwritten: i", + ): + op.process_state(State({"i": 999}), port=0) + + assert op.state["i"] == 0 + class TestSourceOperatorFinalMethods: """SourceOperator replaces both TupleOperatorV2 tuple hooks: on_finish is a From b10127eb19237a9bb1d4714ef7ab6626a54abd1b Mon Sep 17 00:00:00 2001 From: Martin Vu <22mvu7@gmail.com> Date: Sun, 30 Aug 2026 01:12:42 -0700 Subject: [PATCH 2/4] style: format operator test --- amber/src/test/python/core/models/test_operator.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/amber/src/test/python/core/models/test_operator.py b/amber/src/test/python/core/models/test_operator.py index 194ecbec45a..bde76d2ec18 100644 --- a/amber/src/test/python/core/models/test_operator.py +++ b/amber/src/test/python/core/models/test_operator.py @@ -161,6 +161,7 @@ def process_table(self, table, port): self.received_tables.append(table) yield None + class _ConcreteLoopStart(LoopStartOperator): def process_table(self, table, port): yield table @@ -401,6 +402,7 @@ def test_buffers_are_keyed_by_port(self): rows = list(op.received_tables[0].as_tuples()) assert rows == [Tuple({"x": 1})] + class TestLoopStartOperator: def test_process_state_rejects_overwriting_loop_variable(self): op = _ConcreteLoopStart() From ccc280ee4bfd228c287384a593c73b923dac6cef Mon Sep 17 00:00:00 2001 From: Martin Vu <22mvu7@gmail.com> Date: Sun, 30 Aug 2026 01:46:04 -0700 Subject: [PATCH 3/4] prevent LoopStart state variable collisions --- .../python/core/models/test_loop_operators.py | 18 +++++++++++++--- .../test/python/core/models/test_operator.py | 21 +------------------ 2 files changed, 16 insertions(+), 23 deletions(-) diff --git a/amber/src/test/python/core/models/test_loop_operators.py b/amber/src/test/python/core/models/test_loop_operators.py index 6b496a5f17b..8b844d39ac4 100644 --- a/amber/src/test/python/core/models/test_loop_operators.py +++ b/amber/src/test/python/core/models/test_loop_operators.py @@ -134,12 +134,24 @@ def test_first_time_state_is_merged_into_self_state_and_none_is_returned(self): # nothing flows downstream of LoopStart until the table is in. op = _StubLoopStart() op.open() - op.state["i"] = 0 # simulate the user's initialization + op.state["i"] = 0 result = op.process_state(State({"upstream_key": "v"}), port=0) - assert result is None, "first-time state must not be forwarded" - assert op.state["upstream_key"] == "v", "state was not merged into self.state" + assert result is None + assert op.state["upstream_key"] == "v" + + def test_state_rejects_overwriting_loop_variable(self): + op = _StubLoopStart() + op.open() + + with pytest.raises( + ValueError, + match=r"Loop state variable\(s\) cannot be overwritten: i", + ): + op.process_state(State({"i": 999}), port=0) + + assert op.state["i"] == 0 # NOTE: LoopStart re-entry (+1) is owned by the worker runtime now, not the # operator (which only does the first-entry merge above). It and the nested diff --git a/amber/src/test/python/core/models/test_operator.py b/amber/src/test/python/core/models/test_operator.py index bde76d2ec18..bb18853350e 100644 --- a/amber/src/test/python/core/models/test_operator.py +++ b/amber/src/test/python/core/models/test_operator.py @@ -28,7 +28,7 @@ Tuple, TupleOperatorV2, ) -from core.models.operator import Operator, TableOperator, LoopStartOperator +from core.models.operator import Operator, TableOperator class _ConcreteOperator(TupleOperatorV2): @@ -162,11 +162,6 @@ def process_table(self, table, port): yield None -class _ConcreteLoopStart(LoopStartOperator): - def process_table(self, table, port): - yield table - - class TestPythonTemplateDecoder: def test_stdlib_decoder_decodes_str_input(self): decoder = Operator.PythonTemplateDecoder.StdlibBase64Decoder() @@ -403,20 +398,6 @@ def test_buffers_are_keyed_by_port(self): assert rows == [Tuple({"x": 1})] -class TestLoopStartOperator: - def test_process_state_rejects_overwriting_loop_variable(self): - op = _ConcreteLoopStart() - op.state = State({"i": 0}) - - with pytest.raises( - ValueError, - match=r"Loop state variable\(s\) cannot be overwritten: i", - ): - op.process_state(State({"i": 999}), port=0) - - assert op.state["i"] == 0 - - class TestSourceOperatorFinalMethods: """SourceOperator replaces both TupleOperatorV2 tuple hooks: on_finish is a source's only output path, and process_tuple is deliberately inert because From dc5faff162b95f12cc1690c85c4a33bdd5e70ebe Mon Sep 17 00:00:00 2001 From: Martin Vu <22mvu7@gmail.com> Date: Sun, 30 Aug 2026 01:56:55 -0700 Subject: [PATCH 4/4] update LoopStart state collision test --- amber/src/test/python/core/models/test_loop_operators.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/amber/src/test/python/core/models/test_loop_operators.py b/amber/src/test/python/core/models/test_loop_operators.py index 8b844d39ac4..f9d8aea234e 100644 --- a/amber/src/test/python/core/models/test_loop_operators.py +++ b/amber/src/test/python/core/models/test_loop_operators.py @@ -325,8 +325,8 @@ def test_accumulator_persists_and_reserved_names_never_leak(self): # # Each pass of the while loop mimics one engine iteration: the # LoopStart region is re-executed (a fresh operator whose open() seeds - # the loop variables, the back-edge state overriding them, and the - # upstream table re-read), the produced state crosses the materialized + # the loop variables, the worker runtime restores the back-edge state, + # and the upstream table re-read), the produced state crosses the materialized # channel (a State to_tuple/from_tuple round-trip), the LoopEnd runs # the user update and evaluates the condition, and on continuation # only the user loop variables cross the back-edge. @@ -344,7 +344,7 @@ def test_accumulator_persists_and_reserved_names_never_leak(self): ) start.open() if back_edge is not None: - start.process_state(back_edge, port=0) + start.state.update(back_edge) for row in rows: list(start.process_tuple(row, port=0)) emitted.extend(o for o in start.on_finish(port=0) if o is not None)