[Feature]: Expose whether the operation is just started to `execute` method

[zhongkechen]

What would you like?

Currently check_result_status always return the checkpointed result after STARTED is checkpointed. This hides the information of whether this operation is the first time started at the execution phase in execute method. This information is required in some cases. For example, when calling plugin interface, we want to know if a CONTEXT operation is calling the user function for the first time.

Possible Implementation

refactor the base OperationExecutor interface to carry more info

@dataclass(frozen=True)
class CheckResult(Generic[T]):
    is_ready_to_execute: bool
    has_checkpointed_result: bool
    checkpointed_result: CheckpointedResult | None = None
    deserialized_result: T | None = None

class OperationExecutor(ABC, Generic[T]):
    @abstractmethod
    def check_result_status(self) -> CheckResult[T]:
        ...

    @abstractmethod
    def execute(self, checkpointed_result: CheckpointedResult) -> T:
        ...

    def process(self) -> T:
        result = self.check_result_status()
        if not result.is_ready_to_execute and not result.has_checkpointed_result:
            result = self.check_result_status()
        if result.has_checkpointed_result:
            return result.deserialized_result
        if result.is_ready_to_execute:
            if result.checkpointed_result is None:
                msg = "CheckResult is marked ready to execute but checkpointed result is not set."
                raise InvalidStateError(msg)
            return self.execute(result.checkpointed_result)   # <<--- this is currently always called with checkpointed result after STARTED

We have a few choices

1. choose the JS SDK style - each operation executor (handler) handles its own logic without sharing anything with others. This means execute and check_result_status methods are internal to each operation. The operation executors don't have to implement the same interface.
2. choose the Java SDK style - operations share the common parts in BaseOperation while each operation implement its own start and replay methods. Using start and replay is more flexible than execute and check_result_status.
3. fix this with minimal changes required for plugin interface: patch ChildOperationExecutor in Python SDK to override process method and implement the required logic specifically for what the plugin interface requires

Is this a breaking change?

No

Does this require an RFC?

No

Additional Context

No response

[yaythomas]

2 is a nice idea and fits well, since execute / check_result_status is more or less already the start / replay shape familiar from Java SDK, with shared base, per-op methods. 😄 Python doesn't necessarily need the full plumbing for CompletableFuture-per-operation start/replay.

The thing Python doesn't have is the explicit first-time-vs-replay signal. Both JS and Java SDKs already have this enum

• Java ExecutionMode (REPLAY/EXECUTION)
• JS DurableExecutionMode (ExecutionMode/ReplayMode/ReplaySucceededContext).

The third value on JS covers replay-children explicitly with ReplaySucceededContext, which is the case we actually need for child contexts.

class OperationExecutionPhase(Enum):
    INITIAL = "INITIAL"           # first call; user fn runs, result is checkpointed
    REPLAY = "REPLAY"             # resumed STARTED; user fn runs, result is checkpointed
    RECONSTRUCT = "RECONSTRUCT"   # already SUCCEEDED; user fn runs for side-effects, no checkpoint

class CheckResult(Generic[T]):
    ...
    phase: OperationExecutionPhase  # populated in check_result_status, where we already know

    @classmethod
    def create_is_ready_to_execute(
        cls,
        checkpoint: CheckpointedResult,
        *,
        phase: OperationExecutionPhase,
    ) -> CheckResult[T]:
        return cls(
            is_ready_to_execute=True,
            has_checkpointed_result=False,
            checkpointed_result=checkpoint,
            phase=phase,
        )

OperationExecutor.execute() and process() thread it through.

class OperationExecutor(ABC, Generic[T]):
    @abstractmethod
    def execute(
        self,
        checkpointed_result: CheckpointedResult,
        *,
        phase: OperationExecutionPhase,
    ) -> T: ...

    def process(self) -> T:
        result = self.check_result_status()
        if not result.is_ready_to_execute and not result.has_checkpointed_result:
            result = self.check_result_status()

        if result.has_checkpointed_result:
            return result.deserialized_result  # type: ignore[return-value]

        if result.is_ready_to_execute:
            if result.checkpointed_result is None or result.phase is None:
                msg = "CheckResult ready-to-execute but checkpointed_result/phase missing"
                raise InvalidStateError(msg)
            return self.execute(result.checkpointed_result, phase=result.phase)

        raise InvalidStateError("Invalid CheckResult state")

And then each executor populates phase where the decision is made. ChildOperationExecutor is the motivating case.

def check_result_status(self) -> CheckResult[T]:
    checkpointed_result = self.state.get_checkpoint_result(
        self.operation_identifier.operation_id
    )

    # Terminal SUCCEEDED — execute not called, no phase needed
    if checkpointed_result.is_succeeded() and not checkpointed_result.is_replay_children():
        ...
        return CheckResult.create_completed(result)

    # SUCCEEDED with replay_children — re-execute on replay
    if checkpointed_result.is_succeeded() and checkpointed_result.is_replay_children():
        return CheckResult.create_is_ready_to_execute(
            checkpointed_result,
            phase=OperationExecutionPhase.RECONSTRUCT,
        )

    if checkpointed_result.is_failed():
        checkpointed_result.raise_callable_error()

    # Decide phase BEFORE creating the async START — is_existent reflects what was there on entry
    is_first_attempt = not checkpointed_result.is_existent()
    phase = (
        OperationExecutionPhase.INITIAL
        if is_first_attempt
        else OperationExecutionPhase.REPLAY
    )

    if is_first_attempt and not self.is_virtual:
        start_operation = OperationUpdate.create_context_start(...)
        self.state.create_checkpoint(operation_update=start_operation, is_sync=False)

    return CheckResult.create_is_ready_to_execute(checkpointed_result, phase=phase)


def execute(
    self,
    checkpointed_result: CheckpointedResult,
    *,
    phase: OperationExecutionPhase,
) -> T:
    ...
    raw_result: T = self.func()

    if self.is_virtual:
        return raw_result

    # Was: `if checkpointed_result.is_replay_children(): return raw_result`
    if phase is OperationExecutionPhase.RECONSTRUCT:
        return raw_result

    # ... rest unchanged

And then in #371 it becomes

# operation/child.py (inside execute)
wrapped_user_func = self.state.wrap_user_function(
    self.func,
    self.operation_identifier,
    phase=phase,  # plugin gets the explicit phase, no peeking at is_existent
)