Synced from monorepo
Changes: - Classify clipboard delivery confidence - Add durable session update append - Scope the xAI session bearer to first-party memory embedding endpoints - Persist subagent outputs to disk and bound long-lived agent state - Add MiniSweAgent:bash for mini-swe-agent parity - Revert taking local sessions off the persistent shell - Contextual tip recommending grok wrap on SSH sessions - Voice STT bearer from model BYOK env_key/api_key - Define exact website policies for sandbox - Gate unsafe shell environments - Shared pin hoist; single require_sha gate for marketplace plugins - Server-signed is-managed claim (closes sidecar-removal downgrade) - Optional require_sha pin for remote plugin installs - Show session title and last exchange in the exit resume hint - Gate shell output redirects - Warn when fail_closed is present but not a boolean - Add canonical text editing core (ratatui-textarea) - Keep execution state out of goal scratch - Add acknowledged persistence primitives - Inherit child network restrictions in sandbox - Fail closed when hook matchers fail to recompile - Add MCP setup preferences for plugin MCPs - Gate sourced shell scripts - Gate file-typed project hooks - grok wrap: restore terminal modes on child death - Harden owner-only permissions on auth and MCP credentials - Create crash dump files with owner-only permissions - Write the agent_id cache owner-only (0600) - SessionMetrics mode skips Mixpanel profile sync - Dashboard: slim live-tail peek - Yank full queued prompt text, not (+N lines) - Defeat clock-rollback on the signed managed-config cache - Stop early session/cancel from overtaking the prompt and wedging the turn slot - Self-heal a diverged agent entrypoint on startup - Add matched inference expectations in test-support - Add AuthSingleFlight cancel/successor gap tests - Remove consumer from external OTEL allowlist and pin scrub coverage - Enable /copy in minimal mode - Surface capacity and API-key detail on 429 errors - Single-flight interactive auth - Fix PageUp/PageDown skipping lines behind sticky prompt header
This commit is contained in:
grokkybara[bot]
parent
8adf9013a0
commit
98c3b2438a
225 changed files with 18836 additions and 7156 deletions
|
|
@ -96,7 +96,7 @@ libc = { workspace = true }
|
|||
# AssignProcessToJobObject, TerminateJobObject) and process creation flags
|
||||
# (CREATE_NO_WINDOW, DETACHED_PROCESS, CREATE_NEW_PROCESS_GROUP).
|
||||
[target.'cfg(windows)'.dependencies]
|
||||
windows = { workspace = true }
|
||||
windows = { workspace = true, features = ["Win32_Storage_FileSystem"] }
|
||||
|
||||
[dev-dependencies]
|
||||
dirs = { workspace = true }
|
||||
|
|
|
|||
|
|
@ -150,14 +150,8 @@ enum TerminalCommand {
|
|||
reply: oneshot::Sender<Option<PathBuf>>,
|
||||
},
|
||||
|
||||
WarmShell {
|
||||
cwd: PathBuf,
|
||||
},
|
||||
|
||||
/// Kill all running foreground processes owned by a specific session.
|
||||
KillForegroundCommandsByOwner {
|
||||
owner_session_id: String,
|
||||
},
|
||||
KillForegroundCommandsByOwner { owner_session_id: String },
|
||||
|
||||
/// Kill all running background tasks owned by a specific session.
|
||||
KillTasksByOwner {
|
||||
|
|
@ -597,25 +591,6 @@ impl LocalTerminalActor {
|
|||
})
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
async fn ensure_persistent_shell_initialized(&mut self, cwd: &std::path::Path) {
|
||||
if self.shell_state.is_some() {
|
||||
return;
|
||||
}
|
||||
let shell = shell_state::ShellKind::detect();
|
||||
match shell_state::ShellState::init(shell, cwd).await {
|
||||
Ok(state) => self.shell_state = Some(state),
|
||||
Err(e) => {
|
||||
tracing::warn!("persistent shell init failed, using empty state: {e}");
|
||||
self.shell_state = Some(shell_state::ShellState {
|
||||
cwd: cwd.to_path_buf(),
|
||||
snapshot: String::new(),
|
||||
shell,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn a command with persistent shell state: restore the prior snapshot
|
||||
/// via fd 3, run the user command, dump the new state to fd 4.
|
||||
#[cfg(unix)]
|
||||
|
|
@ -627,7 +602,20 @@ impl LocalTerminalActor {
|
|||
) -> Result<SpawnResult, ComputerError> {
|
||||
use command_fds::CommandFdExt;
|
||||
|
||||
self.ensure_persistent_shell_initialized(cwd).await;
|
||||
if self.shell_state.is_none() {
|
||||
let shell = shell_state::ShellKind::detect();
|
||||
match shell_state::ShellState::init(shell, cwd).await {
|
||||
Ok(state) => self.shell_state = Some(state),
|
||||
Err(e) => {
|
||||
tracing::warn!("persistent shell init failed, using empty state: {e}");
|
||||
self.shell_state = Some(shell_state::ShellState {
|
||||
cwd: cwd.to_path_buf(),
|
||||
snapshot: String::new(),
|
||||
shell,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let shell_state = self.shell_state.as_ref().unwrap();
|
||||
// When the persistent shell already tracks a
|
||||
|
|
@ -809,14 +797,6 @@ impl LocalTerminalActor {
|
|||
let cwd = None;
|
||||
let _ = reply.send(cwd);
|
||||
}
|
||||
TerminalCommand::WarmShell { cwd } => {
|
||||
#[cfg(unix)]
|
||||
if self.persistent_shell {
|
||||
self.ensure_persistent_shell_initialized(&cwd).await;
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
let _ = cwd;
|
||||
}
|
||||
TerminalCommand::KillForegroundCommands => {
|
||||
self.kill_foreground_commands().await;
|
||||
}
|
||||
|
|
@ -2281,15 +2261,6 @@ impl TerminalBackend for LocalTerminalBackend {
|
|||
reply_rx.await.ok().flatten()
|
||||
}
|
||||
|
||||
async fn warm_persistent_shell(&self, cwd: &std::path::Path) {
|
||||
let _ = self
|
||||
.cmd_tx
|
||||
.send(TerminalCommand::WarmShell {
|
||||
cwd: cwd.to_path_buf(),
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn kill_foreground_commands(&self) {
|
||||
let _ = self
|
||||
.cmd_tx
|
||||
|
|
|
|||
|
|
@ -292,10 +292,6 @@ pub trait TerminalBackend: Send + Sync {
|
|||
/// only the subagent's own tasks are killed — not the parent's.
|
||||
async fn kill_all_background_tasks_by_owner(&self, _owner_session_id: &str) {}
|
||||
|
||||
/// Fire-and-forget prewarm of the persistent login shell; default no-op for
|
||||
/// backends without one (ACP/remote, non-persistent).
|
||||
async fn warm_persistent_shell(&self, _cwd: &std::path::Path) {}
|
||||
|
||||
/// Reparent notification handles for all tasks owned by `old_owner_session_id`.
|
||||
/// Swaps the dead child session's notification handle with the parent's
|
||||
/// live handle so events from surviving processes route correctly.
|
||||
|
|
|
|||
264
crates/codegen/xai-grok-tools/src/notification/handle.rs
Normal file
264
crates/codegen/xai-grok-tools/src/notification/handle.rs
Normal file
|
|
@ -0,0 +1,264 @@
|
|||
use std::sync::Arc;
|
||||
|
||||
use super::types::{
|
||||
BashExecutionBackgrounded, BashExecutionComplete, BashExecutionFailed, BashExecutionTimeout,
|
||||
BashOutputChunk, FileWritten, LspServerCrashed, LspServerFailed, LspServerReady,
|
||||
LspServerRetrying, LspServerStarting, MonitorEvent, PlanModeEntered, PlanModeExited,
|
||||
ScheduledTaskCreated, ScheduledTaskFired, ScheduledTaskRemoved, ToolNotification,
|
||||
UserQuestionAsked,
|
||||
};
|
||||
use crate::types::TaskSnapshot;
|
||||
|
||||
/// Envelope for consumers that can acknowledge durable notification handling.
|
||||
pub struct AcknowledgedToolNotification {
|
||||
/// Notification delivered in the same FIFO as unacknowledged events.
|
||||
pub notification: ToolNotification,
|
||||
/// Completion sender present only when the producer requested acknowledgement.
|
||||
pub acknowledgement: Option<tokio::sync::oneshot::Sender<Result<(), String>>>,
|
||||
}
|
||||
|
||||
/// Failure reported after all acknowledged notification targets have settled.
|
||||
#[derive(Debug, PartialEq, Eq, thiserror::Error)]
|
||||
pub enum NotificationAcknowledgementError {
|
||||
#[error("{0} acknowledging notification target(s) closed during dispatch")]
|
||||
DispatchClosed(usize),
|
||||
#[error("{0} notification acknowledgement(s) were dropped")]
|
||||
AcknowledgementDropped(usize),
|
||||
#[error("notification consumer rejected delivery: {0:?}")]
|
||||
ConsumerRejected(Vec<String>),
|
||||
#[error(
|
||||
"notification acknowledgement failed: {dispatch_closed} dispatch closed, {acknowledgements_dropped} acknowledgement(s) dropped, consumer rejections: {consumer_rejections:?}"
|
||||
)]
|
||||
Multiple {
|
||||
dispatch_closed: usize,
|
||||
acknowledgements_dropped: usize,
|
||||
consumer_rejections: Vec<String>,
|
||||
},
|
||||
}
|
||||
|
||||
/// Receipts for one acknowledged fan-out operation.
|
||||
#[must_use = "acknowledged notification receipts must be awaited"]
|
||||
pub struct NotificationAcknowledgementBatch {
|
||||
receipts: Vec<tokio::sync::oneshot::Receiver<Result<(), String>>>,
|
||||
durable_targets: usize,
|
||||
dispatch_closed: usize,
|
||||
}
|
||||
|
||||
/// Whether an acknowledged send has configured durable notification targets.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum DurableNotificationTargets {
|
||||
None,
|
||||
Present,
|
||||
}
|
||||
|
||||
impl NotificationAcknowledgementBatch {
|
||||
/// Whether any target was configured for durable acknowledgement.
|
||||
pub fn durable_targets(&self) -> DurableNotificationTargets {
|
||||
if self.durable_targets == 0 {
|
||||
DurableNotificationTargets::None
|
||||
} else {
|
||||
DurableNotificationTargets::Present
|
||||
}
|
||||
}
|
||||
|
||||
/// Wait for every live durable target and report all observed failure classes.
|
||||
pub async fn wait(self) -> Result<(), NotificationAcknowledgementError> {
|
||||
let mut acknowledgements_dropped = 0;
|
||||
let mut consumer_rejections = Vec::new();
|
||||
for receipt in self.receipts {
|
||||
match receipt.await {
|
||||
Ok(Ok(())) => {}
|
||||
Ok(Err(error)) => consumer_rejections.push(error),
|
||||
Err(_) => acknowledgements_dropped += 1,
|
||||
}
|
||||
}
|
||||
match (
|
||||
self.dispatch_closed,
|
||||
acknowledgements_dropped,
|
||||
consumer_rejections.is_empty(),
|
||||
) {
|
||||
(0, 0, true) => Ok(()),
|
||||
(dispatch_closed, 0, true) => Err(NotificationAcknowledgementError::DispatchClosed(
|
||||
dispatch_closed,
|
||||
)),
|
||||
(0, acknowledgements_dropped, true) => Err(
|
||||
NotificationAcknowledgementError::AcknowledgementDropped(acknowledgements_dropped),
|
||||
),
|
||||
(0, 0, false) => Err(NotificationAcknowledgementError::ConsumerRejected(
|
||||
consumer_rejections,
|
||||
)),
|
||||
(dispatch_closed, acknowledgements_dropped, _) => {
|
||||
Err(NotificationAcknowledgementError::Multiple {
|
||||
dispatch_closed,
|
||||
acknowledgements_dropped,
|
||||
consumer_rejections,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
enum ToolNotificationTarget {
|
||||
Plain(tokio::sync::mpsc::UnboundedSender<ToolNotification>),
|
||||
Acknowledged(tokio::sync::mpsc::UnboundedSender<AcknowledgedToolNotification>),
|
||||
}
|
||||
|
||||
/// Cloneable notification fan-out with per-target FIFO ordering.
|
||||
#[derive(Clone)]
|
||||
pub struct ToolNotificationHandle {
|
||||
targets: Arc<[ToolNotificationTarget]>,
|
||||
}
|
||||
|
||||
impl Default for ToolNotificationHandle {
|
||||
fn default() -> Self {
|
||||
Self::noop()
|
||||
}
|
||||
}
|
||||
|
||||
macro_rules! convenience_sends {
|
||||
($($method:ident, $ty:ty, $variant:ident);+ $(;)?) => {
|
||||
$(pub fn $method(&self, value: $ty) { self.send(ToolNotification::$variant(value)); })+
|
||||
};
|
||||
}
|
||||
|
||||
impl ToolNotificationHandle {
|
||||
pub fn new(sender: tokio::sync::mpsc::UnboundedSender<ToolNotification>) -> Self {
|
||||
Self {
|
||||
targets: Arc::from([ToolNotificationTarget::Plain(sender)]),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn from_sender(sender: tokio::sync::mpsc::UnboundedSender<ToolNotification>) -> Self {
|
||||
Self::new(sender)
|
||||
}
|
||||
|
||||
pub fn channel() -> (Self, tokio::sync::mpsc::UnboundedReceiver<ToolNotification>) {
|
||||
let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
(Self::new(sender), receiver)
|
||||
}
|
||||
|
||||
pub fn acknowledged_channel() -> (
|
||||
Self,
|
||||
tokio::sync::mpsc::UnboundedReceiver<AcknowledgedToolNotification>,
|
||||
) {
|
||||
let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
(
|
||||
Self {
|
||||
targets: Arc::from([ToolNotificationTarget::Acknowledged(sender)]),
|
||||
},
|
||||
receiver,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn noop() -> Self {
|
||||
let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
Self::new(sender)
|
||||
}
|
||||
|
||||
/// Combine handles while preserving each target's send order.
|
||||
pub fn tee(handles: Vec<ToolNotificationHandle>) -> ToolNotificationHandle {
|
||||
let targets = handles
|
||||
.iter()
|
||||
.flat_map(|handle| handle.targets.iter().cloned())
|
||||
.collect::<Vec<_>>();
|
||||
Self {
|
||||
targets: Arc::from(targets),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn send(&self, notification: ToolNotification) {
|
||||
let last = self.targets.len().saturating_sub(1);
|
||||
let mut notification = Some(notification);
|
||||
for (index, target) in self.targets.iter().enumerate() {
|
||||
let notification = if index == last {
|
||||
let Some(notification) = notification.take() else {
|
||||
break;
|
||||
};
|
||||
notification
|
||||
} else {
|
||||
let Some(notification) = notification.as_ref() else {
|
||||
break;
|
||||
};
|
||||
notification.clone()
|
||||
};
|
||||
match target {
|
||||
ToolNotificationTarget::Plain(target) => {
|
||||
let _ = target.send(notification);
|
||||
}
|
||||
ToolNotificationTarget::Acknowledged(target) => {
|
||||
let _ = target.send(AcknowledgedToolNotification {
|
||||
notification,
|
||||
acknowledgement: None,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a removal to every target and collect all durable acknowledgements.
|
||||
pub fn send_scheduled_task_removed_acknowledged(
|
||||
&self,
|
||||
removed: ScheduledTaskRemoved,
|
||||
) -> NotificationAcknowledgementBatch {
|
||||
let notification = ToolNotification::ScheduledTaskRemoved(removed);
|
||||
let mut batch = NotificationAcknowledgementBatch {
|
||||
receipts: Vec::new(),
|
||||
durable_targets: 0,
|
||||
dispatch_closed: 0,
|
||||
};
|
||||
for target in self.targets.iter() {
|
||||
match target {
|
||||
ToolNotificationTarget::Plain(target) => {
|
||||
let _ = target.send(notification.clone());
|
||||
}
|
||||
ToolNotificationTarget::Acknowledged(target) => {
|
||||
batch.durable_targets += 1;
|
||||
let (acknowledgement, receipt) = tokio::sync::oneshot::channel();
|
||||
if target
|
||||
.send(AcknowledgedToolNotification {
|
||||
notification: notification.clone(),
|
||||
acknowledgement: Some(acknowledgement),
|
||||
})
|
||||
.is_ok()
|
||||
{
|
||||
batch.receipts.push(receipt);
|
||||
} else {
|
||||
batch.dispatch_closed += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
batch
|
||||
}
|
||||
|
||||
convenience_sends! {
|
||||
send_output_chunk, BashOutputChunk, BashOutputChunk;
|
||||
send_complete, BashExecutionComplete, BashExecutionComplete;
|
||||
send_timeout, BashExecutionTimeout, BashExecutionTimeout;
|
||||
send_backgrounded, BashExecutionBackgrounded, BashExecutionBackgrounded;
|
||||
send_failed, BashExecutionFailed, BashExecutionFailed;
|
||||
send_file_written, FileWritten, FileWritten;
|
||||
send_task_complete, TaskSnapshot, TaskCompleted;
|
||||
send_plan_mode_entered, PlanModeEntered, PlanModeEntered;
|
||||
send_plan_mode_exited, PlanModeExited, PlanModeExited;
|
||||
send_user_question_asked, UserQuestionAsked, UserQuestionAsked;
|
||||
send_lsp_starting, LspServerStarting, LspServerStarting;
|
||||
send_lsp_ready, LspServerReady, LspServerReady;
|
||||
send_lsp_crashed, LspServerCrashed, LspServerCrashed;
|
||||
send_lsp_retrying, LspServerRetrying, LspServerRetrying;
|
||||
send_lsp_failed, LspServerFailed, LspServerFailed;
|
||||
send_scheduled_task_fired, ScheduledTaskFired, ScheduledTaskFired;
|
||||
send_scheduled_task_removed, ScheduledTaskRemoved, ScheduledTaskRemoved;
|
||||
send_scheduled_task_created, ScheduledTaskCreated, ScheduledTaskCreated;
|
||||
send_monitor_event, MonitorEvent, MonitorEvent;
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-call notification override applied in addition to the session-wide sink.
|
||||
#[derive(Clone)]
|
||||
pub struct PerCallNotificationSink(pub ToolNotificationHandle);
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "handle_tests.rs"]
|
||||
mod tests;
|
||||
117
crates/codegen/xai-grok-tools/src/notification/handle_tests.rs
Normal file
117
crates/codegen/xai-grok-tools/src/notification/handle_tests.rs
Normal file
|
|
@ -0,0 +1,117 @@
|
|||
use super::*;
|
||||
|
||||
fn removed(task_id: &str) -> ScheduledTaskRemoved {
|
||||
ScheduledTaskRemoved {
|
||||
task_id: task_id.into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn created(task_id: &str) -> ScheduledTaskCreated {
|
||||
ScheduledTaskCreated {
|
||||
task_id: task_id.into(),
|
||||
prompt: task_id.into(),
|
||||
human_schedule: "every 5 minutes".into(),
|
||||
next_fire_at: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn task_id(notification: &ToolNotification) -> &str {
|
||||
match notification {
|
||||
ToolNotification::ScheduledTaskCreated(value) => &value.task_id,
|
||||
ToolNotification::ScheduledTaskRemoved(value) => &value.task_id,
|
||||
other => panic!("unexpected notification: {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn acknowledged_removal_stays_in_fifo() {
|
||||
let (handle, mut receiver) = ToolNotificationHandle::acknowledged_channel();
|
||||
handle.send_scheduled_task_created(created("before"));
|
||||
let batch = handle.send_scheduled_task_removed_acknowledged(removed("deleted"));
|
||||
handle.send_scheduled_task_created(created("after"));
|
||||
|
||||
let first = receiver.recv().await.unwrap();
|
||||
assert_eq!(task_id(&first.notification), "before");
|
||||
assert!(first.acknowledgement.is_none());
|
||||
let second = receiver.recv().await.unwrap();
|
||||
assert_eq!(task_id(&second.notification), "deleted");
|
||||
second.acknowledgement.unwrap().send(Ok(())).unwrap();
|
||||
let third = receiver.recv().await.unwrap();
|
||||
assert_eq!(task_id(&third.notification), "after");
|
||||
assert!(third.acknowledgement.is_none());
|
||||
|
||||
assert_eq!(batch.durable_targets(), DurableNotificationTargets::Present);
|
||||
batch.wait().await.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mixed_fanout_attempts_every_target_before_reporting_closed_dispatch() {
|
||||
let (closed, closed_rx) = ToolNotificationHandle::acknowledged_channel();
|
||||
drop(closed_rx);
|
||||
let (plain, mut plain_rx) = ToolNotificationHandle::channel();
|
||||
let (durable, mut durable_rx) = ToolNotificationHandle::acknowledged_channel();
|
||||
let handle = ToolNotificationHandle::tee(vec![closed, plain, durable]);
|
||||
|
||||
handle.send_scheduled_task_created(created("before"));
|
||||
let batch = handle.send_scheduled_task_removed_acknowledged(removed("deleted"));
|
||||
handle.send_scheduled_task_created(created("after"));
|
||||
|
||||
assert_eq!(task_id(&plain_rx.recv().await.unwrap()), "before");
|
||||
assert_eq!(task_id(&plain_rx.recv().await.unwrap()), "deleted");
|
||||
assert_eq!(task_id(&plain_rx.recv().await.unwrap()), "after");
|
||||
let durable_before = durable_rx.recv().await.unwrap();
|
||||
assert_eq!(task_id(&durable_before.notification), "before");
|
||||
let durable_removed = durable_rx.recv().await.unwrap();
|
||||
assert_eq!(task_id(&durable_removed.notification), "deleted");
|
||||
durable_removed
|
||||
.acknowledgement
|
||||
.unwrap()
|
||||
.send(Ok(()))
|
||||
.unwrap();
|
||||
let durable_after = durable_rx.recv().await.unwrap();
|
||||
assert_eq!(task_id(&durable_after.notification), "after");
|
||||
|
||||
assert_eq!(
|
||||
batch.wait().await,
|
||||
Err(NotificationAcknowledgementError::DispatchClosed(1))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn batch_distinguishes_dropped_and_rejected_acknowledgements() {
|
||||
let (dropped, mut dropped_rx) = ToolNotificationHandle::acknowledged_channel();
|
||||
let (rejected, mut rejected_rx) = ToolNotificationHandle::acknowledged_channel();
|
||||
let handle = ToolNotificationHandle::tee(vec![dropped, rejected]);
|
||||
let batch = handle.send_scheduled_task_removed_acknowledged(removed("deleted"));
|
||||
|
||||
drop(dropped_rx.recv().await.unwrap().acknowledgement);
|
||||
rejected_rx
|
||||
.recv()
|
||||
.await
|
||||
.unwrap()
|
||||
.acknowledgement
|
||||
.unwrap()
|
||||
.send(Err("rejected".into()))
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
batch.wait().await,
|
||||
Err(NotificationAcknowledgementError::Multiple {
|
||||
dispatch_closed: 0,
|
||||
acknowledgements_dropped: 1,
|
||||
consumer_rejections: vec!["rejected".into()],
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn plain_and_noop_batches_make_zero_durable_targets_explicit() {
|
||||
for handle in [
|
||||
ToolNotificationHandle::channel().0,
|
||||
ToolNotificationHandle::noop(),
|
||||
] {
|
||||
let batch = handle.send_scheduled_task_removed_acknowledged(removed("deleted"));
|
||||
assert_eq!(batch.durable_targets(), DurableNotificationTargets::None);
|
||||
batch.wait().await.unwrap();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +1,12 @@
|
|||
pub mod handle;
|
||||
pub mod types;
|
||||
|
||||
pub use handle::AcknowledgedToolNotification;
|
||||
pub use handle::DurableNotificationTargets;
|
||||
pub use handle::NotificationAcknowledgementBatch;
|
||||
pub use handle::NotificationAcknowledgementError;
|
||||
pub use handle::PerCallNotificationSink;
|
||||
pub use handle::ToolNotificationHandle;
|
||||
pub use types::ALL_NOTIFICATION_TAGS;
|
||||
pub use types::BashExecutionBackgrounded;
|
||||
pub use types::BashExecutionComplete;
|
||||
|
|
@ -15,13 +22,11 @@ pub use types::LspServerReady;
|
|||
pub use types::LspServerRetrying;
|
||||
pub use types::LspServerStarting;
|
||||
pub use types::MonitorEvent;
|
||||
pub use types::PerCallNotificationSink;
|
||||
pub use types::PlanModeEntered;
|
||||
pub use types::PlanModeExited;
|
||||
pub use types::ScheduledTaskCreated;
|
||||
pub use types::ScheduledTaskFired;
|
||||
pub use types::ScheduledTaskRemoved;
|
||||
pub use types::ToolNotification;
|
||||
pub use types::ToolNotificationHandle;
|
||||
pub use types::UserQuestionAsked;
|
||||
pub use types::notification_schema_catalog;
|
||||
|
|
|
|||
|
|
@ -4,10 +4,11 @@
|
|||
//! - updates being sent by the tools as they are executing (for example bash tools)
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::types::TaskSnapshot;
|
||||
|
||||
pub use super::handle::{PerCallNotificationSink, ToolNotificationHandle};
|
||||
|
||||
/// Common fields for all bash execution notifications.
|
||||
/// Extracting these ensures consistent naming and makes refactoring easier.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, schemars::JsonSchema)]
|
||||
|
|
@ -480,236 +481,10 @@ notification_variants! {
|
|||
MonitorEvent => MonitorEvent,
|
||||
}
|
||||
|
||||
/// Handle for sending notifications to consumers.
|
||||
/// Clone-able so it can be passed to multiple tool implementations.
|
||||
///
|
||||
/// Internally holds one-or-many sender targets. Every existing constructor
|
||||
/// (`new`, `from_sender`, `channel`, `noop`) builds a single-target handle and
|
||||
/// behaves exactly as before; [`ToolNotificationHandle::tee`] builds a
|
||||
/// fan-out handle whose [`send`](Self::send) delivers each notification to all
|
||||
/// targets, in order, preserving per-target ordering.
|
||||
#[derive(Clone)]
|
||||
pub struct ToolNotificationHandle {
|
||||
targets: Arc<[tokio::sync::mpsc::UnboundedSender<ToolNotification>]>,
|
||||
}
|
||||
|
||||
impl Default for ToolNotificationHandle {
|
||||
fn default() -> Self {
|
||||
Self::noop()
|
||||
}
|
||||
}
|
||||
|
||||
impl ToolNotificationHandle {
|
||||
/// Create a new handle with the given sender
|
||||
pub fn new(sender: tokio::sync::mpsc::UnboundedSender<ToolNotification>) -> Self {
|
||||
Self {
|
||||
targets: Arc::from([sender]),
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a handle from an existing unbounded sender.
|
||||
/// Alias for `new()` — used by tests and consumers that want to receive notifications.
|
||||
pub fn from_sender(sender: tokio::sync::mpsc::UnboundedSender<ToolNotification>) -> Self {
|
||||
Self::new(sender)
|
||||
}
|
||||
|
||||
/// Create a channel pair (handle + receiver)
|
||||
pub fn channel() -> (Self, tokio::sync::mpsc::UnboundedReceiver<ToolNotification>) {
|
||||
let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
(Self::new(sender), receiver)
|
||||
}
|
||||
|
||||
/// Create a no-op handle (sends are silently dropped)
|
||||
pub fn noop() -> Self {
|
||||
let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
|
||||
Self::new(sender)
|
||||
}
|
||||
|
||||
/// Fan-out: build a handle that delivers every notification to all the
|
||||
/// underlying targets of the given `handles`, in order.
|
||||
///
|
||||
/// Each send is delivered to every target in `handles` (flattened), so a
|
||||
/// single tool call's notifications can be surfaced on several sinks at
|
||||
/// once (e.g. the session-wide handle plus a per-call sink). Per-target
|
||||
/// ordering is preserved: targets observe sends in the same order on the
|
||||
/// caller's thread. [`ToolNotification`] derives `Clone`, so each extra
|
||||
/// target receives a clone.
|
||||
///
|
||||
/// An empty input (`tee(vec![])`) yields a handle with no targets whose
|
||||
/// `send` silently drops every notification — i.e. equivalent to
|
||||
/// [`noop`](Self::noop).
|
||||
pub fn tee(handles: Vec<ToolNotificationHandle>) -> ToolNotificationHandle {
|
||||
let targets: Vec<_> = handles
|
||||
.iter()
|
||||
.flat_map(|h| h.targets.iter().cloned())
|
||||
.collect();
|
||||
Self {
|
||||
targets: Arc::from(targets),
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a notification to all targets, in order.
|
||||
pub fn send(&self, notification: ToolNotification) {
|
||||
// Single-target hot path is one send with no clone; for fan-out we
|
||||
// clone for every target except the last, which takes ownership.
|
||||
let last = self.targets.len().saturating_sub(1);
|
||||
for (i, target) in self.targets.iter().enumerate() {
|
||||
if i == last {
|
||||
let _ = target.send(notification);
|
||||
break;
|
||||
}
|
||||
let _ = target.send(notification.clone());
|
||||
}
|
||||
}
|
||||
|
||||
// === Convenience methods ===
|
||||
|
||||
pub fn send_output_chunk(&self, chunk: BashOutputChunk) {
|
||||
self.send(ToolNotification::BashOutputChunk(chunk));
|
||||
}
|
||||
|
||||
pub fn send_complete(&self, complete: BashExecutionComplete) {
|
||||
self.send(ToolNotification::BashExecutionComplete(complete));
|
||||
}
|
||||
|
||||
pub fn send_timeout(&self, timeout: BashExecutionTimeout) {
|
||||
self.send(ToolNotification::BashExecutionTimeout(timeout));
|
||||
}
|
||||
|
||||
pub fn send_backgrounded(&self, backgrounded: BashExecutionBackgrounded) {
|
||||
self.send(ToolNotification::BashExecutionBackgrounded(backgrounded));
|
||||
}
|
||||
|
||||
pub fn send_failed(&self, failed: BashExecutionFailed) {
|
||||
self.send(ToolNotification::BashExecutionFailed(failed));
|
||||
}
|
||||
|
||||
pub fn send_file_written(&self, written: FileWritten) {
|
||||
self.send(ToolNotification::FileWritten(written));
|
||||
}
|
||||
|
||||
pub fn send_task_complete(&self, task_completed: TaskSnapshot) {
|
||||
self.send(ToolNotification::TaskCompleted(task_completed))
|
||||
}
|
||||
|
||||
pub fn send_plan_mode_entered(&self, entered: PlanModeEntered) {
|
||||
self.send(ToolNotification::PlanModeEntered(entered));
|
||||
}
|
||||
|
||||
pub fn send_plan_mode_exited(&self, exited: PlanModeExited) {
|
||||
self.send(ToolNotification::PlanModeExited(exited));
|
||||
}
|
||||
|
||||
pub fn send_user_question_asked(&self, asked: UserQuestionAsked) {
|
||||
self.send(ToolNotification::UserQuestionAsked(asked));
|
||||
}
|
||||
|
||||
pub fn send_lsp_starting(&self, starting: LspServerStarting) {
|
||||
self.send(ToolNotification::LspServerStarting(starting));
|
||||
}
|
||||
|
||||
pub fn send_lsp_ready(&self, ready: LspServerReady) {
|
||||
self.send(ToolNotification::LspServerReady(ready));
|
||||
}
|
||||
|
||||
pub fn send_lsp_crashed(&self, crashed: LspServerCrashed) {
|
||||
self.send(ToolNotification::LspServerCrashed(crashed));
|
||||
}
|
||||
|
||||
pub fn send_lsp_retrying(&self, retrying: LspServerRetrying) {
|
||||
self.send(ToolNotification::LspServerRetrying(retrying));
|
||||
}
|
||||
|
||||
pub fn send_lsp_failed(&self, failed: LspServerFailed) {
|
||||
self.send(ToolNotification::LspServerFailed(failed));
|
||||
}
|
||||
|
||||
pub fn send_scheduled_task_fired(&self, fired: ScheduledTaskFired) {
|
||||
self.send(ToolNotification::ScheduledTaskFired(fired));
|
||||
}
|
||||
|
||||
pub fn send_scheduled_task_removed(&self, removed: ScheduledTaskRemoved) {
|
||||
self.send(ToolNotification::ScheduledTaskRemoved(removed));
|
||||
}
|
||||
|
||||
pub fn send_scheduled_task_created(&self, created: ScheduledTaskCreated) {
|
||||
self.send(ToolNotification::ScheduledTaskCreated(created));
|
||||
}
|
||||
|
||||
pub fn send_monitor_event(&self, event: MonitorEvent) {
|
||||
self.send(ToolNotification::MonitorEvent(event));
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-call notification override.
|
||||
///
|
||||
/// When present in `ToolCallContext::extensions`, tools tee their execution
|
||||
/// notifications here IN ADDITION to the session-wide handle, so a single
|
||||
/// call's notifications (e.g. bash output chunks) can be surfaced as in-band
|
||||
/// progress for that one tool call without disturbing the session-wide
|
||||
/// side-channel.
|
||||
///
|
||||
/// This follows the same per-call ctx-extension pattern as `InnerDispatch` /
|
||||
/// `Cwd`: a simple clone-able newtype wrapper inserted into and pulled out of
|
||||
/// `ToolCallContext::extensions`.
|
||||
#[derive(Clone)]
|
||||
pub struct PerCallNotificationSink(pub ToolNotificationHandle);
|
||||
|
||||
#[cfg(test)]
|
||||
mod handle_tests {
|
||||
mod payload_tests {
|
||||
use super::*;
|
||||
|
||||
fn chunk(tool_call_id: &str) -> ToolNotification {
|
||||
ToolNotification::BashOutputChunk(BashOutputChunk {
|
||||
base: BashNotificationBase {
|
||||
tool_call_id: tool_call_id.into(),
|
||||
command: "echo hi".into(),
|
||||
output: b"hi".to_vec(),
|
||||
total_bytes: 2,
|
||||
truncated: false,
|
||||
cwd: PathBuf::from("/"),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
fn tool_call_id(n: &ToolNotification) -> &str {
|
||||
match n {
|
||||
ToolNotification::BashOutputChunk(c) => &c.base.tool_call_id,
|
||||
other => panic!("expected BashOutputChunk, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn single_target_hot_path_receives_exactly_what_was_sent() {
|
||||
let (handle, mut rx) = ToolNotificationHandle::channel();
|
||||
handle.send(chunk("a"));
|
||||
handle.send(chunk("b"));
|
||||
drop(handle);
|
||||
|
||||
assert_eq!(tool_call_id(&rx.try_recv().unwrap()), "a");
|
||||
assert_eq!(tool_call_id(&rx.try_recv().unwrap()), "b");
|
||||
assert!(rx.try_recv().is_err(), "no extra notifications expected");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tee_delivers_to_all_targets_in_order() {
|
||||
let (h1, mut rx1) = ToolNotificationHandle::channel();
|
||||
let (h2, mut rx2) = ToolNotificationHandle::channel();
|
||||
let teed = ToolNotificationHandle::tee(vec![h1, h2]);
|
||||
|
||||
teed.send(chunk("a"));
|
||||
teed.send(chunk("b"));
|
||||
teed.send(chunk("c"));
|
||||
drop(teed);
|
||||
|
||||
for rx in [&mut rx1, &mut rx2] {
|
||||
assert_eq!(tool_call_id(&rx.try_recv().unwrap()), "a");
|
||||
assert_eq!(tool_call_id(&rx.try_recv().unwrap()), "b");
|
||||
assert_eq!(tool_call_id(&rx.try_recv().unwrap()), "c");
|
||||
assert!(rx.try_recv().is_err(), "no extra notifications expected");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn catalog_has_one_schema_per_variant() {
|
||||
let catalog = notification_schema_catalog();
|
||||
|
|
|
|||
|
|
@ -3,9 +3,12 @@
|
|||
//! [`ResourcesPersistence`] persists `Resources` state (the new architecture).
|
||||
//! Old `ToolStatePersistence` and `PersistenceLayer` have been deleted.
|
||||
|
||||
use std::io;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::Duration;
|
||||
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
use crate::types::resources::Resources;
|
||||
|
||||
/// Background persistence for `Resources` state/params.
|
||||
|
|
@ -23,11 +26,16 @@ pub struct ResourcesPersistence {
|
|||
state_path: PathBuf,
|
||||
/// Channel to send serialized state to the background writer
|
||||
tx: tokio::sync::mpsc::UnboundedSender<ResourcesPersistenceCommand>,
|
||||
noop: bool,
|
||||
}
|
||||
|
||||
enum ResourcesPersistenceCommand {
|
||||
/// Write this serialized Resources value to disk
|
||||
Save(serde_json::Value),
|
||||
SaveAndFlush {
|
||||
snapshot: serde_json::Value,
|
||||
respond_to: tokio::sync::oneshot::Sender<io::Result<()>>,
|
||||
},
|
||||
/// Flush pending writes and notify when done
|
||||
Flush(tokio::sync::oneshot::Sender<()>),
|
||||
}
|
||||
|
|
@ -39,6 +47,7 @@ impl ResourcesPersistence {
|
|||
Self {
|
||||
state_path: PathBuf::from("/dev/null"),
|
||||
tx,
|
||||
noop: true,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -51,7 +60,11 @@ impl ResourcesPersistence {
|
|||
Self::writer_loop(rx, writer_path).await;
|
||||
});
|
||||
|
||||
Self { state_path, tx }
|
||||
Self {
|
||||
state_path,
|
||||
tx,
|
||||
noop: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Load existing Resources state from disk, if the file exists.
|
||||
|
|
@ -96,10 +109,55 @@ impl ResourcesPersistence {
|
|||
/// Save the current Resources state (non-blocking).
|
||||
/// Sends a serialized snapshot to the background writer.
|
||||
pub fn save(&self, resources: &Resources) {
|
||||
if self.noop {
|
||||
return;
|
||||
}
|
||||
let snapshot = resources.serialize();
|
||||
let _ = self.tx.send(ResourcesPersistenceCommand::Save(snapshot));
|
||||
}
|
||||
|
||||
/// Replace pending snapshots, write this snapshot, and acknowledge the result.
|
||||
pub fn enqueue_save_and_flush(
|
||||
&self,
|
||||
snapshot: serde_json::Value,
|
||||
) -> io::Result<tokio::sync::oneshot::Receiver<io::Result<()>>> {
|
||||
if self.noop {
|
||||
let (respond_to, response) = tokio::sync::oneshot::channel();
|
||||
let _ = respond_to.send(Ok(()));
|
||||
return Ok(response);
|
||||
}
|
||||
let (respond_to, response) = tokio::sync::oneshot::channel();
|
||||
self.tx
|
||||
.send(ResourcesPersistenceCommand::SaveAndFlush {
|
||||
snapshot,
|
||||
respond_to,
|
||||
})
|
||||
.map_err(|_| {
|
||||
io::Error::new(
|
||||
io::ErrorKind::BrokenPipe,
|
||||
"resources persistence writer stopped",
|
||||
)
|
||||
})?;
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
/// Await an acknowledgement returned by [`Self::enqueue_save_and_flush`].
|
||||
pub async fn await_save_and_flush(
|
||||
response: tokio::sync::oneshot::Receiver<io::Result<()>>,
|
||||
) -> io::Result<()> {
|
||||
response.await.map_err(|_| {
|
||||
io::Error::new(
|
||||
io::ErrorKind::BrokenPipe,
|
||||
"resources persistence writer dropped acknowledgement",
|
||||
)
|
||||
})?
|
||||
}
|
||||
|
||||
/// Replace pending snapshots, write this snapshot, and await the result.
|
||||
pub async fn save_and_flush(&self, snapshot: serde_json::Value) -> io::Result<()> {
|
||||
Self::await_save_and_flush(self.enqueue_save_and_flush(snapshot)?).await
|
||||
}
|
||||
|
||||
/// Path to the persisted state file.
|
||||
pub fn state_path(&self) -> &std::path::Path {
|
||||
&self.state_path
|
||||
|
|
@ -107,6 +165,9 @@ impl ResourcesPersistence {
|
|||
|
||||
/// Flush pending writes. Call on graceful shutdown.
|
||||
pub async fn flush(&self) {
|
||||
if self.noop {
|
||||
return;
|
||||
}
|
||||
let (done_tx, done_rx) = tokio::sync::oneshot::channel();
|
||||
let _ = self.tx.send(ResourcesPersistenceCommand::Flush(done_tx));
|
||||
let _ = done_rx.await;
|
||||
|
|
@ -147,59 +208,151 @@ impl ResourcesPersistence {
|
|||
Some(ResourcesPersistenceCommand::Save(snapshot)) => {
|
||||
pending = Some(snapshot);
|
||||
}
|
||||
Some(ResourcesPersistenceCommand::SaveAndFlush {
|
||||
snapshot,
|
||||
respond_to,
|
||||
}) => {
|
||||
pending = None;
|
||||
let result = Self::write_json_durable(&state_path, &snapshot).await;
|
||||
let _ = respond_to.send(result);
|
||||
}
|
||||
Some(ResourcesPersistenceCommand::Flush(done)) => {
|
||||
if let Some(snapshot) = pending.take() {
|
||||
Self::write_json(&state_path, &snapshot).await;
|
||||
if let Some(snapshot) = pending.take()
|
||||
&& let Err(error) = Self::write_json(&state_path, &snapshot).await
|
||||
{
|
||||
tracing::warn!(
|
||||
?error,
|
||||
?state_path,
|
||||
"Failed to flush resources state"
|
||||
);
|
||||
}
|
||||
let _ = done.send(());
|
||||
}
|
||||
None => {
|
||||
if let Some(snapshot) = pending.take() {
|
||||
Self::write_json(&state_path, &snapshot).await;
|
||||
if let Some(snapshot) = pending.take()
|
||||
&& let Err(error) = Self::write_json(&state_path, &snapshot).await
|
||||
{
|
||||
tracing::warn!(
|
||||
?error,
|
||||
?state_path,
|
||||
"Failed to flush resources state"
|
||||
);
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = debounce.tick() => {
|
||||
if let Some(snapshot) = pending.take() {
|
||||
Self::write_json(&state_path, &snapshot).await;
|
||||
if let Some(snapshot) = pending.take()
|
||||
&& let Err(error) = Self::write_json(&state_path, &snapshot).await
|
||||
{
|
||||
tracing::warn!(
|
||||
?error,
|
||||
?state_path,
|
||||
"Failed to save resources state"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn write_json(path: &Path, value: &serde_json::Value) {
|
||||
match serde_json::to_string_pretty(value) {
|
||||
Ok(json) => {
|
||||
let tmp_path = path.with_extension("json.tmp");
|
||||
if let Err(e) = tokio::fs::write(&tmp_path, json.as_bytes()).await {
|
||||
tracing::warn!("Failed to write resources state to {:?}: {}", tmp_path, e);
|
||||
return;
|
||||
}
|
||||
// Guard: if a previous bug left a directory at `path`, remove it
|
||||
// so the atomic rename can succeed.
|
||||
if path.is_dir() {
|
||||
tracing::warn!(
|
||||
"Resources state path {:?} is a directory — removing before write",
|
||||
path
|
||||
);
|
||||
let _ = tokio::fs::remove_dir_all(path).await;
|
||||
}
|
||||
if let Err(e) = tokio::fs::rename(&tmp_path, path).await {
|
||||
tracing::warn!(
|
||||
"Failed to rename resources state {:?} -> {:?}: {}",
|
||||
tmp_path,
|
||||
path,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to serialize resources state: {}", e);
|
||||
}
|
||||
async fn write_json(path: &Path, value: &serde_json::Value) -> io::Result<()> {
|
||||
let (tmp_path, json) = Self::prepare_write(path, value)?;
|
||||
tokio::fs::write(&tmp_path, json).await?;
|
||||
Self::replace_state_path(path, &tmp_path).await
|
||||
}
|
||||
|
||||
async fn write_json_durable(path: &Path, value: &serde_json::Value) -> io::Result<()> {
|
||||
let (tmp_path, json) = Self::prepare_write(path, value)?;
|
||||
let result = async {
|
||||
let mut file = tokio::fs::File::create(&tmp_path).await?;
|
||||
file.write_all(&json).await?;
|
||||
file.sync_all().await?;
|
||||
drop(file);
|
||||
Self::publish_durable(path, &tmp_path).await
|
||||
}
|
||||
.await;
|
||||
Self::cleanup_temp_on_error(&tmp_path, result).await
|
||||
}
|
||||
|
||||
async fn cleanup_temp_on_error(tmp_path: &Path, result: io::Result<()>) -> io::Result<()> {
|
||||
if result.is_err() {
|
||||
let _ = tokio::fs::remove_file(tmp_path).await;
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
fn prepare_write(path: &Path, value: &serde_json::Value) -> io::Result<(PathBuf, Vec<u8>)> {
|
||||
let json = serde_json::to_vec_pretty(value)
|
||||
.map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
|
||||
Ok((path.with_extension("json.tmp"), json))
|
||||
}
|
||||
|
||||
async fn replace_state_path(path: &Path, tmp_path: &Path) -> io::Result<()> {
|
||||
if path.is_dir() {
|
||||
tracing::warn!(
|
||||
"Resources state path {:?} is a directory — removing before write",
|
||||
path
|
||||
);
|
||||
tokio::fs::remove_dir_all(path).await?;
|
||||
}
|
||||
tokio::fs::rename(tmp_path, path).await
|
||||
}
|
||||
|
||||
#[cfg(not(windows))]
|
||||
async fn publish_durable(path: &Path, tmp_path: &Path) -> io::Result<()> {
|
||||
Self::replace_state_path(path, tmp_path).await?;
|
||||
let parent = path.parent().ok_or_else(|| {
|
||||
io::Error::new(io::ErrorKind::InvalidInput, "resources state has no parent")
|
||||
})?;
|
||||
tokio::fs::File::open(parent).await?.sync_all().await
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
async fn publish_durable(path: &Path, tmp_path: &Path) -> io::Result<()> {
|
||||
use windows::Win32::Storage::FileSystem::MoveFileExW;
|
||||
use windows::core::PCWSTR;
|
||||
if path.is_dir() {
|
||||
tokio::fs::remove_dir_all(path).await?;
|
||||
}
|
||||
let from = Self::windows_extended_path(tmp_path)?;
|
||||
let to = Self::windows_extended_path(path)?;
|
||||
unsafe {
|
||||
MoveFileExW(
|
||||
PCWSTR(from.as_ptr()),
|
||||
PCWSTR(to.as_ptr()),
|
||||
Self::WINDOWS_MOVE_FLAGS,
|
||||
)
|
||||
}
|
||||
.map_err(io::Error::other)
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
const WINDOWS_MOVE_FLAGS: windows::Win32::Storage::FileSystem::MOVE_FILE_FLAGS =
|
||||
windows::Win32::Storage::FileSystem::MOVE_FILE_FLAGS(1 | 8);
|
||||
|
||||
#[cfg(windows)]
|
||||
fn windows_extended_path(path: &Path) -> io::Result<Vec<u16>> {
|
||||
use std::os::windows::ffi::OsStrExt;
|
||||
let path = std::path::absolute(path)?;
|
||||
let mut wide = path.as_os_str().encode_wide().collect::<Vec<_>>();
|
||||
if wide.contains(&0) {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::InvalidInput,
|
||||
"path contains NUL",
|
||||
));
|
||||
}
|
||||
let unc = wide.starts_with(&[92, 92]);
|
||||
let mut result = if unc { r"\\?\UNC\" } else { r"\\?\" }
|
||||
.encode_utf16()
|
||||
.collect::<Vec<_>>();
|
||||
if unc {
|
||||
wide.drain(..2);
|
||||
}
|
||||
result.extend(wide);
|
||||
result.push(0);
|
||||
Ok(result)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -337,4 +490,139 @@ mod tests {
|
|||
// Should have "state" category with "grok_build.WebCitation" key
|
||||
assert!(parsed["state"]["grok_build.WebCitation"].is_object());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_and_flush_supersedes_older_pending_snapshot() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let state_path = dir.path().join("resources_state.json");
|
||||
let persistence = ResourcesPersistence::new(state_path.clone());
|
||||
persistence.flush().await;
|
||||
|
||||
let mut resources = Resources::new();
|
||||
resources.register_state::<WebCitationCounter>();
|
||||
resources
|
||||
.get_or_default::<State<WebCitationCounter>>()
|
||||
.counter = 1;
|
||||
persistence.save(&resources);
|
||||
|
||||
resources
|
||||
.get_or_default::<State<WebCitationCounter>>()
|
||||
.counter = 2;
|
||||
persistence
|
||||
.save_and_flush(resources.serialize())
|
||||
.await
|
||||
.unwrap();
|
||||
persistence.flush().await;
|
||||
|
||||
let content = std::fs::read_to_string(state_path).unwrap();
|
||||
let parsed: serde_json::Value = serde_json::from_str(&content).unwrap();
|
||||
assert_eq!(parsed["state"]["grok_build.WebCitation"]["counter"], 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_and_flush_error_can_be_retried() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let parent = dir.path().join("missing");
|
||||
let state_path = parent.join("resources_state.json");
|
||||
let persistence = ResourcesPersistence::new(state_path.clone());
|
||||
|
||||
let mut resources = Resources::new();
|
||||
resources.register_state::<WebCitationCounter>();
|
||||
resources
|
||||
.get_or_default::<State<WebCitationCounter>>()
|
||||
.counter = 7;
|
||||
let snapshot = resources.serialize();
|
||||
|
||||
assert!(persistence.save_and_flush(snapshot.clone()).await.is_err());
|
||||
|
||||
std::fs::create_dir(parent).unwrap();
|
||||
persistence.save_and_flush(snapshot).await.unwrap();
|
||||
|
||||
let content = std::fs::read_to_string(state_path).unwrap();
|
||||
let parsed: serde_json::Value = serde_json::from_str(&content).unwrap();
|
||||
assert_eq!(parsed["state"]["grok_build.WebCitation"]["counter"], 7);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn enqueued_acknowledged_save_precedes_a_newer_snapshot() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let state_path = dir.path().join("resources_state.json");
|
||||
let persistence = ResourcesPersistence::new(state_path.clone());
|
||||
persistence.flush().await;
|
||||
|
||||
let mut resources = Resources::new();
|
||||
resources.register_state::<WebCitationCounter>();
|
||||
resources
|
||||
.get_or_default::<State<WebCitationCounter>>()
|
||||
.counter = 1;
|
||||
let acknowledgement = persistence
|
||||
.enqueue_save_and_flush(resources.serialize())
|
||||
.unwrap();
|
||||
|
||||
resources
|
||||
.get_or_default::<State<WebCitationCounter>>()
|
||||
.counter = 2;
|
||||
persistence.save(&resources);
|
||||
|
||||
ResourcesPersistence::await_save_and_flush(acknowledgement)
|
||||
.await
|
||||
.unwrap();
|
||||
persistence.flush().await;
|
||||
|
||||
let content = std::fs::read_to_string(state_path).unwrap();
|
||||
let parsed: serde_json::Value = serde_json::from_str(&content).unwrap();
|
||||
assert_eq!(parsed["state"]["grok_build.WebCitation"]["counter"], 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn post_create_failure_cleans_temp_and_allows_retry() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let tmp_path = dir.path().join("resources_state.json.tmp");
|
||||
std::fs::write(&tmp_path, "partial").unwrap();
|
||||
let error = io::Error::other("publish failed");
|
||||
let returned = ResourcesPersistence::cleanup_temp_on_error(&tmp_path, Err(error))
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(returned.to_string(), "publish failed");
|
||||
assert!(!tmp_path.exists());
|
||||
std::fs::write(&tmp_path, "retry").unwrap();
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
#[tokio::test]
|
||||
async fn windows_publish_supports_long_paths_and_legacy_directory() {
|
||||
use windows::Win32::Storage::FileSystem::{
|
||||
MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH,
|
||||
};
|
||||
assert_eq!(
|
||||
ResourcesPersistence::WINDOWS_MOVE_FLAGS.0,
|
||||
MOVEFILE_REPLACE_EXISTING.0 | MOVEFILE_WRITE_THROUGH.0
|
||||
);
|
||||
let long = PathBuf::from(format!(r"C:\{}", "long\\".repeat(60)));
|
||||
let wide = ResourcesPersistence::windows_extended_path(&long).unwrap();
|
||||
assert!(wide.len() > 260 && String::from_utf16_lossy(&wide).starts_with(r"\\?\"));
|
||||
let unc =
|
||||
ResourcesPersistence::windows_extended_path(Path::new(r"\\server\share\state.json"))
|
||||
.unwrap();
|
||||
assert!(String::from_utf16_lossy(&unc).starts_with(r"\\?\UNC\"));
|
||||
assert!(ResourcesPersistence::windows_extended_path(Path::new("bad\0path")).is_err());
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let target = dir.path().join("state.json");
|
||||
std::fs::create_dir(&target).unwrap();
|
||||
let temp = dir.path().join("state.json.tmp");
|
||||
std::fs::write(&temp, "new").unwrap();
|
||||
ResourcesPersistence::publish_durable(&target, &temp)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(std::fs::read_to_string(target).unwrap(), "new");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn noop_save_and_flush_acknowledges_without_writing() {
|
||||
ResourcesPersistence::noop()
|
||||
.save_and_flush(serde_json::json!({"state": {}}))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2189,6 +2189,7 @@ mod tests {
|
|||
]
|
||||
.into_iter()
|
||||
.map(|id| ToolConfig::from_id(format!("GrokBuild:{id}")))
|
||||
.chain(std::iter::empty::<ToolConfig>())
|
||||
.collect(),
|
||||
behavior_preset: None,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -345,7 +345,16 @@ mod tests {
|
|||
let mut expected: serde_json::Value =
|
||||
serde_json::from_str(tool_meta_json_schema_str()).expect("checked-in schema parses");
|
||||
if let Some(values) = expected["definitions"]["ToolNamespace"]["enum"].as_array_mut() {
|
||||
values.retain(|v| v != "cursor");
|
||||
use std::collections::HashSet;
|
||||
use strum::IntoEnumIterator;
|
||||
let compiled: HashSet<String> = ToolNamespace::iter()
|
||||
.filter_map(|ns| {
|
||||
serde_json::to_value(ns)
|
||||
.ok()
|
||||
.and_then(|v| v.as_str().map(str::to_owned))
|
||||
})
|
||||
.collect();
|
||||
values.retain(|v| matches!(v.as_str(), Some(s) if compiled.contains(s)));
|
||||
}
|
||||
let expected = format!("{}\n", serde_json::to_string_pretty(&expected).unwrap());
|
||||
assert_eq!(
|
||||
|
|
|
|||
Loading…
Reference in a new issue