Read the agent status through its Arc instead of cloning the record

golem-worker-executor/src/durable_host/call_coordinator.rs

@@ -505,7 +505,7 @@ where
                 ctx.state.card_event_boundary_scan =
                     Some(crate::durable_host::CardEventBoundaryScan::new(
                         status.oplog_idx,
-                        status.pending_card_events,
+                        status.pending_card_events.clone(),
                     ));
             }
         }

golem-worker-executor/src/durable_host/mod.rs

@@ -1788,7 +1788,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
             .get_non_detached_last_known_status()
             .await;
         let status_idx = status.oplog_idx;
-        let status_pending = status.pending_card_events;
+        let status_pending = status.pending_card_events.clone();
 
         let oplog = self.public_state.worker().oplog();
         let current_idx = oplog.current_oplog_index().await;

golem-worker-executor/src/worker/invocation_loop.rs

@@ -645,7 +645,8 @@ impl<Ctx: WorkerCtx> InvocationLoop<Ctx> {
                                             .parent
                                             .get_non_detached_last_known_status()
                                             .await
-                                            .current_idempotency_key;
+                                            .current_idempotency_key
+                                            .clone();
                                         match kind {
                                             InterruptKind::Suspend(_) => {
                                                 self.parent.add_and_commit_oplog(OplogEntry::suspend()).await;

golem-worker-executor/src/worker/mod.rs

@@ -1632,7 +1632,7 @@ impl<Ctx: WorkerCtx> Worker<Ctx> {
     // Outside of reverts and updates, this will return the same status as get_latest_worker_metadata.
     // This just has an additional assert built in for when decisions need to be sure that they are fully up to date on the oplog.
     // _NEVER_ call this from outside the invocation loop, as that is the only place that can reason about whether the status is detached or not.
-    pub async fn get_non_detached_last_known_status(&self) -> AgentStatusRecord {
+    pub async fn get_non_detached_last_known_status(&self) -> Arc<AgentStatusRecord> {
         // Runs on the worker-state actor's status queue so the detached flag and the published
         // status are observed consistently with any in-flight commit/reattach transaction.
         self.state_actor.non_detached_status().await
@@ -2428,7 +2428,7 @@ impl<Ctx: WorkerCtx> Worker<Ctx> {
 
     // should only be called from invocation loop
     pub async fn store_invocation_failure(&self, key: &IdempotencyKey, trap_type: &TrapType) {
-        let status = self.last_known_status.load_full().as_ref().clone();
+        let status = self.last_known_status.load_full();
         let keys_to_fail =
             invocation_keys_to_fail(&status, Some(key), !trap_type.is_invocation_rejection());
         let stderr = self.worker_event_service.get_last_invocation_errors();
@@ -5316,7 +5316,12 @@ impl<Ctx: WorkerCtx> Worker<Ctx> {
     }
 
     pub async fn lookup_invocation_result(&self, key: &IdempotencyKey) -> LookupResult {
-        let status = self.last_known_status.load_full().as_ref().clone();
+        // Kept as an `Arc` rather than cloned out of. The record owns
+        // `invocation_results`, which gains an entry per invocation and is never
+        // pruned, so deep-copying it to read one key made each lookup cost more
+        // than the last. `load_full` already gives a consistent snapshot with the
+        // lifetime this needs.
+        let status = self.last_known_status.load_full();
         let maybe_result = self
             .invocation_results
             .read()
@@ -5709,7 +5714,7 @@ impl<Ctx: WorkerCtx> Worker<Ctx> {
             }
         }
 
-        let status = self.last_known_status.load_full().as_ref().clone();
+        let status = self.last_known_status.load_full();
         let keys_to_fail = invocation_keys_to_fail(&status, None, true);
 
         let mut invocation_results = self.invocation_results.write().await;
@@ -6170,7 +6175,7 @@ impl<Ctx: WorkerCtx> Worker<Ctx> {
         if self.last_known_status_detached.load(Ordering::Acquire) {
             return;
         }
-        let status = self.last_known_status.load_full().as_ref().clone();
+        let status = self.last_known_status.load_full();
         self.status_checkpointer
             .maybe_checkpoint(&status, reason)
             .await;
@@ -6193,7 +6198,7 @@ impl<Ctx: WorkerCtx> Worker<Ctx> {
         if self.last_known_status_detached.load(Ordering::Acquire) {
             return;
         }
-        let status = self.last_known_status.load_full().as_ref().clone();
+        let status = self.last_known_status.load_full();
         if let Some(marker) = min_exposed_marker
             && status.oplog_idx > marker
         {

golem-worker-executor/src/worker/state_actor.rs

@@ -129,7 +129,7 @@ enum StatusJob {
     /// but the caller decides how to react (it asserts): a job whose caller was cancelled must
     /// not be able to panic the actor.
     NonDetachedStatus {
-        done: oneshot::Sender<Option<AgentStatusRecord>>,
+        done: oneshot::Sender<Option<Arc<AgentStatusRecord>>>,
     },
     /// Commits, then — if the status became detached (a jump or revert made it non-foldable) —
     /// recomputes it from the oplog, republishes it, and forces a cache flush.
@@ -266,7 +266,7 @@ impl<Ctx: WorkerCtx> WorkerStateActor<Ctx> {
                         let status = if state.detached.load(Ordering::Acquire) {
                             None
                         } else {
-                            Some(state.last_known_status.load_full().as_ref().clone())
+                            Some(state.last_known_status.load_full())
                         };
                         let _ = done.send(status);
                     }
@@ -389,7 +389,7 @@ impl<Ctx: WorkerCtx> WorkerStateActor<Ctx> {
     /// Returns the published status, asserting it is attached to the oplog. Serialized behind
     /// any in-flight commit/reattach transactions. The assert lives here on the caller side, so
     /// a job left behind by a cancelled caller cannot panic the actor.
-    pub async fn non_detached_status(&self) -> AgentStatusRecord {
+    pub async fn non_detached_status(&self) -> Arc<AgentStatusRecord> {
         self.commit
             .run_status_job(|done| StatusJob::NonDetachedStatus { done })
             .await