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:
2026-10-05 17:43:11 +09:00
parent a00a0575f6
commit 3174142bd1
2 changed files with 962 additions and 112 deletions
+80
View File
@@ -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