feat(engine): add TryCatch/Finally control blocks with in-band error handling

amber/src/bench/scala/org/apache/texera/amber/bench/ArrowFlightActorBench.scala

@@ -287,7 +287,8 @@ object ArrowFlightActorBench {
             1,
             OpExecWithCode(IdentityPythonCode, "python"),
             isSource = false,
-            loopStartStateUris = Map.empty
+            loopStartStateUris = Map.empty,
+            guarded = false
           ),
           ctx,
           0L

amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/rpc/controlcommands.proto

@@ -260,6 +260,11 @@ message InitializeExecutorRequest {
   // the consumed StateFrame and writes the next-iteration state there. Empty
   // for plans without loops.
   map<string, string> loopStartStateUris = 4;
+  // True iff this operator sits inside a try/catch frame's cone (set by
+  // TryCatchFramePass). A guarded worker turns its own executor failure into
+  // an in-band error State and drains; an unguarded worker keeps the default
+  // behavior: report and pause.
+  bool guarded = 5;
 }
 
 message UpdateExecutorRequest {

amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/sendsemantics/partitionings.proto

@@ -35,6 +35,7 @@ message Partitioning{
     HashBasedShufflePartitioning hashBasedShufflePartitioning = 3;
     RangeBasedShufflePartitioning rangeBasedShufflePartitioning = 4;
     BroadcastPartitioning broadcastPartitioning = 5;
+    SignalPartitioning signalPartitioning = 6;
   }
 }
 
@@ -66,3 +67,11 @@ message BroadcastPartitioning{
   int32 batchSize = 1;
   repeated core.ChannelIdentity channels = 2;
 }
+
+// Signal links (try/catch frame wiring): data tuples are dropped at the
+// sender; only States, ECMs and END_CHANNEL traverse the link. Channels must
+// still be declared so the buffers/channels exist for those control payloads.
+message SignalPartitioning{
+  int32 batchSize = 1;
+  repeated core.ChannelIdentity channels = 2;
+}

amber/src/main/python/core/architecture/handlers/control/initialize_executor_handler.py

@@ -32,4 +32,7 @@ async def initialize_executor(self, req: InitializeExecutorRequest) -> EmptyRetu
         )
         # Loop-back write addresses; see the proto field doc on loopStartStateUris.
         self.context.loop_start_state_uris = dict(req.loop_start_state_uris)
+        # Frame membership decides the failure path: drain (guarded) vs the
+        # default report-and-pause (unguarded).
+        self.context.guarded = req.guarded
         return EmptyReturn()

amber/src/main/python/core/architecture/managers/context.py

@@ -90,6 +90,11 @@ def __init__(self, worker_id, input_queue):
         # Loop-back write addresses delivered at setup; see the proto field doc
         # on InitializeExecutorRequest.loopStartStateUris (controlcommands.proto).
         self.loop_start_state_uris: Dict[str, str] = {}
