Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

feat: merge develop into hare/develop

Browse Source
This commit is contained in:
Hare
2026-09-18 21:36:04 +09:00
parent d7917512ac fba12b89bc
commit b06e2860ca
202 changed files with 18312 additions and 3945 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/Cargo.toml
+1 -1
View File
@@ -28,7 +28,7 @@ toml = { workspace = true }
tracing = { workspace = true }
tools = { workspace = true }
workdir = { workspace = true }
workspace-api = { workspace = true }
server-api = { workspace = true }
minijinja = "2.19.0"
chrono = "0.4"
config-source = { path = "../config-source" }
crates/worker/src/controller.rs
+509 -210
View File
File diff suppressed because it is too large Load Diff
crates/worker/src/discovery.rs
+6 -15
View File
@@ -389,7 +389,7 @@ where
reason: detail.live.error,
};
}
match send_notify(&detail.live.socket_path, message, false).await {
match send_notify(&detail.live.socket_path, message).await {
Ok(()) => WeakNotifyDelivery::Delivered,
Err(error) => WeakNotifyDelivery::SendFailed {
error: error.to_string(),
@@ -1009,17 +1009,16 @@ where
}
async fn send_peer_notify(socket_path: &Path, message: String) -> io::Result<()> {
send_notify(socket_path, message, true).await
send_notify(socket_path, message).await
}
async fn send_notify(socket_path: &Path, message: String, auto_run: bool) -> io::Result<()> {
async fn send_notify(socket_path: &Path, message: String) -> io::Result<()> {
let notification_request_id = protocol::new_submission_request_id();
connect_and_send(
socket_path,
&Method::NotifyTracked {
notification_request_id: notification_request_id.clone(),
message,
auto_run,
source: protocol::AuthenticatedInputSource::Backend {
operation_id: notification_request_id,
},
@@ -1552,11 +1551,7 @@ mod tests {
.await
.unwrap();
let method = reader.next::<Method>().await.unwrap().unwrap();
if let Method::NotifyTracked {
message, auto_run, ..
} = method
{
assert!(auto_run);
if let Method::NotifyTracked { message, .. } = method {
tx.send(message).await.unwrap();
} else {
panic!("expected Notify, got {method:?}");
@@ -1575,7 +1570,7 @@ mod tests {
}
#[tokio::test(flavor = "current_thread")]
async fn weak_notify_to_live_peer_uses_notify_without_auto_run_and_noops_when_missing() {
async fn weak_notify_to_live_peer_uses_realtime_notify_and_noops_when_missing() {
let root = TempDir::new().unwrap();
let store_dir = root.path().join("store");
let runtime_base = root.path().join("runtime");
@@ -1674,11 +1669,7 @@ mod tests {
.await
.unwrap();
let method = reader.next::<Method>().await.unwrap().unwrap();
if let Method::NotifyTracked {
message, auto_run, ..
} = method
{
assert!(!auto_run);
if let Method::NotifyTracked { message, .. } = method {
tx.send(message).await.unwrap();
} else {
panic!("expected Notify, got {method:?}");
crates/worker/src/feature/builtin/manage_workdir.rs
+1 -1
View File
@@ -23,7 +23,7 @@ use workdir::{
WorkdirSessionCapabilities, WorkdirSessionHandle, WriteRequest, WriteResult,
};
use workspace_api::{
use server_api::{
WorkingDirectoryCreateRequest as WorkdirCreateRequest,
WorkingDirectoryCreateResponse as WorkdirCreateResponse,
WorkingDirectoryListResponse as WorkdirListResponse,
crates/worker/src/feature/builtin/workspace_worker_discovery.rs
+1 -1
View File
@@ -168,7 +168,7 @@ fn input_schema() -> serde_json::Value {
mod tests {
use std::sync::Mutex;
use workspace_api::{
use server_api::{
WorkspaceWorkerDiscoveryItem, WorkspaceWorkerDiscoveryPage, WorkspaceWorkerSubject,
};
crates/worker/src/ipc/interceptor.rs
+5 -5
View File
@@ -1543,7 +1543,7 @@ mod tests {
PromptCatalog::from_projection(projection).unwrap(),
));
buffer.push_notify("updated".to_string(), false);
buffer.push_notify("updated".to_string());
let appends = interceptor
.pending_history_appends(PendingHistoryAppendsContext {
invocation: Default::default(),
@@ -1598,7 +1598,7 @@ mod tests {
prompts.store(Arc::new(
PromptCatalog::from_projection(projection).unwrap(),
));
buffer.push_notify("must persist".to_string(), false);
buffer.push_notify("must persist".to_string());
let error = interceptor
.pending_history_appends(PendingHistoryAppendsContext {
@@ -1621,8 +1621,8 @@ mod tests {
async fn pending_history_appends_drains_buffer_into_items() {
let registry = Arc::new(HookRegistryBuilder::new().build());
let buffer = NotifyBuffer::new();
buffer.push_notify("first".into(), false);
buffer.push_notify("second".into(), false);
buffer.push_notify("first".into());
buffer.push_notify("second".into());
let interceptor = WorkerInterceptor::new(
registry,
@@ -1671,7 +1671,7 @@ mod tests {
// anything itself.
let registry = Arc::new(HookRegistryBuilder::new().build());
let buffer = NotifyBuffer::new();
buffer.push_notify("msg".into(), false);
buffer.push_notify("msg".into());
let interceptor = WorkerInterceptor::new(
registry,
crates/worker/src/ipc/notify_buffer.rs
+27 -60
View File
@@ -26,13 +26,9 @@ use std::sync::{Arc, Mutex};
use protocol::WorkerEvent;
use session_store::{LoggedSessionHistoryOrigin, SessionExtension, SystemItem};
use tracing::warn;
use crate::prompt::catalog::{CatalogError, PromptCatalog};
/// Maximum queued pending entries. Oldest entries are dropped beyond this.
const CAPACITY: usize = 128;
/// One pending entry awaiting drain into the next LLM request.
///
/// The buffer keeps the raw input shape so the drain step can decide
@@ -43,7 +39,6 @@ const CAPACITY: usize = 128;
pub enum PendingNotify {
Notify {
message: String,
auto_run: bool,
extensions: Vec<SessionExtension>,
history_provenance: Option<LoggedSessionHistoryOrigin>,
},
@@ -76,6 +71,7 @@ impl PendingNotify {
#[derive(Clone, Default)]
pub struct NotifyBuffer {
inner: Arc<Mutex<VecDeque<PendingNotify>>>,
wake: Arc<tokio::sync::Notify>,
}
impl NotifyBuffer {
@@ -86,10 +82,9 @@ impl NotifyBuffer {
/// Push a notify entry onto the queue. If the queue is full, the
/// oldest entry is dropped and a `tracing::warn` is emitted — the
/// caller should never hit this in normal operation.
pub fn push_notify(&self, message: String, auto_run: bool) {
pub fn push_notify(&self, message: String) {
self.push_entry(PendingNotify::Notify {
message,
auto_run,
extensions: Vec::new(),
history_provenance: None,
});
@@ -98,53 +93,32 @@ impl NotifyBuffer {
pub fn push_durable_notify(
&self,
message: String,
auto_run: bool,
history_provenance: LoggedSessionHistoryOrigin,
extension: SessionExtension,
) {
self.push_entry(PendingNotify::Notify {
message,
auto_run,
extensions: vec![extension],
history_provenance: Some(history_provenance),
});
}
pub(crate) fn replace_durable_notification_extension(
&self,
extension: SessionExtension,
) -> bool {
let mut queue = self.inner.lock().expect("notify buffer poisoned");
let Some(extensions) = queue.iter_mut().rev().find_map(|pending| match pending {
PendingNotify::Notify {
auto_run: false,
extensions,
..
} if !extensions.is_empty() => Some(extensions),
_ => None,
}) else {
return false;
};
*extensions = vec![extension];
true
}
/// Push a typed worker-event entry onto the queue.
pub fn push_worker_event(&self, event: WorkerEvent) {
self.push_entry(PendingNotify::WorkerEvent { event });
}
fn push_entry(&self, entry: PendingNotify) {
let mut q = self.inner.lock().expect("notify buffer poisoned");
if q.len() >= CAPACITY {
let dropped = q.pop_front();
warn!(
capacity = CAPACITY,
dropped = ?dropped,
"notify buffer overflow; dropped oldest"
);
}
q.push_back(entry);
self.inner
.lock()
.expect("notify buffer poisoned")
.push_back(entry);
self.wake.notify_one();
}
/// Wait until a producer adds work outside the controller method channel.
pub async fn notified(&self) {
self.wake.notified().await;
}
/// Remove and return all pending entries in FIFO order.
@@ -160,23 +134,17 @@ impl NotifyBuffer {
for entry in entries.into_iter().rev() {
q.push_front(entry);
}
while q.len() > CAPACITY {
let dropped = q.pop_front();
warn!(
capacity = CAPACITY,
dropped = ?dropped,
"notify buffer overflow while restoring failed drain; dropped oldest"
);
}
drop(q);
self.wake.notify_one();
}
/// Whether an undrained `Method::Notify { auto_run: true }` remains.
pub fn has_auto_run_pending(&self) -> bool {
/// Whether an undrained notification remains.
pub fn has_notification_pending(&self) -> bool {
self.inner
.lock()
.expect("notify buffer poisoned")
.iter()
.any(|entry| matches!(entry, PendingNotify::Notify { auto_run: true, .. }))
.any(|entry| matches!(entry, PendingNotify::Notify { .. }))
}
/// Number of pending entries. Primarily for tests.
@@ -233,12 +201,12 @@ mod tests {
#[test]
fn push_then_drain_preserves_order() {
let buf = NotifyBuffer::new();
buf.push_notify("one".into(), false);
assert!(!buf.has_auto_run_pending());
buf.push_notify("two".into(), true);
assert!(buf.has_auto_run_pending());
buf.push_notify("one".into());
assert!(buf.has_notification_pending());
buf.push_notify("two".into());
assert!(buf.has_notification_pending());
let drained = buf.drain();
assert!(!buf.has_auto_run_pending());
assert!(!buf.has_notification_pending());
assert_eq!(drained.len(), 2);
match &drained[0] {
PendingNotify::Notify { message, .. } => assert_eq!(message, "one"),
@@ -248,15 +216,15 @@ mod tests {
}
#[test]
fn capacity_drops_oldest() {
fn notification_burst_is_not_truncated() {
let buf = NotifyBuffer::new();
for i in 0..(CAPACITY + 5) {
buf.push_notify(format!("msg{i}"), false);
for i in 0..300 {
buf.push_notify(format!("msg{i}"));
}
let drained = buf.drain();
assert_eq!(drained.len(), CAPACITY);
assert_eq!(drained.len(), 300);
match &drained[0] {
PendingNotify::Notify { message, .. } => assert_eq!(message, "msg5"),
PendingNotify::Notify { message, .. } => assert_eq!(message, "msg0"),
other => panic!("unexpected: {other:?}"),
}
}
@@ -265,7 +233,6 @@ mod tests {
fn build_system_item_for_notify_carries_wrapper_body() {
let entry = PendingNotify::Notify {
message: "hello".into(),
auto_run: false,
extensions: Vec::new(),
history_provenance: None,
};
crates/worker/src/runtime/worker_allocation.rs
+6 -2
View File
@@ -18,12 +18,16 @@ pub use conflict::{
ConflictOwner, find_conflict_owner, find_conflict_owners, is_within_effective_write,
};
pub use error::ScopeLockError;
#[cfg(test)]
pub(crate) use lifecycle::install_top_level_at_for_test;
pub use lifecycle::{
ScopeAllocationGuard, SegmentLockInfo, adopt_allocation, install_top_level,
install_top_level_with_deny, lookup_segment, update_segment,
ScopeAllocationGuard, SegmentActivationCommitError, SegmentActivationGuard, SegmentLockInfo,
adopt_allocation, install_top_level, install_top_level_with_deny, lookup_segment,
};
pub use mutate::{
delegate_scope, reclaim_delegated_scope, reclaim_stale, reclaim_stale_with, register_worker,
register_worker_with_deny, release_worker,
};
#[cfg(test)]
pub(crate) use table::fail_next_save_for_test;
pub use table::{Allocation, LockFile, LockFileGuard, default_allocation_path};
crates/worker/src/runtime/worker_allocation/error.rs
+8
View File
@@ -37,4 +37,12 @@ pub enum ScopeLockError {
worker_name: String,
socket: PathBuf,
},
#[error(
"worker `{worker_name}` allocation moved from expected Segment {expected} to {actual:?}"
)]
SegmentChanged {
worker_name: String,
expected: SegmentId,
actual: Option<SegmentId>,
},
}
crates/worker/src/runtime/worker_allocation/lifecycle.rs
+335 -65
View File
@@ -20,6 +20,41 @@ pub struct ScopeAllocationGuard {
lock_path: PathBuf,
}
/// Exclusive authority to activate a replacement Segment for one allocated Worker.
///
/// The allocation lock remains held across the durable metadata CAS and the final
/// allocation-table update, so restore admission cannot observe an intermediate
/// ownership state.
pub struct SegmentActivationGuard {
guard: LockFileGuard,
worker_name: String,
expected: SegmentId,
replacement: SegmentId,
}
/// A failed allocation-table publication together with the still-held machine-wide
/// activation authority. Callers must retain `authority` until recovery or process
/// teardown so restore admission cannot register a competing writer.
pub struct SegmentActivationCommitError {
source: ScopeLockError,
authority: SegmentActivationGuard,
}
impl std::fmt::Debug for SegmentActivationCommitError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("SegmentActivationCommitError")
.field("source", &self.source)
.finish_non_exhaustive()
}
}
impl SegmentActivationCommitError {
pub fn into_parts(self) -> (ScopeLockError, SegmentActivationGuard) {
(self.source, self.authority)
}
}
impl ScopeAllocationGuard {
pub fn worker_name(&self) -> &str {
&self.worker_name
@@ -28,6 +63,86 @@ impl ScopeAllocationGuard {
pub fn lock_path(&self) -> &Path {
&self.lock_path
}
/// Lock the machine-wide table and validate a Segment replacement before
/// its durable metadata is changed.
pub fn begin_segment_activation(
&self,
expected: SegmentId,
replacement: SegmentId,
) -> Result<SegmentActivationGuard, ScopeLockError> {
let guard = LockFileGuard::open(&self.lock_path)?;
let actual = guard
.data()
.allocations
.iter()
.find(|allocation| allocation.worker_name == self.worker_name)
.and_then(|allocation| allocation.segment_id);
if actual != Some(expected) {
return Err(ScopeLockError::SegmentChanged {
worker_name: self.worker_name.clone(),
expected,
actual,
});
}
if let Some(existing) = guard.data().allocations.iter().find(|allocation| {
allocation.segment_id == Some(replacement) && allocation.worker_name != self.worker_name
}) {
return Err(ScopeLockError::SegmentConflict {
segment_id: replacement,
worker_name: existing.worker_name.clone(),
socket: existing.socket.clone(),
});
}
Ok(SegmentActivationGuard {
guard,
worker_name: self.worker_name.clone(),
expected,
replacement,
})
}
}
impl SegmentActivationGuard {
/// Persist the replacement Segment while retaining exclusive allocation
/// authority. This is intentionally consumed so a successful metadata CAS
/// cannot accidentally be followed by an unlocked update.
pub fn commit(mut self) -> Result<(), SegmentActivationCommitError> {
let actual = self
.guard
.data()
.allocations
.iter()
.find(|allocation| allocation.worker_name == self.worker_name)
.and_then(|allocation| allocation.segment_id);
let Some(allocation) = self
.guard
.data_mut()
.allocations
.iter_mut()
.find(|allocation| allocation.worker_name == self.worker_name)
else {
return Err(SegmentActivationCommitError {
source: ScopeLockError::SegmentChanged {
worker_name: self.worker_name.clone(),
expected: self.expected,
actual,
},
authority: self,
});
};
allocation.segment_id = Some(self.replacement);
if let Err(source) = self.guard.save() {
return Err(SegmentActivationCommitError {
source: ScopeLockError::Io(source),
authority: self,
});
}
Ok(())
}
}
impl Drop for ScopeAllocationGuard {
@@ -85,6 +200,31 @@ pub fn install_top_level_with_deny(
})
}
#[cfg(test)]
pub(crate) fn install_top_level_at_for_test(
lock_path: PathBuf,
worker_name: String,
pid: u32,
socket: PathBuf,
scope_allow: Vec<ScopeRule>,
segment_id: SegmentId,
) -> Result<ScopeAllocationGuard, ScopeLockError> {
let mut guard = LockFileGuard::open(&lock_path)?;
super::mutate::register_worker_with_deny(
&mut guard,
worker_name.clone(),
pid,
socket,
scope_allow,
Vec::new(),
segment_id,
)?;
Ok(ScopeAllocationGuard {
worker_name,
lock_path,
})
}
/// Take ownership of an existing allocation that was pre-registered by
/// a spawning Worker.
///
@@ -114,46 +254,6 @@ pub fn adopt_allocation(
})
}
/// Rewrite the `segment_id` recorded for `worker_name` to
/// `new_segment_id`.
///
/// The Worker's in-memory `segment_id` can change underneath the
/// allocation in two normal places:
///
/// - `Worker::compact` mints a fresh Segment in the same Session.
/// - `session_store::ensure_head_or_fork` auto-forks within that Session when another
/// writer has advanced the store head behind our back.
///
/// Both paths must call this so subsequent [`lookup_segment`] queries
/// find the live Segment id, not the old one. Without this update a
/// concurrent `restore_from_manifest(new_id)` would see "no live
/// writer" and proceed to register a competing allocation on the
/// Segment lineage this Worker just moved into.
///
/// The lock is opened once and the allocation is rewritten inside the
/// guard, so the segment_id collision check is atomic with the
/// rewrite.
pub fn update_segment(worker_name: &str, new_segment_id: SegmentId) -> Result<(), ScopeLockError> {
let lock_path = default_allocation_path()?;
let mut guard = LockFileGuard::open(&lock_path)?;
if let Some(other) = guard.data().find_by_segment(new_segment_id) {
if other.worker_name != worker_name {
return Err(ScopeLockError::SegmentConflict {
segment_id: new_segment_id,
worker_name: other.worker_name.clone(),
socket: other.socket.clone(),
});
}
}
let alloc = guard
.data_mut()
.find_mut(worker_name)
.ok_or_else(|| ScopeLockError::UnknownWorker(worker_name.into()))?;
alloc.segment_id = Some(new_segment_id);
guard.save()?;
Ok(())
}
/// Information about a Worker that currently holds an allocation for a
/// given session.
#[derive(Debug, Clone)]
@@ -184,7 +284,7 @@ pub fn lookup_segment(segment_id: SegmentId) -> Result<Option<SegmentLockInfo>,
#[cfg(test)]
mod tests {
use super::super::table::Allocation;
use super::super::table::{Allocation, LockFile};
use super::super::test_util::*;
use super::*;
use tempfile::TempDir;
@@ -283,32 +383,12 @@ mod tests {
}
#[test]
fn update_session_rewrites_allocation_session_id() {
let dir = TempDir::new().unwrap();
let _sandbox = RuntimeDirSandbox::new(dir.path());
let original = sid();
let updated = sid();
let _guard = install_top_level(
"p".into(),
std::process::id(),
sock("p"),
vec![write_rule("/work", true)],
original,
)
.unwrap();
update_segment("p", updated).unwrap();
// lookup against the original is now empty, the updated id wins.
assert!(lookup_segment(original).unwrap().is_none());
assert_eq!(lookup_segment(updated).unwrap().unwrap().worker_name, "p");
}
#[test]
fn update_session_rejects_when_target_already_held() {
fn segment_activation_rejects_when_target_already_held() {
let dir = TempDir::new().unwrap();
let _sandbox = RuntimeDirSandbox::new(dir.path());
let s_a = sid();
let s_b = sid();
let _g_a = install_top_level(
let g_a = install_top_level(
"a".into(),
std::process::id(),
sock("a"),
@@ -324,8 +404,11 @@ mod tests {
s_b,
)
.unwrap();
// `a` cannot adopt b's live session id.
let err = update_segment("a", s_b).unwrap_err();
// `a` cannot activate b's live Segment id.
let err = match g_a.begin_segment_activation(s_a, s_b) {
Ok(_) => panic!("expected SegmentConflict"),
Err(error) => error,
};
match err {
ScopeLockError::SegmentConflict {
worker_name,
@@ -338,4 +421,191 @@ mod tests {
other => panic!("expected SegmentConflict, got {other:?}"),
}
}
#[test]
fn segment_activation_commits_verified_replacement() {
let dir = TempDir::new().unwrap();
let _sandbox = RuntimeDirSandbox::new(dir.path());
let lock_path = dir.path().join("workers.json");
let old_segment = sid();
let replacement = sid();
let guard = install_top_level(
"activation".into(),
std::process::id(),
sock("activation"),
vec![write_rule("/work", true)],
old_segment,
)
.unwrap();
guard
.begin_segment_activation(old_segment, replacement)
.unwrap()
.commit()
.unwrap();
let table = LockFileGuard::open(&lock_path).unwrap();
assert_eq!(
table
.data()
.find("activation")
.and_then(|allocation| allocation.segment_id),
Some(replacement)
);
}
#[test]
fn failed_segment_activation_returns_locked_recovery_authority() {
let dir = TempDir::new().unwrap();
let _sandbox = RuntimeDirSandbox::new(dir.path());
let old_segment = sid();
let replacement = sid();
let guard = install_top_level(
"activation".into(),
std::process::id(),
sock("activation"),
vec![write_rule("/work", true)],
old_segment,
)
.unwrap();
let mut activation = guard
.begin_segment_activation(old_segment, replacement)
.unwrap();
activation
.guard
.data_mut()
.allocations
.retain(|allocation| allocation.worker_name != "activation");
let error = activation.commit().unwrap_err();
let (source, authority) = error.into_parts();
assert!(matches!(
source,
ScopeLockError::SegmentChanged {
expected,
actual: None,
..
} if expected == old_segment
));
// The authority is returned rather than dropped on failure. Production
// stores it on Worker to keep restore admission blocked.
let (sent, received) = std::sync::mpsc::channel();
let lookup = std::thread::spawn(move || {
sent.send(lookup_segment(replacement).unwrap()).unwrap();
});
assert!(
received
.recv_timeout(std::time::Duration::from_millis(50))
.is_err()
);
drop(authority);
assert!(
received
.recv_timeout(std::time::Duration::from_secs(1))
.unwrap()
.is_none()
);
lookup.join().unwrap();
}
#[test]
fn allocation_save_failure_preserves_source_table_and_returns_authority() {
let dir = TempDir::new().unwrap();
let _sandbox = RuntimeDirSandbox::new(dir.path());
let lock_path = dir.path().join("workers.json");
let old_segment = sid();
let replacement = sid();
let guard = install_top_level(
"activation".into(),
std::process::id(),
sock("activation"),
vec![write_rule("/work", true)],
old_segment,
)
.unwrap();
let activation = guard
.begin_segment_activation(old_segment, replacement)
.unwrap();
crate::runtime::worker_allocation::table::fail_next_save_for_test(&lock_path);
let error = activation.commit().unwrap_err();
let (source, authority) = error.into_parts();
assert!(matches!(source, ScopeLockError::Io(_)));
let persisted: LockFile =
serde_json::from_slice(&std::fs::read(&lock_path).unwrap()).unwrap();
assert_eq!(
persisted
.find("activation")
.and_then(|allocation| allocation.segment_id),
Some(old_segment)
);
drop(authority);
}
#[test]
fn dropping_segment_activation_preserves_source_allocation() {
let dir = TempDir::new().unwrap();
let _sandbox = RuntimeDirSandbox::new(dir.path());
let lock_path = dir.path().join("workers.json");
let old_segment = sid();
let replacement = sid();
let guard = install_top_level(
"activation".into(),
std::process::id(),
sock("activation"),
vec![write_rule("/work", true)],
old_segment,
)
.unwrap();
drop(
guard
.begin_segment_activation(old_segment, replacement)
.unwrap(),
);
let table = LockFileGuard::open(&lock_path).unwrap();
assert_eq!(
table
.data()
.find("activation")
.and_then(|allocation| allocation.segment_id),
Some(old_segment)
);
}
#[test]
fn segment_activation_rejects_stale_expected_segment_without_mutation() {
let dir = TempDir::new().unwrap();
let _sandbox = RuntimeDirSandbox::new(dir.path());
let lock_path = dir.path().join("workers.json");
let allocated = sid();
let stale = sid();
let replacement = sid();
let guard = install_top_level(
"activation".into(),
std::process::id(),
sock("activation"),
vec![write_rule("/work", true)],
allocated,
)
.unwrap();
assert!(matches!(
guard.begin_segment_activation(stale, replacement),
Err(ScopeLockError::SegmentChanged {
expected,
actual: Some(actual),
..
}) if expected == stale && actual == allocated
));
let table = LockFileGuard::open(&lock_path).unwrap();
assert_eq!(
table
.data()
.find("activation")
.and_then(|allocation| allocation.segment_id),
Some(allocated)
);
}
}
crates/worker/src/runtime/worker_allocation/table.rs
+163 -14
View File
@@ -1,12 +1,48 @@
//! On-disk allocation table and the `flock`-protected guard.
use std::fs::{DirBuilder, File, OpenOptions};
use std::fs::{self, DirBuilder, File, OpenOptions};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
use std::path::{Path, PathBuf};
use std::thread;
use std::time::{Duration, Instant};
#[cfg(test)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SaveFaultPoint {
BeforeRename,
AfterRename,
}
#[cfg(test)]
static SAVE_FAULTS: std::sync::LazyLock<
std::sync::Mutex<std::collections::HashMap<PathBuf, SaveFaultPoint>>,
> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashMap::new()));
#[cfg(test)]
pub(crate) fn fail_next_save_for_test(path: &Path) {
fail_next_save_at_for_test(path, SaveFaultPoint::BeforeRename);
}
#[cfg(test)]
pub(crate) fn fail_next_save_at_for_test(path: &Path, point: SaveFaultPoint) {
SAVE_FAULTS
.lock()
.unwrap()
.insert(path.to_path_buf(), point);
}
#[cfg(test)]
fn take_save_fault(path: &Path, point: SaveFaultPoint) -> bool {
let mut faults = SAVE_FAULTS.lock().unwrap();
if faults.get(path) == Some(&point) {
faults.remove(path);
true
} else {
false
}
}
use fs4::fs_std::FileExt;
use manifest::{ScopeRule, paths};
use serde::{Deserialize, Serialize};
@@ -101,7 +137,9 @@ pub fn default_allocation_path() -> io::Result<PathBuf> {
/// never call `save` leave the table unchanged, which is the right
/// behaviour for error paths.
pub struct LockFileGuard {
file: File,
data_file: File,
lock_file: File,
data_path: PathBuf,
data: LockFile,
}
@@ -121,16 +159,19 @@ impl LockFileGuard {
.mode(0o700)
.create(parent)?;
}
let file = OpenOptions::new()
let mut lock_os = path.as_os_str().to_owned();
lock_os.push(".lock");
let lock_path = PathBuf::from(lock_os);
let lock_file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.mode(0o600)
.open(path)?;
.open(&lock_path)?;
let started = Instant::now();
loop {
match FileExt::try_lock_exclusive(&file) {
match FileExt::try_lock_exclusive(&lock_file) {
Ok(true) => break,
Ok(false) => {
if started.elapsed() >= LOCK_WAIT_TIMEOUT {
@@ -159,8 +200,17 @@ impl LockFileGuard {
Err(error) => return Err(error),
}
}
let data_file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.mode(0o600)
.open(path)?;
let mut this = Self {
file,
data_file,
lock_file,
data_path: path.to_path_buf(),
data: LockFile::default(),
};
this.reload()?;
@@ -168,9 +218,9 @@ impl LockFileGuard {
}
fn reload(&mut self) -> io::Result<()> {
self.file.seek(SeekFrom::Start(0))?;
self.data_file.seek(SeekFrom::Start(0))?;
let mut buf = String::new();
self.file.read_to_string(&mut buf)?;
self.data_file.read_to_string(&mut buf)?;
self.data = if buf.trim().is_empty() {
LockFile::default()
} else {
@@ -192,20 +242,49 @@ impl LockFileGuard {
&mut self.data
}
/// Serialise `self.data` back to the file (truncate + rewrite).
/// Persist with atomic replacement while the separate allocation lock stays
/// held. A crash exposes either the old complete table or the new one.
pub fn save(&mut self) -> io::Result<()> {
let json = serde_json::to_vec_pretty(&self.data).map_err(io::Error::other)?;
self.file.seek(SeekFrom::Start(0))?;
self.file.set_len(0)?;
self.file.write_all(&json)?;
self.file.sync_data()?;
let mut temp_os = self.data_path.as_os_str().to_owned();
temp_os.push(".tmp");
let temp_path = PathBuf::from(temp_os);
let mut temp_file = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.mode(0o600)
.open(&temp_path)?;
temp_file.write_all(&json)?;
temp_file.sync_all()?;
#[cfg(test)]
if take_save_fault(&self.data_path, SaveFaultPoint::BeforeRename) {
return Err(io::Error::other(
"injected allocation save failure before rename",
));
}
fs::rename(&temp_path, &self.data_path)?;
#[cfg(test)]
if take_save_fault(&self.data_path, SaveFaultPoint::AfterRename) {
return Err(io::Error::other(
"injected allocation save failure after rename",
));
}
if let Some(parent) = self.data_path.parent() {
File::open(parent)?.sync_all()?;
}
self.data_file = OpenOptions::new()
.read(true)
.write(true)
.mode(0o600)
.open(&self.data_path)?;
Ok(())
}
}
impl Drop for LockFileGuard {
fn drop(&mut self) {
let _ = FileExt::unlock(&self.file);
let _ = FileExt::unlock(&self.lock_file);
}
}
@@ -259,6 +338,76 @@ mod tests {
assert_eq!(guard.data().allocations[0].worker_name, "a");
}
#[test]
fn atomic_save_reopens_complete_table_at_both_rename_boundaries() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("workers.json");
let old_segment = sid();
let replacement = sid();
{
let mut guard = open_empty(&path);
register_worker(
&mut guard,
"a".into(),
std::process::id(),
sock("a"),
vec![write_rule("/src", true)],
old_segment,
)
.unwrap();
}
{
let mut guard = LockFileGuard::open(&path).unwrap();
guard.data_mut().find_mut("a").unwrap().segment_id = Some(replacement);
fail_next_save_at_for_test(&path, SaveFaultPoint::BeforeRename);
assert!(guard.save().is_err());
}
let guard = LockFileGuard::open(&path).unwrap();
assert_eq!(
guard
.data()
.find("a")
.and_then(|allocation| allocation.segment_id),
Some(old_segment)
);
drop(guard);
{
let mut guard = LockFileGuard::open(&path).unwrap();
guard.data_mut().find_mut("a").unwrap().segment_id = Some(replacement);
fail_next_save_at_for_test(&path, SaveFaultPoint::AfterRename);
assert!(guard.save().is_err());
}
let guard = LockFileGuard::open(&path).unwrap();
assert_eq!(
guard
.data()
.find("a")
.and_then(|allocation| allocation.segment_id),
Some(replacement)
);
}
#[test]
fn allocation_lock_remains_exclusive_across_atomic_save() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("workers.json");
let mut guard = open_empty(&path);
guard.save().unwrap();
let (sent, received) = std::sync::mpsc::channel();
let contender_path = path.clone();
let contender = std::thread::spawn(move || {
let contender = LockFileGuard::open(&contender_path).unwrap();
sent.send(contender.data().allocations.len()).unwrap();
});
assert!(received.recv_timeout(Duration::from_millis(50)).is_err());
drop(guard);
assert_eq!(received.recv_timeout(Duration::from_secs(1)).unwrap(), 0);
contender.join().unwrap();
}
#[test]
fn find_by_session_skips_none_placeholders() {
let dir = TempDir::new().unwrap();
crates/worker/src/skill.rs
+2 -2
View File
@@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
use crate::worker::WorkspaceClient;
pub use workspace_api::{
pub use server_api::{
SkillActivationStatus, SkillCatalogEntry, SkillCatalogResponse, SkillDetailResponse,
SkillDiagnostic, SkillDiagnosticSeverity, SkillProjectionIdentity, SkillProjectionStatus,
SkillProvenance, SkillResourceRef, SkillSourceKind,
@@ -29,7 +29,7 @@ pub enum SkillClientError {
#[error("Skill API response JSON is invalid: {0}")]
Json(#[from] serde_json::Error),
#[error("Skill API response violates the shared contract: {0}")]
InvalidResponse(#[from] workspace_api::SkillApiValidationError),
InvalidResponse(#[from] server_api::SkillApiValidationError),
#[error("Skill API returned HTTP {status}: {body}")]
Http {
status: reqwest::StatusCode,
crates/worker/src/spawn/registry.rs
+58 -2
View File
@@ -8,6 +8,8 @@
//! Parent registry drop closes all session handles and synchronously returns delegated Write deny
//! rules to the parent scope.
#[cfg(test)]
use std::collections::HashMap;
use std::collections::{BTreeMap, HashSet};
use std::io;
use std::sync::{
@@ -311,6 +313,8 @@ impl Drop for InternalSpawnReservation {
pub struct SpawnedWorkerRegistry {
internal_records: std::sync::Mutex<Vec<InternalSpawnedWorkerRecord>>,
service_records: std::sync::Mutex<Vec<InternalServiceWorkerRecord>>,
#[cfg(test)]
fail_service_stops: std::sync::Mutex<HashMap<String, usize>>,
internal_names: std::sync::Mutex<HashSet<String>>,
internal_shutting_down: AtomicBool,
pending_internal_spawns: AtomicUsize,
@@ -331,6 +335,8 @@ impl SpawnedWorkerRegistry {
Arc::new(Self {
internal_records: std::sync::Mutex::new(Vec::new()),
service_records: std::sync::Mutex::new(Vec::new()),
#[cfg(test)]
fail_service_stops: std::sync::Mutex::new(HashMap::new()),
internal_names: std::sync::Mutex::new(HashSet::new()),
internal_shutting_down: AtomicBool::new(false),
pending_internal_spawns: AtomicUsize::new(0),
@@ -346,6 +352,8 @@ impl SpawnedWorkerRegistry {
Arc::new(Self {
internal_records: std::sync::Mutex::new(Vec::new()),
service_records: std::sync::Mutex::new(Vec::new()),
#[cfg(test)]
fail_service_stops: std::sync::Mutex::new(HashMap::new()),
internal_names: std::sync::Mutex::new(HashSet::new()),
internal_shutting_down: AtomicBool::new(false),
pending_internal_spawns: AtomicUsize::new(0),
@@ -360,6 +368,8 @@ impl SpawnedWorkerRegistry {
Arc::new(Self {
internal_records: std::sync::Mutex::new(Vec::new()),
service_records: std::sync::Mutex::new(Vec::new()),
#[cfg(test)]
fail_service_stops: std::sync::Mutex::new(HashMap::new()),
internal_names: std::sync::Mutex::new(HashSet::new()),
internal_shutting_down: AtomicBool::new(false),
pending_internal_spawns: AtomicUsize::new(0),
@@ -443,6 +453,8 @@ impl SpawnedWorkerRegistry {
registry: Arc::new(Self {
internal_records: std::sync::Mutex::new(Vec::new()),
service_records: std::sync::Mutex::new(Vec::new()),
#[cfg(test)]
fail_service_stops: std::sync::Mutex::new(HashMap::new()),
internal_names: std::sync::Mutex::new(HashSet::new()),
internal_shutting_down: AtomicBool::new(false),
pending_internal_spawns: AtomicUsize::new(0),
@@ -535,9 +547,53 @@ impl SpawnedWorkerRegistry {
Ok(worker_ref)
}
/// Stop and remove one parent-owned service Worker. This is host-only and is
/// intentionally separate from the SubWorker control surface.
#[cfg(test)]
pub(crate) fn install_service_for_test(
&self,
) -> (String, tokio::sync::broadcast::Sender<protocol::Event>) {
let (session, status_tx) = crate::internal_worker::test_internal_worker_session(
crate::internal_worker::InternalWorkerVisibility::ServicePrivate,
);
let session_id = session.session_id_string();
self.service_records
.lock()
.unwrap()
.push(InternalServiceWorkerRecord::new(
"test-service",
"test-service",
session,
));
(session_id, status_tx)
}
#[cfg(test)]
pub(crate) fn has_service_for_test(&self, session_id: &str) -> bool {
self.service_records
.lock()
.unwrap()
.iter()
.any(|record| record.session.session_id_string() == session_id)
}
#[cfg(test)]
pub(crate) fn fail_service_stops_for_test(&self, session_id: &str, count: usize) {
self.fail_service_stops
.lock()
.unwrap()
.insert(session_id.to_owned(), count);
}
pub(crate) async fn stop_service(&self, session_id: &str) -> io::Result<bool> {
#[cfg(test)]
{
let mut failures = self.fail_service_stops.lock().unwrap();
if let Some(remaining) = failures.get_mut(session_id) {
if *remaining > 0 {
*remaining -= 1;
return Err(io::Error::other("injected service stop failure"));
}
}
}
let record = self
.service_records
.lock()
crates/worker/src/spawn/tool.rs
+34 -12
View File
@@ -262,28 +262,28 @@ impl ParentNotificationTarget {
Self::Controller { sender, fallback }
}
pub(crate) fn notify(&self, child_session_id: String, message: String, auto_run: bool) {
pub(crate) fn notify(&self, child_session_id: String, message: String) {
let method = Method::NotifyTracked {
notification_request_id: protocol::new_submission_request_id(),
message,
auto_run,
source: protocol::AuthenticatedInputSource::SubWorker {
session_id: child_session_id,
},
};
match self {
Self::Controller { sender, fallback } => {
// Persist through the Worker-owned pending authority before the
// child completion path returns. The Controller delivery is only
// a wake-up hint; replay is idempotent by notification_request_id.
fallback(method.clone());
let Some(parent_method_tx) = sender.upgrade() else {
fallback(method);
return;
};
let fallback = fallback.clone();
tokio::spawn(async move {
if let Err(error) = parent_method_tx.send(method).await {
if parent_method_tx.send(method).await.is_err() {
tracing::warn!(
"failed to notify parent Controller; using durable pending authority"
"failed to wake parent Controller after durable notification acceptance"
);
fallback(error.0);
}
});
}
@@ -588,7 +588,7 @@ impl Tool for SubWorkerSpawnTool {
let message = format!(
"SubWorker `{child_name}` turn ended with status {status:?}. Inspect its committed session with worker-observation tools before making completion decisions."
);
parent_notifications.notify(child_name.clone(), message, true);
parent_notifications.notify(child_name.clone(), message);
})),
Some(child_workdir_tool_broker.clone()),
)
@@ -1199,10 +1199,34 @@ enabled = false
drop(parent_method_tx);
assert!(parent_method_rx.recv().await.is_none());
target.notify("child-session".into(), "late completion".to_string(), true);
target.notify("child-session".into(), "late completion".to_string());
assert!(*captured.lock().unwrap());
}
#[tokio::test]
async fn parent_controller_notification_is_durable_before_wakeup_delivery() {
let (parent_method_tx, mut parent_method_rx) = mpsc::channel(1);
let durable = Arc::new(AtomicBool::new(false));
let durable_for_fallback = Arc::clone(&durable);
let target = ParentNotificationTarget::with_controller_fallback(
parent_method_tx.downgrade(),
ParentNotificationTarget::Durable(Arc::new(move |_| {
durable_for_fallback.store(true, Ordering::Release);
})),
);
target.notify("child-session".into(), "completed".into());
let wakeup = parent_method_rx.recv().await.unwrap();
assert!(durable.load(Ordering::Acquire));
assert!(matches!(
wakeup,
Method::NotifyTracked {
source: protocol::AuthenticatedInputSource::SubWorker { session_id },
..
} if session_id == "child-session"
));
}
#[test]
fn durable_parent_notification_target_preserves_child_source() {
let captured = Arc::new(std::sync::Mutex::new(None));
@@ -1211,13 +1235,12 @@ enabled = false
*captured_for_target.lock().unwrap() = Some(method);
}));
target.notify("child-session".into(), "completed".into(), true);
target.notify("child-session".into(), "completed".into());
assert!(matches!(
captured.lock().unwrap().take(),
Some(Method::NotifyTracked {
message,
auto_run: true,
source: protocol::AuthenticatedInputSource::SubWorker { session_id },
..
}) if session_id == "child-session" && message == "completed"
@@ -1361,7 +1384,6 @@ enabled = false
completion,
Method::NotifyTracked {
message,
auto_run: true,
source: protocol::AuthenticatedInputSource::SubWorker { session_id },
..
} if session_id == "reviewer-child"
crates/worker/src/worker.rs
+746 -219
View File
File diff suppressed because it is too large Load Diff
crates/worker/tests/compact_events_test.rs
+43 -6
View File
@@ -344,7 +344,10 @@ target = "./"
permission = "write"
"#;
async fn make_worker_with_manifest<C>(manifest_toml: &str, client: C) -> Worker<C, TestStore>
async fn make_worker_with_manifest_and_store<C>(
manifest_toml: &str,
client: C,
) -> (Worker<C, TestStore>, TestStore)
where
C: LlmClient + Clone + Send + Sync + 'static,
{
@@ -364,6 +367,7 @@ where
let worker =
Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(client);
let observed_store = store.clone();
let mut worker = Worker::new(
manifest,
worker,
@@ -375,7 +379,16 @@ where
.await
.unwrap();
worker.enable_worker_metadata_write_through().unwrap();
worker
(worker, observed_store)
}
async fn make_worker_with_manifest<C>(manifest_toml: &str, client: C) -> Worker<C, TestStore>
where
C: LlmClient + Clone + Send + Sync + 'static,
{
make_worker_with_manifest_and_store(manifest_toml, client)
.await
.0
}
async fn make_worker(client: MockClient) -> Worker<MockClient, TestStore> {
@@ -922,7 +935,7 @@ async fn pre_run_compact_publishes_runtime_progress_phases() {
let session_before = worker.session_id();
let segment_before = worker.segment_id();
worker.try_pre_run_compact().await;
worker.try_pre_run_compact().await.unwrap();
assert_eq!(worker.session_id(), session_before);
assert_ne!(worker.segment_id(), segment_before);
@@ -1083,7 +1096,7 @@ async fn pre_run_compact_failure_clears_runtime_progress() {
let _ = drain(&mut rx);
// Best-effort: returns Ok(()) even on failure and clears runtime progress.
worker.try_pre_run_compact().await;
worker.try_pre_run_compact().await.unwrap();
let events = drain(&mut rx);
assert!(events.iter().any(|event| matches!(
@@ -1104,8 +1117,11 @@ async fn pre_run_compact_failure_clears_runtime_progress() {
#[tokio::test]
async fn manual_compact_cancel_clears_progress_before_returning_idle() {
let worker =
make_worker_with_manifest(POST_RUN_MANIFEST_TOML, BlockingCompactClient::new()).await;
let (worker, observed_store) =
make_worker_with_manifest_and_store(POST_RUN_MANIFEST_TOML, BlockingCompactClient::new())
.await;
let source_session = worker.session_id();
let source_segment = worker.segment_id();
let runtime_tmp = tempfile::tempdir().unwrap();
let bash_output_dir = runtime_tmp.path().join("bash-output");
let (handle, shutdown_receiver) =
@@ -1201,6 +1217,15 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() {
break;
}
}
let completion_request_id = "subworker-completion-during-compact";
handle
.send(Method::NotifyTracked {
notification_request_id: completion_request_id.into(),
message: "subworker completed".into(),
source: protocol::AuthenticatedInputSource::UntrustedWire,
})
.await
.expect("send completion notification during compact");
let shutdown = protocol::WorkerCommandEnvelope::new(4);
handle
.send(Method::Shutdown { command: shutdown })
@@ -1230,6 +1255,18 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() {
.await
.expect("controller shutdown timeout")
.expect("shutdown confirmation");
let persisted = observed_store
.read_all(source_session, source_segment)
.expect("read source Segment after shutdown");
assert!(persisted.iter().any(|entry| {
matches!(
entry,
LogEntry::Extension { domain, payload, .. }
if domain == "worker.pending_activations.v1"
&& payload.to_string().contains(completion_request_id)
)
}));
}
#[tokio::test]
crates/worker/tests/controller_test.rs
+63 -47
View File
@@ -1783,7 +1783,9 @@ async fn run_with_unresolved_segment_emits_alert_and_placeholder() {
async fn notify_while_idle_auto_starts_turn_and_injects_system_message() {
let client = MockClient::new(simple_text_events());
let client_for_assert = client.clone();
let worker = make_worker(client).await;
let worker = make_worker(client)
.await
.with_notification_coalesce_delay(std::time::Duration::from_millis(20));
let handle = spawn_controller(worker).await;
let mut rx = handle.subscribe();
@@ -1791,7 +1793,6 @@ async fn notify_while_idle_auto_starts_turn_and_injects_system_message() {
.send(Method::Notify {
notification_request_id: protocol::new_submission_request_id(),
message: "turn finished".into(),
auto_run: true,
})
.await
.unwrap();
@@ -1883,10 +1884,12 @@ async fn notify_while_idle_auto_starts_turn_and_injects_system_message() {
}
#[tokio::test]
async fn notify_while_idle_with_auto_run_false_waits_for_explicit_run() {
async fn repeated_notify_while_idle_coalesces_and_auto_starts_one_turn() {
let client = MockClient::new(simple_text_events());
let client_for_assert = client.clone();
let worker = make_worker(client).await;
let worker = make_worker(client)
.await
.with_notification_coalesce_delay(std::time::Duration::from_millis(50));
let handle = spawn_controller(worker).await;
let notification_request_id = protocol::new_submission_request_id();
@@ -1895,26 +1898,18 @@ async fn notify_while_idle_with_auto_run_false_waits_for_explicit_run() {
.send(Method::Notify {
notification_request_id: notification_request_id.clone(),
message: "progress snapshot".into(),
auto_run: false,
})
.await
.unwrap();
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Idle);
assert!(
client_for_assert.captured_requests().is_empty(),
"weak Notify must not stage RunForNotification while idle"
);
handle
.send(Method::submit_text(
protocol::new_submission_request_id(),
"continue",
))
.send(Method::Notify {
notification_request_id: "notify-coalesced-2".into(),
message: "second coalesced notification".into(),
})
.await
.unwrap();
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
loop {
if !client_for_assert.captured_requests().is_empty() {
@@ -1922,30 +1917,26 @@ async fn notify_while_idle_with_auto_run_false_waits_for_explicit_run() {
}
assert!(
tokio::time::Instant::now() < deadline,
"explicit run did not reach the mock LLM"
"coalescing deadline did not start a notification turn"
);
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
wait_for_status(&handle, WorkerStatus::Idle).await;
let requests = client_for_assert.captured_requests();
assert_eq!(requests.len(), 1, "coalesced notifications need one turn");
let notifications = requests[0]
.items
.iter()
.filter_map(|item| item.as_text())
.filter(|text| text.contains("[Notification]"))
.collect::<Vec<_>>();
assert_eq!(
requests.len(),
1,
"explicit run should drain the queued notification"
);
let notify_in_request = requests[0].items.iter().any(|i| {
i.as_text()
.is_some_and(|t| t.contains("[Notification]") && t.contains("progress snapshot"))
});
assert!(
notify_in_request,
"queued weak notification must be history-backed on the next explicit run; got items: {:?}",
requests[0]
.items
.iter()
.filter_map(|i| i.as_text())
.collect::<Vec<_>>()
notifications.len(),
2,
"duplicate receipt must not duplicate content"
);
assert!(notifications[0].contains("progress snapshot"));
assert!(notifications[1].contains("second coalesced notification"));
}
#[tokio::test]
@@ -2105,7 +2096,6 @@ async fn notify_while_running_does_not_emit_already_running_error() {
.send(Method::Notify {
notification_request_id: protocol::new_submission_request_id(),
message: "ping".into(),
auto_run: true,
})
.await
.unwrap();
@@ -2137,7 +2127,7 @@ async fn notify_while_running_does_not_emit_already_running_error() {
}
#[tokio::test]
async fn weak_notify_while_running_is_deduped_and_survives_until_next_submit() {
async fn notify_while_running_is_deduped_and_survives_until_next_model_boundary() {
let client = MockClient::sequential(vec![
MockResponse::Hang(Vec::new()),
MockResponse::Complete(simple_text_events()),
@@ -2160,7 +2150,6 @@ async fn weak_notify_while_running_is_deduped_and_survives_until_next_submit() {
.send(Method::Notify {
notification_request_id: notification_request_id.clone(),
message: "durable weak notice".into(),
auto_run: false,
})
.await
.unwrap();
@@ -2586,11 +2575,11 @@ async fn drain_until<F: FnMut(&Event) -> bool>(
}
}
/// Pause mid-stream, then Resume: status round-trips Running →
/// Paused → Running → Idle, and the final history contains exactly
/// one user turn plus the assistant reply produced by the resume call.
/// Paused → Running → Idle. A notification accepted while Paused must not
/// auto-resume the Worker, and explicit Resume must inject it while preserving
/// the interrupted turn's history consistency.
#[tokio::test]
async fn pause_then_resume_transitions_and_preserves_history_consistency() {
async fn pause_then_resume_preserves_notifications_and_history_consistency() {
// Response 1: hang after opening a text block (no stop / completed),
// so the Engine is parked inside the stream read and `cancel_rx`
// races it cleanly on Method::Pause.
@@ -2608,7 +2597,10 @@ async fn pause_then_resume_transitions_and_preserves_history_consistency() {
}),
]);
let client = MockClient::sequential(vec![hang, ok]);
let worker = make_worker(client).await;
let client_for_assert = client.clone();
let worker = make_worker(client)
.await
.with_notification_coalesce_delay(std::time::Duration::from_millis(20));
let handle = spawn_controller(worker).await;
let mut rx = handle.subscribe();
@@ -2654,6 +2646,25 @@ async fn pause_then_resume_transitions_and_preserves_history_consistency() {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Paused);
handle
.send(Method::Notify {
notification_request_id: "notify-while-paused".into(),
message: "resume context".into(),
})
.await
.unwrap();
tokio::time::sleep(std::time::Duration::from_millis(80)).await;
assert_eq!(
handle.shared_state.catalog_status(),
WorkerStatus::Paused,
"notification deadline must not auto-resume a paused Worker"
);
assert_eq!(
client_for_assert.captured_requests().len(),
1,
"paused notification must wait for an explicit resume"
);
handle
.send(Method::Resume {
command: worker_command(&handle),
@@ -2674,10 +2685,15 @@ async fn pause_then_resume_transitions_and_preserves_history_consistency() {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Idle);
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| text.contains("[Notification]") && text.contains("resume context"))
}));
// History consistency: exactly [user "hello", assistant
// "resumed output"]. No artifacts from the aborted stream
// (partial text is not committed), no orphan tool_use.
// History consistency: the interrupted partial response is absent, while
// the queued notification is committed on the explicit resume.
let history = history_from_sink(&handle);
let roles: Vec<&str> = history
.iter()
@@ -2692,8 +2708,8 @@ async fn pause_then_resume_transitions_and_preserves_history_consistency() {
.collect();
assert_eq!(
roles,
vec!["user", "assistant"],
"history = user + assistant only; got {history:?}"
vec!["user", "system", "assistant"],
"history should retain user input, queued notification, and resumed output; got {history:?}"
);
let assistant_text = history
.iter()