[Brook] Named the last completed and failed tasks on a local activity start.

crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto

@@ -77,6 +77,12 @@ message WorkflowTaskAttempt {
     uint32 attempt = 2;
     // When the server started the task.
     google.protobuf.Timestamp started_time = 3;
+    // The started event id of the run's last completed workflow task, from the poll response.
+    // 0 when no task of the run has completed.
+    int64 previous_started_event_id = 4;
+    // The started event ids of the run's failed and timed out workflow tasks in history, in event
+    // order. It holds the last 100 only, to keep each start small.
+    repeated int64 failed_tasks = 5;
 }
 
 // Attempt to cancel a running activity

crates/sdk-core/src/core_tests/local_activity_workflow_task.rs

@@ -111,6 +111,8 @@ async fn a_local_activity_names_the_live_task_that_dispatched_it() {
             started_event_id: 8,
             attempt: 1,
             started_time: Some(started_time.into()),
+            previous_started_event_id: 3,
+            failed_tasks: vec![],
         })
     );
     finish(worker, la).await;
@@ -135,6 +137,10 @@ async fn a_rerun_after_a_timed_out_task_names_the_later_task() {
     // History has no marker for it, so it runs again, dispatched by the live task.
     let la = worker.poll_activity_task().await.unwrap();
     assert_eq!(dispatching(start_of(&la)), (9, 1));
+    // The start also names the last completed task and the task that timed out.
+    let task = start_of(&la).workflow_task.as_ref().unwrap();
+    assert_eq!(task.previous_started_event_id, 3);
+    assert_eq!(task.failed_tasks, [6]);
     finish(worker, la).await;
 }
 

crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs

@@ -78,6 +78,10 @@ use temporalio_common::{
 type Result<T, E = WFMachinesError> = std::result::Result<T, E>;
 
 slotmap::new_key_type! { struct MachineKey; }
+
+/// The most failed workflow tasks that a local activity start names.
+const MAX_FAILED_WFTS: usize = 100;
+
 /// Handles all the logic for driving a workflow. It orchestrates many state machines that together
 /// comprise the logic of an executing workflow. One instance will exist per currently executing
 /// (or cached) workflow on the worker.
@@ -86,6 +90,9 @@ pub(crate) struct WorkflowMachines {
     /// kept because the lang side polls & completes for every workflow task, but we do not need
     /// to poll the server that often during replay.
     last_history_from_server: HistoryUpdate,
+    /// The started event ids of the run's failed and timed out workflow tasks, for the local
+    /// activities that need to know which tasks never completed.
+    failed_wft_started_ids: VecDeque<i64>,
     /// Protocol messages that have yet to be processed for the current WFT.
     protocol_msgs: Vec<IncomingProtocolMessage>,
     /// EventId of the last handled WorkflowTaskStarted event
@@ -276,6 +283,7 @@ impl WorkflowMachines {
         };
         Self {
             last_history_from_server: basics.history,
+            failed_wft_started_ids: VecDeque::new(),
             protocol_msgs: vec![],
             workflow_id: basics.workflow_id,
             workflow_type: basics.workflow_type,
@@ -345,6 +353,7 @@ impl WorkflowMachines {
     }
 
     pub(crate) fn new_history_from_server(&mut self, update: HistoryUpdate) -> Result<()> {
+        self.note_failed_wfts(update.get_events());
         self.last_history_from_server = update;
         self.replaying = self.preparing_legacy_query
             || self.last_history_from_server.previous_wft_started_id > 0;
@@ -417,6 +426,38 @@ impl WorkflowMachines {
         self.current_started_event_id
     }
 
+    pub(crate) fn failed_wft_started_ids(&self) -> Vec<i64> {
+        self.failed_wft_started_ids.iter().copied().collect()
+    }
+
+    /// Only the last ones are kept, which bounds the list on a long run. A local activity's task
+    /// facts only need the failed tasks since its run's last write to a Stream.
+    fn note_failed_wfts(&mut self, events: &[HistoryEvent]) {
+        for event in events {
+            let started = match &event.attributes {
+                Some(history_event::Attributes::WorkflowTaskFailedEventAttributes(a)) => {
+                    a.started_event_id
+                }
+                Some(history_event::Attributes::WorkflowTaskTimedOutEventAttributes(a)) => {
+                    a.started_event_id
+                }
+                _ => continue,
+            };
+            // A page of history may come again after a cache miss.
+            if self
+                .failed_wft_started_ids
+                .back()
+                .is_some_and(|&last| last >= started)
+            {
+                continue;
+            }
+            if self.failed_wft_started_ids.len() == MAX_FAILED_WFTS {
+                self.failed_wft_started_ids.pop_front();
+            }
+            self.failed_wft_started_ids.push_back(started);
+        }
+    }
+
     pub(crate) fn history_size_bytes(&self) -> u64 {
         self.history_size_bytes
     }

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

@@ -197,6 +197,7 @@ impl ManagedRun {
             attempt: work.attempt,
             started_event_id: work.update.wft_started_id,
             started_time: work.started_time,
+            previous_started_event_id: work.update.previous_wft_started_id,
             task_token: work.task_token,
             wf_id: work.execution.workflow_id.clone(),
         };
@@ -1292,10 +1293,12 @@ impl ManagedRun {
             started_event_id: wft.info.started_event_id,
             attempt: wft.info.attempt,
             started_time: wft.info.started_time.map(Into::into),
+            previous_started_event_id: wft.info.previous_started_event_id,
+            failed_tasks: self.wfm.machines.failed_wft_started_ids(),
         };
         for request in requests {
             if let LocalActRequest::New(new_la) = request {
-                new_la.workflow_task = Some(task);
+                new_la.workflow_task = Some(task.clone());
             }
         }
     }

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

@@ -1095,6 +1095,8 @@ struct WorkflowTaskInfo {
     /// From the poll response, like the attempt.
     started_event_id: i64,
     started_time: Option<SystemTime>,
+    /// The started event id of the run's last completed task, from the poll response.
+    previous_started_event_id: i64,
     /// Exists to allow easy tagging of spans with workflow ids. Is duplicative of info inside the
     /// run machines themselves, but that can't be accessed easily. Would be nice to somehow have a
     /// shared repository, or refcounts, or whatever, for strings like these that get duped all