fix: format notification batching changes

crates/worker/src/controller.rs

@@ -368,11 +368,7 @@ fn stage_pending_notifications<St: Store + Clone>(
     }
     let count = notifications.len();
     for (notification, extension) in notifications {
-        notify_buffer.push_durable_notify(
+        notify_buffer.push_durable_notify(notification.message, notification.provenance, extension);
-            notification.message,
-            notification.provenance,
-            extension,
-        );
     }
     count
 }
@@ -409,7 +405,8 @@ fn prepare_pending_run<St: Store + Clone>(
         }
         None if allow_notification_run
             && (stage_pending_notifications(pending_submissions, notify_buffer) > 0
-                || notify_buffer.has_notification_pending()) => {
+                || notify_buffer.has_notification_pending()) =>
+        {
             Some(PendingRun::RunForNotification {
                 invoke_kind: protocol::InvokeKind::Notify,
             })
@@ -3149,7 +3146,6 @@ mod tests {
             Some(PendingRun::Submit(_))
         ));
         assert_eq!(notify_buffer.len(), 1);
-
     }
 
     #[test]

crates/worker/src/worker.rs

@@ -156,11 +156,13 @@ where
         Many(VecDeque<PendingNotification>),
     }
 
-    Ok(match <Option<OneOrMany> as serde::Deserialize>::deserialize(deserializer)? {
+    Ok(
-        Some(OneOrMany::One(notification)) => VecDeque::from([notification]),
+        match <Option<OneOrMany> as serde::Deserialize>::deserialize(deserializer)? {
-        Some(OneOrMany::Many(notifications)) => notifications,
+            Some(OneOrMany::One(notification)) => VecDeque::from([notification]),
-        None => VecDeque::new(),
+            Some(OneOrMany::Many(notifications)) => notifications,
-    })
+            None => VecDeque::new(),
+        },
+    )
 }
 
 impl PendingActivationState {
@@ -10239,9 +10241,11 @@ mod build_summary_prompt_tests {
             handle.accept_notification("notification-1".into(), "different".into()),
             Err(PendingSubmissionError::IdempotencyConflict)
         ));
-        assert!(handle
+        assert!(
-            .accept_notification("notification-2".into(), "second notice".into())
+            handle
-            .unwrap());
+                .accept_notification("notification-2".into(), "second notice".into())
+                .unwrap()
+        );
         handle
             .accept("request-1".into(), vec![Segment::text("submit")], false)
             .unwrap();

crates/worker/tests/controller_test.rs

@@ -1930,7 +1930,11 @@ async fn repeated_notify_while_idle_coalesces_and_auto_starts_one_turn() {
         .filter_map(|item| item.as_text())
         .filter(|text| text.contains("[Notification]"))
         .collect::<Vec<_>>();
-    assert_eq!(notifications.len(), 2, "duplicate receipt must not duplicate content");
+    assert_eq!(
+        notifications.len(),
+        2,
+        "duplicate receipt must not duplicate content"
+    );
     assert!(notifications[0].contains("progress snapshot"));
     assert!(notifications[1].contains("second coalesced notification"));
 }
@@ -2684,9 +2688,8 @@ async fn pause_then_resume_preserves_notifications_and_history_consistency() {
     let requests = client_for_assert.captured_requests();
     assert_eq!(requests.len(), 2);
     assert!(requests[1].items.iter().any(|item| {
-        item.as_text().is_some_and(|text| {
+        item.as_text()
-            text.contains("[Notification]") && text.contains("resume context")
+            .is_some_and(|text| text.contains("[Notification]") && text.contains("resume context"))
-        })
     }));
 
     // History consistency: the interrupted partial response is absent, while