722 feature exit gracefully with pending when a checkpoint response has no checkpointtoken

[hln33]

Issue #, if available:

Fixes #722

Description of changes:

This PR updates the SDK to return PENDING when a checkpoint response omits the next checkpoint token for non-terminal work, instead of treating that response as execution completion.

Changes include:

• Stop further checkpointing after token revocation and wake queued work through the existing SuspendExecution path.
• Ensure execution exits return bare PENDING when the token has been revoked.
• Preserve true terminal completion when the accepted checkpoint batch contains the execution terminal update.
• Add local test runner pause/resume support for withholding checkpoint tokens and deferring invocations while paused.
• Add unit and e2e coverage for revoked-token handling, queued work, refreshes, oversized results, and pause/resume scenarios.

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[yaythomas]

Thank you @hln33, this is a careful port of the JS fix, and the state_test.py coverage of the real checkpoint loop is excellent!

Thanks for the follow-up commit for the Codex findings. _invoke_execution no longer reads and saves outside the worker lane, a retried checkpoint no longer receives the withheld token, and the mid-step pause test now fails without the fix (I checked against main's core SDK).

Please could you rename the PR to the conventional commit format, e.g. fix(sdk): answer PENDING on revoked token (your first commit's subject)? The PR title becomes the squash commit message: https://github.com/aws/aws-durable-execution-sdk-python/blob/main/CONTRIBUTING.md#pull-request-title-and-commit-message-format

Maybe this behaviour also warrants a conformance requirement? If so, please could you note it in the description or open an issue in aws-durable-execution-conformance-tests?

[yaythomas]

suggestion: I think these two flags would be safer as one enum, held in a private field and changed only through methods on Execution.

1. Two bools allow four combinations, and only three are valid: not paused, paused with nothing owed, and paused with an invocation owed.
2. The fourth combination, "not paused, invocation owed", does happen. The _invoke_execution race you just fixed produced it. The new processor_test.py test also builds it: it sets paused = False after a paused checkpoint has set deferred_invocation = True.
3. One enum field cannot hold that combination.

Something like this, next to ExecutionStatus:

class PauseState(Enum):
    """Whether a test paused this execution, and whether resume must invoke."""

    NOT_PAUSED = "NOT_PAUSED"
    PAUSED = "PAUSED"
    # While paused, an invocation was cut off or a wake-up was held back.
    PAUSED_INVOCATION_DEFERRED = "PAUSED_INVOCATION_DEFERRED"

with these on Execution:

    @property
    def is_paused(self) -> bool:
        return self._pause_state is not PauseState.NOT_PAUSED

    def pause(self) -> None:
        if self._pause_state is PauseState.NOT_PAUSED:
            self._pause_state = PauseState.PAUSED

    def defer_invocation(self) -> None:
        if self._pause_state is PauseState.PAUSED:
            self._pause_state = PauseState.PAUSED_INVOCATION_DEFERRED

    def resume(self) -> bool:
        """Clear the pause. Return whether resume must start an invocation."""
        deferred: bool = self._pause_state is PauseState.PAUSED_INVOCATION_DEFERRED
        self._pause_state = PauseState.NOT_PAUSED
        return deferred

Why methods: code outside Execution assigns these flags on five lines, in CheckpointCore.apply, _begin_invocation, _set_paused and _resume_execution. With methods, the rules live in one class, the way start(), begin_new_invocation() and the complete_* methods already work. is_paused would also match is_complete, and to_json_dict would store one "PauseState" key instead of two.

Why PauseState: it matches InvocationState, the package's other internal state enum. In this package, ...Status names values that go out in API responses, like ExecutionStatus.

[yaythomas]

I think not execution.paused turns this check off for more invocations than the pause needs.

1. While the execution is paused, the emulator answers the running invocation's next checkpoint without a token. The SDK then stops and answers PENDING, often with nothing pending. That PENDING is correct, so it needs an exception.
2. paused also exempts an invocation that answers PENDING with nothing pending although none of its checkpoints was answered without a token. That is the SDK bug this check catches.
3. I tried that second case on this head. The emulator accepts the answer and counts no failed attempt, and resume_execution() starts no invocation, because deferred_invocation was never set. So the execution makes no progress until it times out, and nothing names the cause.

CheckpointCore.apply sets deferred_invocation when it withholds a token, and it's now the only code that sets it while an invocation runs. With not execution.deferred_invocation in place of not execution.paused, the second case is rejected and retried as on main, and the pause/resume tests still pass. With the enum suggestion, the condition would check for PauseState.PAUSED_INVOCATION_DEFERRED.

That also makes main's message accurate again. Please could you keep it, Cannot return PENDING status with no pending operations.? Two tests in executor_test.py matched on it before this PR, so users' tests may too.

[yaythomas]

This loop has no timeout, and two cases make it wait a long time:

1. pause_execution() returns only after the running invocation ends, and the SDK ends it at its next checkpoint. So a step body that runs for a minute keeps pause_execution() waiting for that minute.
2. The testing package accepts any core SDK from 1.0.0 (aws-durable-execution-sdk-python>=1.0.0). A core SDK without this fix raises OrphanedChildException instead of answering PENDING. I ran a two-step pause test against main's core SDK: pause_execution() returned only when the execution's 15 s timeout fired, and the execution ended timed out.

Maybe take a timeout, as wait_for_result(timeout=...) does, and raise DurableFunctionsTestError when it expires? It would also help to say in both docstrings that the pause takes effect at the running invocation's next checkpoint. When this releases, the testing package's core floor could move to the first core version with this fix.

[yaythomas]

suggestion: please could you make execution_arn a required parameter of pause_execution() and resume_execution(), and drop this field?

1. The runner's other per-execution methods take the ARN: wait_for_result, wait_for_callback and get_execution_history.
2. A test can pause a running execution only after run_async() returns, and run_async() returns the ARN. run() returns only after the execution has finished.
3. So any test that can pause already holds the ARN. The default saves one argument.

The default also adds state the test can't see:

• It targets whichever execution run_async() started last. A test that starts two executions pauses the second one.
• It keeps that ARN after the execution finishes. A later pause_execution() with no ARN then does nothing and raises nothing, because _set_paused returns early for a completed execution. The "No execution in progress" error fires only before the first run_async().

JS's pauseExecution() takes no argument, and that fits JS: its run() returns only a promise of the result, and pauseExecution() acts on the execution that run() is driving. Here, run_async() already hands the test the ARN.

Concretely:

    def pause_execution(self, execution_arn: str) -> None:
    def resume_execution(self, execution_arn: str) -> None:

The rest is deletions: _default_execution_arn, its assignment in run_async(), _require_execution_arn, and in both docstrings the "Defaults to..." sentence and the Raises: entry. Worth doing before this ships, because removing a default later breaks every caller that relies on it, experimental or not.

[yaythomas]

suggestion: the cloud runner has neither method, so a test that pauses and later runs against DurableFunctionCloudTestRunner fails with a generic AttributeError: 'DurableFunctionCloudTestRunner' object has no attribute 'pause_execution'. JS gives both runners the same interface, and its cloud runner raises pauseExecution() is not implemented for CloudDurableTestRunner (aws-durable-execution-sdk-js#931).

Please could you add both methods to DurableFunctionCloudTestRunner, raising NotImplementedError with a message along the lines of:

pause_execution() is not implemented for DurableFunctionCloudTestRunner
resume_execution() is not implemented for DurableFunctionCloudTestRunner

NotImplementedError because the cloud runner can gain real pause and resume later, through the API.

[yaythomas]

suggestion: I think a retry of a checkpoint that was answered without a token should fail like any other stale token, rather than be answered again.

1. A response without a token means the invocation may not checkpoint again. apply has also advanced the token sequence, so the call's token is now stale.
2. On this head, a retry without a client token already fails that check with InvalidParameterValueException("Invalid checkpoint token").
3. A retry with the same client token matches this record and is answered again instead. A botocore retry over HTTP takes that path: the core SDK leaves ClientToken out, the model marks it as an idempotency token, so botocore fills one in and resends the same one on retry.
4. So whether the retry is rejected depends on whether it carries a client token.

Wanted: a retry of a checkpoint that was answered without a token, with or without a ClientToken, raises InvalidParameterValueException with the message exactly:

Invalid checkpoint token

That is the emulator's existing stale-token message, so nothing new is needed on the error side. Skipping this record when the token is withheld does it: last_checkpoint stays as it was, so both retries fall through to the existing token check. The withheld token still never reaches the handler. I tried it on this head: the e2e pause tests pass, and only the two test_paused_checkpoint_retries_without_a_token_even_after_resume tests, in executor_pause_resume_test.py and checkpoint/processor_test.py, need to expect the rejection instead of a second token-less answer.

[yaythomas]

suggestion: what pause_execution() and resume_execution() raise for an unknown ARN depends on the store. With the in-memory store, this load() raises a bare KeyError: 'arn:unknown' (I ran both). The filesystem and SQLite stores' load() raise ResourceNotFoundException with Execution arn:unknown not found instead.

Wanted: both methods, with every store, raise ResourceNotFoundException with the message exactly:

Durable Execution does not exist

That is the message the service returns for a missing execution, for example from StopDurableExecution on an unknown ARN. Catching KeyError and ResourceNotFoundException around the load() here and in _resume_execution does it. I tried it: both calls raise that, and no existing test depends on the KeyError.

Related, not for this PR: the emulator's get_execution and the stores say Execution <arn> not found for the same case. Aligning those with the service's text could be a follow-up.

[yaythomas]

suggestion: a few tests would pin down the rules around a withheld token, including the exact error messages. The first two pass on this head. The last two fail on this head and pass with the suggestions on executor.py lines 1468 and 1838. They pass hatch fmt --check.

def test_checkpoint_with_the_withheld_invocations_token_is_rejected():
    executor, store, execution, token_0 = _make_executor_with_started_execution()
    execution.paused = True
    store.save(execution)
    executor.checkpoint_execution(
        execution_arn=execution.durable_execution_arn,
        checkpoint_token=token_0,
        updates=[_step_start_update("step-A")],
    )

    with pytest.raises(InvalidParameterValueException) as exc_info:
        executor.checkpoint_execution(
            execution_arn=execution.durable_execution_arn,
            checkpoint_token=token_0,
            updates=[_step_start_update("step-B")],
        )
    assert str(exc_info.value) == "Invalid checkpoint token"


def test_resume_with_nothing_held_back_starts_no_invocation():
    executor, _, execution, _ = _make_executor_with_started_execution()
    arn = execution.durable_execution_arn
    executor._set_invocation_gate(arn, InvocationState.PRE_INVOKE)  # noqa: SLF001

    executor.pause_execution(arn)
    executor.resume_execution(arn)

    executor._scheduler.call_later.assert_not_called()  # noqa: SLF001


def test_paused_pending_is_accepted_only_when_a_token_was_withheld():
    executor, store, execution, _ = _make_executor_with_started_execution()
    execution.paused = True
    store.save(execution)
    pending = DurableExecutionInvocationOutput(status=InvocationStatus.PENDING)

    with pytest.raises(InvalidParameterValueException) as exc_info:
        executor._validate_invocation_response_and_store(  # noqa: SLF001
            execution.durable_execution_arn, pending, execution, execution.seq_counter
        )
    assert str(exc_info.value) == (
        "Cannot return PENDING status with no pending operations."
    )

    execution.deferred_invocation = True
    executor._validate_invocation_response_and_store(  # noqa: SLF001
        execution.durable_execution_arn, pending, execution, execution.seq_counter
    )


def test_pause_and_resume_raise_for_an_unknown_execution():
    executor, _, _, _ = _make_executor_with_started_execution()

    for call in (executor.pause_execution, executor.resume_execution):
        with pytest.raises(ResourceNotFoundException) as exc_info:
            call("arn:unknown")
        assert str(exc_info.value) == "Durable Execution does not exist"

They need pytest, DurableExecutionInvocationOutput, InvocationStatus, InvalidParameterValueException and ResourceNotFoundException imported.

With all the suggestions applied, I ran the whole testing suite. These are the only existing tests that need a change:

• test_paused_checkpoint_retries_without_a_token_even_after_resume, here and in checkpoint/processor_test.py: the retry now raises InvalidParameterValueException("Invalid checkpoint token").
• test_checkpoint_while_paused_omits_token_but_registers_update: the response's operations list is []; the check that step-A was registered stays.
• executor_test.py, the two tests whose match= this PR changed: back to main's match="no pending operations".
• executor_test.py, test_should_handle_pending_status_when_operations_exist and test_should_retry_when_pending_response_has_no_operations: add mock_execution.deferred_invocation = False, for the same reason this PR added mock_execution.paused = False (an unset Mock() attribute is truthy).

[yaythomas]

suggestion: when the token is withheld, the response still carries every operation the handler hasn't seen, and apply advances handler_seen_seq over them.

1. A token-less response from the service carries an empty Operations list. That is what the SDK gets today when a batch completes the execution.
2. For a batch without an EXECUTION update, the SDK reaches the revoked-token branch and stops before it reads new_execution_state (state.py lines 1069-1071). So the handler never applies these operations.
3. The emulator still records them as delivered, because handler_seen_seq moved over them.

Wanted: when the token is withheld, new_execution_state.operations is [] and handler_seen_seq does not move.

[] if execution.paused else paginator.unseen_operations() does both, because the advance only runs over the returned list. I tried it: the e2e pause tests pass, and test_checkpoint_while_paused_omits_token_but_registers_update would then assert an empty list and keep its check that step-A was registered.

[yaythomas]

no-action note: when not paused, a checkpoint whose batch completes the execution still gets a token here. The service answers that batch without a token and with an empty Operations list, and the SDK treats such a response as a finished execution (state.py lines 1064-1068). So the emulator reaches that branch only while paused. Maybe worth an issue, so an ordinary completion exercises it too.

[github-actions]

Codex AI review · Finding arf_v1_guweetjrqz6qymb4qyevo4qc4h

[P1] Defer retryable errors until cleanup checks revocation

This re-raises inside the with block. If an asynchronous checkpoint returns without a token while the context managers unwind, revocation is recorded after this check; the post-cleanup override is skipped and Lambda retries with the revoked invocation token. Capture the exception, finish checkpoint-thread cleanup, then return bare PENDING if revocation occurred; otherwise re-raise. Add a test racing a retryable InvocationError with a tokenless async checkpoint.

[github-actions]

Codex AI review · Finding arf_v1_w75l3lpqi2d4aselqozh4pokc4

[P2] Apply accepted response state before suspending

The tokenless response accepted this batch and may contain its terminal operation state, but this exits before merging NewExecutionState or emitting plugin updates. After resume, instrumentation never receives on_operation_end, so OTel can omit the logical operation span and invocation-end state is stale. Merge the inline operations without pagination and emit updates before waking waiters with SuspendExecution; add plugin coverage for this path.

[github-actions]

Codex AI review · Finding arf_v1_wbuih3nwxewkn3oiqepguv2fp2

[P3] Validate execution existence before creating a worker

Syntax validation allows an unknown ARN into _registry.submit. The default memory store then exposes a raw KeyError while disk stores return a different exception, and the failed task leaves a worker lane registered for the nonexistent execution. Call get_execution() before submitting in both pause and resume, returning a consistent ResourceNotFoundException, and add unknown-ARN tests.

[github-actions]

Codex AI review

Found three issues: a cleanup-time revocation race, lost instrumentation state on tokenless checkpoints, and inconsistent unknown-ARN handling in pause/resume.

Reviewed commit 88dffdcbbd44c24219bc5f2cbde9581e44e1589b. Workflow run