Fresh replay marks already-processed events as non-replay when a no-command WFT is followed by an empty WFT

While chasing a flaky test on temporalio/sdk-python#1856 I ran into what I think is a core bug in how WFT sequences get chunked. If a workflow task applies some non-command events (e.g., an activity completion), produces no commands, and the server then schedules another WFT with nothing in between, a worker that replays the run from scratch treats the first WFT as a heartbeat and merges the two. The merged sequence is the final one, so replaying flips to false partway through it, and the jobs from the first WFT (the activity resolution) go out to lang with is_replaying: false even though that WFT already ran and completed.

The merge is the !saw_command && next_next_event == WorkflowTaskScheduled branch in find_end_index_of_next_wft_seq (https://github.com/temporalio/sdk-core/blob/0c23f7189898/crates/sdk-core/src/worker/workflow/history_update.rs#L747) and the flip is at https://github.com/temporalio/sdk-core/blob/0c23f7189898/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs#L681. Commands still come out right since the code is deterministic, so nothing fails loudly - it's just that anything gated on replay runs twice: workflow.logger prints again, is_replaying() is wrong, and in our case the Python SDK's replay-safe OTel span ended and got exported twice with the same deterministic span id.

Repro with Python SDK 1.33.0 (core 85b71d7), WorkflowEnvironment.start_time_skipping() and a worker with max_cached_workflows=0. The Java test server produces the shape whenever a query lands while a WFT is running, since it buffers the query and delivers it via a new WFT. fwiw the dev server answers those as direct query tasks (I checked, only two WFTs there), so on a real server I'd expect you need e.g., an LA heartbeat or force_create_new_workflow_task plus a cache miss to hit the same path - I haven't tried that variant.

@workflow.defn(sandboxed=False)
class W:
    def __init__(self) -> None:
        self._ready = False
        self._proceed = False

    @workflow.run
    async def run(self) -> str:
        await workflow.execute_activity(quick_act, start_to_close_timeout=timedelta(seconds=30))
        workflow.logger.info("ACTIVITY-COMPLETION-APPLIED replaying=%s", workflow.unsafe.is_replaying())
        time.sleep(1.2)  # widen the window so a query lands mid-WFT
        self._ready = True
        await workflow.wait_condition(lambda: self._proceed)
        return "done"

    @workflow.query
    def ready(self) -> bool:
        return self._ready

    @workflow.signal
    def proceed(self) -> None:
        self._proceed = True

The client just polls ready every 200ms until it comes back True. workflow.logger drops replay logs, so each line below is one non-replay application of the same completion:

ACTIVITY-COMPLETION-APPLIED replaying=False   <- WFT 11-13, correct
ACTIVITY-COMPLETION-APPLIED replaying=False   <- WFT 14-16, which only carried the buffered query

History around it:

 9 activity_task_started
10 activity_task_completed
11 workflow_task_scheduled
12 workflow_task_started
13 workflow_task_completed
14 workflow_task_scheduled
15 workflow_task_started
16 workflow_task_completed

Core debug for the WFT 14 poll - the server's numbers look right to me, the activation flag doesn't:

HistoryUpdate(previous_started_event_id: 12, started_id: 15, length: 15)
WorkflowActivation(is_replaying: true, jobs: InitializeWorkflow)
WorkflowActivation(is_replaying: false, jobs: ResolveActivity(1, Completed))
WorkflowActivation(is_replaying: true, jobs: QueryWorkflow)

For comparison the next WFT (previous 15 / started 19, with a real signal event in it) replays the same ResolveActivity with is_replaying: true.

My hunch is that jobs coming from events at or before previous_started_event_id shouldn't ride along in a non-replay activation even when the heartbeat merge kicks in, but you all know this code way better than I do. I worked around it on the sdk-python side by not querying during that WFT. Lmk if the full repro script and debug log would help, happy to attach them

Start in crates/sdk-core/src/worker/workflow/history_update.rs at the find_end_index_of_next_wft_seq heartbeat-merge branch, then inspect the replay-state flip in crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs. Reproduce with the Python SDK scenario using start_time_skipping(), max_cached_workflows=0, and a polling query. Done means previously processed activity-resolution jobs retain replay semantics across the merged WFT sequence, with a regression test covering the history shape shown.