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.
This commit is contained in:
@@ -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();
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user