fix(workspace-server): serialize explicit orchestrator backlog attention

T-703: snapshots establish baselines; only observed Running -> Idle rechecks backlog. Scope reservation and success/pending records to the exact Orchestrator; preserve payload and request ID across bounded retries and Restore races. Drop the launch lock before attention delivery.

Regression evidence: Barrier/parallel start, reconnect snapshots, retry payload identity, Restore/replacement; real Runtime adapter + Worker Controller + canonical Session fixture proves start/Restore send/receipt/commit counts and post-acceptance response-loss dedupe. Worker FsStore regression proves same-ID retry after restore versus legitimate new-ID same-body notification.

Validation: cargo check; cargo test -p yoi-workspace-server (564 lib + 11 bin); cargo test -p worker -- --test-threads=1 (703 lib + 96 integration); cargo fmt --all -- --check; git diff --check HEAD. Default parallel worker test reproduced existing pending_restore_reuses_persisted_session_and_recovers_unknown_attribution ScopeLock NotFound; isolated test and full serial suite pass.

crates/worker/src/worker.rs

@@ -13290,6 +13290,86 @@ permission = "write"
         assert_eq!(state.pending_notifications[1].provenance, account_b);
     }
 
+    #[test]
+    fn notification_lost_acceptance_retry_after_restore_commits_once_not_by_body() {
+        let temp = tempfile::tempdir().unwrap();
+        let handle = PendingSubmissionHandle::for_test(temp.path());
+        let source = "backend:orchestrator-attention";
+        let accept = |handle: &PendingSubmissionHandle<session_store::FsStore>, id: &str| {
+            handle
+                .accept_notification_from_source(
+                    id.into(),
+                    "Queued Tickets require attention".into(),
+                    source.into(),
+                    WorkerHistoryProvenance::BackendInstruction { operation_id: None },
+                )
+                .unwrap()
+        };
+        // The sender lost this acceptance response; the receipt is already durable.
+        assert!(accept(&handle, "attention-1"));
+        let restored_state: PendingActivationState = handle
+            .persisted_entries_for_test()
+            .iter()
+            .rev()
+            .find_map(|entry| match entry {
+                LogEntry::Extension {
+                    domain, payload, ..
+                } if domain == SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN => {
+                    Some(serde_json::from_value(payload.clone()).unwrap())
+                }
+                _ => None,
+            })
+            .unwrap();
+        assert_eq!(restored_state.notification_receipts.len(), 1);
+        let handle = PendingSubmissionHandle {
+            state: Arc::new(Mutex::new(restored_state)),
+            writer: handle.writer.clone(),
+        };
+        assert!(!accept(&handle, "attention-1"));
+        let commit_batch = |handle: &PendingSubmissionHandle<session_store::FsStore>| {
+            let batch = handle.prepare_notification_batch();
+            let count = batch.len();
+            for (notification, extension) in batch {
+                handle
+                    .writer
+                    .commit_system_item_with_extensions(
+                        SystemItem::Notification {
+                            body: notification.message.clone(),
+                            message: notification.message,
+                            prompt_provenance: None,
+                        },
+                        vec![extension],
+                        Some(notification.provenance),
+                    )
+                    .unwrap();
+            }
+            handle.finish_notification_batch();
+            count
+        };
+        assert_eq!(commit_batch(&handle), 1);
+        assert!(!accept(&handle, "attention-1"));
+        assert_eq!(commit_batch(&handle), 0);
+        let notification_count = |handle: &PendingSubmissionHandle<session_store::FsStore>| {
+            handle
+                .persisted_entries_for_test()
+                .iter()
+                .filter(|entry| {
+                    matches!(entry,
+                        LogEntry::AnnotatedSystemItem { entry, .. }
+                            if matches!(entry.item, SystemItem::Notification { .. })
+                    )
+                })
+                .count()
+        };
+        assert_eq!(notification_count(&handle), 1);
+        assert_eq!(handle.state.lock().unwrap().notification_receipts.len(), 1);
+        // Same body with a new logical ID is legitimate, not content-deduped.
+        assert!(accept(&handle, "attention-2"));
+        assert_eq!(commit_batch(&handle), 1);
+        assert_eq!(notification_count(&handle), 2);
+        assert_eq!(handle.state.lock().unwrap().notification_receipts.len(), 2);
+    }
+
     #[test]
     fn notification_batch_preserves_fifo_order_and_notification_dedupes() {
         let temp = tempfile::tempdir().unwrap();