fix(core): delay zero-cache eviction until LA resolutions drain

CHANGELOG.md

@@ -89,6 +89,8 @@ to docs, or any other relevant information.
 ### Fixed
 * Panics from update validators now reject the update instead of repeatedly failing workflow
   tasks.
+* Workers with `max_cached_workflows` set to 0 no longer stall when a local activity resolves while
+  the resolution for an earlier one is still being delivered.
 
 ## [0.6.0] - 2026-08-04
 ## [0.5.0]

crates/sdk-core/src/worker/workflow/workflow_stream.rs

@@ -343,7 +343,11 @@ impl WFStream {
         .collect();
         // Keeping the run until its LAs resolve lets their incremental activations share the
         // current WFT. The final completion will queue the zero-cache eviction as usual.
-        if has_zero_sized_cache && !rh.waiting_on_local_activities() {
+        // Jobs still queued for this WFT (ex: resolutions for LAs that finished while an earlier
+        // one was being delivered) count as not-yet-resolved too: they can schedule further LAs,
+        // and the commands they produce are only flushed by the completion that finally answers
+        // the WFT. Evicting first would strand those commands in the discarded machines.
+        if has_zero_sized_cache && !rh.waiting_on_local_activities() && !rh.more_pending_work() {
             acts.extend(self.request_eviction_of_lru_run().into_run_update_resp())
         }
         acts

crates/sdk-core/tests/integ_tests/workflow_tests/local_activities.rs

@@ -26,6 +26,7 @@ use temporalio_common::{
         coresdk::{
             ActivityTaskCompletion, AsJsonPayloadExt, FromJsonPayloadExt,
             activity_result::ActivityExecutionResult,
+            activity_task::activity_task as act_task,
             common::extract_local_activity_marker_data,
             workflow_activation::{
                 WorkflowActivation, WorkflowActivationJob, workflow_activation_job,
@@ -3015,6 +3016,146 @@ async fn local_activity_resolutions_are_delivered_incrementally() {
     handle.fetch_history_and_replay(&mut worker).await.unwrap();
 }
 
+/// With a zero-sized cache, a run keeps its workflow task until every local activity resolution has
+/// been delivered and the task answered. Resolutions queued while an earlier one is outstanding with
+/// lang may schedule further activities, and the markers for all of them belong on the completion
+/// that finally answers the task.
+#[tokio::test]
+async fn zero_cache_doesnt_evict_before_wft_is_answered() {
+    let wfid = "fake_wf_id";
+    let mut t = TestHistoryBuilder::default();
+    t.add_by_type(EventType::WorkflowExecutionStarted);
+    t.add_workflow_task_scheduled_and_started();
+
+    let reported_commands = Arc::new(SegQueue::new());
+    let recorder = reported_commands.clone();
+    let mut mock_cfg =
+        MockPollCfg::from_resp_batches(wfid, t, [ResponseType::AllHistory], mock_worker_client());
+    mock_cfg.enforce_correct_number_of_polls = false;
+    mock_cfg.completion_mock_fn = Some(Box::new(move |c| {
+        recorder.push(
+            c.commands
+                .iter()
+                .map(|cmd| cmd.command_type())
+                .collect::<Vec<_>>(),
+        );
+        Ok(Default::default())
+    }));
+    let mut mock = build_mock_pollers(mock_cfg);
+    mock.worker_cfg(|wc| wc.max_cached_workflows = 0);
+    let core = mock_worker(mock);
+
+    let la_cmd = |seq: u32, id: &str| {
+        schedule_local_activity_cmd(
+            seq,
+            id,
+            ProtoActivityCancellationType::TryCancel,
+            Duration::from_secs(60),
+        )
+    };
+
+    let task = core.poll_workflow_activation().await.unwrap();
+    core.complete_workflow_activation(WorkflowActivationCompletion::from_cmds(
+        task.run_id,
+        vec![la_cmd(1, "1"), la_cmd(2, "2")],
+    ))
+    .await
+    .unwrap();
+
+    // Which activity each task belongs to is matched on activity id, since the order the two are
+    // handed out in is not something this test should depend on.
+    let queued_las = [
+        core.poll_activity_task().await.unwrap(),
+        core.poll_activity_task().await.unwrap(),
+    ];
+    let token_for = |activity_id: &str| {
+        queued_las
+            .iter()
+            .find(|t| {
+                matches!(&t.variant, Some(act_task::Variant::Start(start))
+                    if start.activity_id == activity_id)
+            })
+            .unwrap_or_else(|| panic!("no local activity task for id {activity_id}"))
+            .task_token
+            .clone()
+    };
+
+    core.complete_activity_task(ActivityTaskCompletion {
+        task_token: token_for("1"),
+        result: Some(ActivityExecutionResult::ok(vec![1].into())),
+    })
+    .await
+    .unwrap();
+    let resolve_first = core.poll_workflow_activation().await.unwrap();
+    assert_matches!(
+        resolve_first.jobs.as_slice(),
+        [WorkflowActivationJob {
+            variant: Some(workflow_activation_job::Variant::ResolveActivity(resolution)),
+        }] => assert_eq!(resolution.seq, 1)
+    );
+
+    // Finishing the second activity while the first resolution is outstanding with lang is what
+    // queues it rather than delivering it, leaving no outstanding activities but an undelivered
+    // job once the completion below lands.
+    core.complete_activity_task(ActivityTaskCompletion {
+        task_token: token_for("2"),
+        result: Some(ActivityExecutionResult::ok(vec![2].into())),
+    })
+    .await
+    .unwrap();
+    core.complete_workflow_activation(WorkflowActivationCompletion::empty(resolve_first.run_id))
+        .await
+        .unwrap();
+
+    // Scheduling from the queued resolution keeps the task open with the first two markers still
+    // buffered, which is the point at which the run must survive to report them.
+    let resolve_second = core.poll_workflow_activation().await.unwrap();
+    assert_matches!(
+        resolve_second.jobs.as_slice(),
+        [WorkflowActivationJob {
+            variant: Some(workflow_activation_job::Variant::ResolveActivity(resolution)),
+        }] => assert_eq!(resolution.seq, 2)
+    );
+    core.complete_workflow_activation(WorkflowActivationCompletion::from_cmd(
+        resolve_second.run_id,
+        la_cmd(3, "3"),
+    ))
+    .await
+    .unwrap();
+
+    let third_la = core.poll_activity_task().await.unwrap();
+    core.complete_activity_task(ActivityTaskCompletion {
+        task_token: third_la.task_token,
+        result: Some(ActivityExecutionResult::ok(vec![3].into())),
+    })
+    .await
+    .unwrap();
+    let resolve_third = core.poll_workflow_activation().await.unwrap();
+    assert_matches!(
+        resolve_third.jobs.as_slice(),
+        [WorkflowActivationJob {
+            variant: Some(workflow_activation_job::Variant::ResolveActivity(resolution)),
+        }] => assert_eq!(resolution.seq, 3)
+    );
+    core.complete_execution(&resolve_third.run_id).await;
+
+    core.shutdown().await;
+
+    let mut all_reported = vec![];
+    while let Some(cmds) = reported_commands.pop() {
+        all_reported.push(cmds);
+    }
+    let markers_reported: usize = all_reported
+        .iter()
+        .flatten()
+        .filter(|c| **c == CommandType::RecordMarker)
+        .count();
+    assert_eq!(
+        markers_reported, 3,
+        "all three local activity markers should reach the server, got {all_reported:?}"
+    );
+}
+
 #[workflow]
 #[derive(Default)]
 struct OldBatchedLocalActivityHistoryWf;