+        # True iff this operator sits inside a try/catch frame's cone (from
+        # InitializeExecutorRequest.guarded, set by TryCatchFramePass).
+        # Guarded: own executor failure becomes an in-band error State and the
+        # worker drains. Unguarded: default behavior — report and pause.
+        self.guarded: bool = False
 
     def report_exception(self, err: BaseException) -> None:
         """Route an operator-facing exception to the exception manager and

amber/src/main/python/core/architecture/packaging/output_manager.py

@@ -41,6 +41,7 @@
 from core.architecture.sendsemantics.round_robin_partitioner import (
     RoundRobinPartitioner,
 )
+from core.architecture.sendsemantics.signal_partitioner import SignalPartitioner
 from core.models import Tuple, Schema, StateFrame
 from core.models.payload import DataPayload, DataFrame
 from core.models.state import State
@@ -66,6 +67,7 @@
     RoundRobinPartitioning,
     RangeBasedShufflePartitioning,
     BroadcastPartitioning,
+    SignalPartitioning,
 )
 
 
@@ -81,6 +83,7 @@ def __init__(self, worker_id: str):
             HashBasedShufflePartitioning: HashBasedShufflePartitioner,
             RangeBasedShufflePartitioning: RangeBasedShufflePartitioner,
             BroadcastPartitioning: BroadcastPartitioner,
+            SignalPartitioning: SignalPartitioner,
         }
         self._ports: typing.Dict[PortIdentity, WorkerPort] = dict()
         self._channels: typing.Dict[ChannelIdentity, Channel] = dict()

amber/src/main/python/core/architecture/sendsemantics/signal_partitioner.py

@@ -0,0 +1,75 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import typing
+from overrides import overrides
+from typing import Iterator
+
+from core.architecture.sendsemantics.partitioner import Partitioner
+from core.models import Tuple
+from core.models.state import State
+from core.util import set_one_of
+from proto.org.apache.texera.amber.core import ActorVirtualIdentity
+from proto.org.apache.texera.amber.engine.architecture.rpc import EmbeddedControlMessage
+from proto.org.apache.texera.amber.engine.architecture.sendsemantics import (
+    Partitioning,
+    SignalPartitioning,
+)
+
+
+class SignalPartitioner(Partitioner):
+    """Partitioner for signal links (try/catch frame wiring).
+
+    Data tuples are dropped at the sender -- never batched, serialized, or
+    networked -- while States and ECMs still travel, because a frame's gate
+    only needs the error signal and the end-of-stream marker. Mirror of the
+    Scala `SignalPartitioner`.
+    """
+
+    def __init__(self, partitioning: SignalPartitioning):
+        super().__init__(set_one_of(Partitioning, partitioning))
+        self.receivers = list(
+            {channel.to_worker_id for channel in partitioning.channels}
+        )
+
+    @overrides
+    def add_tuple_to_batch(
+        self, tuple_: Tuple
+    ) -> Iterator[typing.Tuple[ActorVirtualIdentity, typing.List[Tuple]]]:
+        # tuples never traverse a signal link
+        return iter(())
+
+    @overrides
+    def flush(
+        self, to: ActorVirtualIdentity, ecm: EmbeddedControlMessage
+    ) -> Iterator[typing.Union[EmbeddedControlMessage, typing.List[Tuple]]]:
+        for receiver in self.receivers:
+            if receiver == to:
+                yield ecm
+
+    @overrides
+    def flush_state(
+        self, state: State
+    ) -> Iterator[
+        typing.Tuple[ActorVirtualIdentity, typing.Union[State, typing.List[Tuple]]]
+    ]:
+        for receiver in self.receivers:
+            yield receiver, state
+
+    @overrides
+    def reset(self) -> None:
+        pass

amber/src/main/python/core/models/state.py

@@ -24,6 +24,14 @@
 
 
 class State(dict):
+    # Reserved key marking a State as an in-band error signal (an operator
+    # failure traveling forward as a dataflow event). Mirror of the Scala
+    # `State.ErrorKey` (core/state/State.scala): the value is an envelope with
+    # operatorId / workerId / errorType / message. Recognition and port
+    # poisoning live in the worker runtime (MainLoop), not in operators.
+    ERROR_KEY = "__error__"
+    _ERR_OPERATOR_ID = "operatorId"
+
     CONTENT = "content"
     # Loop-control bookkeeping owned by the worker runtime, NOT user state -- it
     # never appears in the content JSON. In memory it rides on the StateFrame
@@ -78,6 +86,31 @@ def from_json(cls, payload: str) -> "State":
     def from_tuple(cls, row: Tuple) -> "State":
         return cls.from_json(row[cls.CONTENT])
 
+    @classmethod
+    def error(cls, operator_id: str, worker_id: str, err: BaseException) -> "State":
+        """Build the in-band error signal for a failed operator."""
+        return cls(
+            {
+                cls.ERROR_KEY: {
+                    cls._ERR_OPERATOR_ID: operator_id,
+                    "workerId": worker_id,
+                    "errorType": type(err).__name__,
+                    "message": str(err),
+                }
+            }
+        )
+
+    def is_error(self) -> bool:
+        return State.ERROR_KEY in self
+
+    def error_operator_id(self) -> Any:
+        """The failing operator's canonical physical id, or None if not an
+        error State. Used by try/catch frames for own-cone attribution."""
+        envelope = self.get(State.ERROR_KEY)
+        if isinstance(envelope, dict):
+            return envelope.get(State._ERR_OPERATOR_ID)
+        return None
+
 
 _TYPE_MARKER = "__texera_type__"
 _PAYLOAD_MARKER = "payload"

amber/src/main/python/core/runnables/data_processor.py

@@ -65,6 +65,15 @@ def run(self) -> None:
             else:
                 self.process_tuple()
 
+    # NOTE on the failure path: `_executor_session` sets `finished_current`
+    # in its except branch, BEFORE the final context switch. The loop above
+    # re-checks the input slots only after MainLoop queued the next input and
+    # notified -- which is exactly the invariant the assertion enforces. If
+    # the cycle were marked finished only after that switch (on this thread's
+    # next wake-up), MainLoop's cycle loop would see an unfinished cycle,
+    # wake this thread once more WITHOUT queuing anything, the assertion
+    # would kill this thread, and MainLoop's next switch would wait forever.
+
     def process_internal_marker(self, internal_marker: InternalMarker) -> None:
         with self._executor_session() as (executor, port_id):
             if isinstance(internal_marker, StartChannel):
@@ -102,8 +111,18 @@ def _executor_session(self):
         and queue the stack trace as a console message, and always switch
         back to MainLoop on exit. Reporting must happen *before* the
         switch: MainLoop's post-switch hook flushes console messages and
-        then enters EXCEPTION_PAUSE, so anything queued after the switch
-        would arrive at the coordinator only after the worker resumes.
+        broadcasts the error State, so anything queued after the switch
+        would reach the coordinator late.
+
+        On an exception, `_set_output_tuple` never ran, so the cycle is
+        finished HERE, before the final switch — the same ordering the
+        normal path has (`_set_output_tuple` sets `finished_current` before
+        its last switch). MainLoop's cycle loop then ends the cycle on its
+        first check, exactly like a normal cycle that yielded its trailing
+        None, and this thread parks in the switch until the next input is
+        queued. (For unguarded workers the failure ping-pong is parked by
+        EXCEPTION_PAUSE instead; guarded workers drain — MainLoop's drain
+        guard skips the executor for the rest of the poisoned port.)
         """
         try:
             executor = self._context.executor_manager.executor
@@ -115,6 +134,12 @@ def _executor_session(self):
                 yield executor, port_id
         except Exception as err:
             self._context.report_exception(err)
+            if self._context.guarded:
+                # Drain semantics: end the cycle now. Unguarded workers leave
+                # the cycle OPEN instead — MainLoop pauses on its side of the
+                # switch and the current input stays retriable
+                # (RetryCurrentTuple), exactly the pause-and-retry flow.
+                self._context.tuple_processing_manager.finished_current.set()
         finally:
             self._switch_context()