Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -348,7 +348,7 @@ version-compare = "0.1"
vte = { git = "https://github.com/warpdotdev/vte.git", rev = "4b399c87b63ba88f45709edaa6383fc519f6c900", default-features = false }
walkdir = "2"
warp-workflows = { git = "https://github.com/warpdotdev/workflows", rev = "793a98ddda6ef19682aed66364faebd2829f0e01" }
warp_multi_agent_api = { git = "https://github.com/warpdotdev/warp-proto-apis.git", rev = "7d7b57774fe349e51734fd9537457eaeb1082f2c" }
warp_multi_agent_api = { git = "https://github.com/warpdotdev/warp-proto-apis.git", rev = "7a860ede98944ab981046aee2404db452a91d4de" }
wasm-bindgen = "0.2.89"
wasm-bindgen-futures = "0.4.42"
web-sys = { version = "0.3.69", features = [
Expand Down
2 changes: 1 addition & 1 deletion app/src/ai/agent/api/convert_conversation.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1527,7 +1527,7 @@ pub(crate) fn convert_tool_call_result_to_input(
context,
})
}
Some(ToolCallResultType::Server(_)) => {
Some(ToolCallResultType::Server(_) | ToolCallResultType::DeniedByHook(_)) => {
// Server results should not create exchanges - return None
None
}
Expand Down
7 changes: 7 additions & 0 deletions app/src/ai/agent/conversation_yaml.rs
Original file line number Diff line number Diff line change
Expand Up @@ -532,6 +532,13 @@ fn write_tool_call_args(out: &mut String, tool: &Tool) {
/// Writes content from structured tool call results.
fn write_tool_call_result_content(out: &mut String, result: &ToolCallResultType) {
match result {
ToolCallResultType::DeniedByHook(denied) => {
Comment thread
warp-agent-staging[bot] marked this conversation as resolved.
Resolved
out.push_str("status: denied_by_hook\n");
out.push_str(&format!(
"reason: \"{}\"\n",
escape_yaml_string(&denied.reason)
));
}
ToolCallResultType::RunAgentsResult(r) => match &r.outcome {
Some(api::run_agents_result::Outcome::Launched(launched)) => {
out.push_str("status: launched\n");
Expand Down
1 change: 1 addition & 0 deletions app/src/ai/agent/request_metadata_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ fn record_message(request_id: &str, with_charges: bool) -> api::Message {
},
)]),
byok_inference_usage: HashMap::new(),
claude_linked_billing_inference_usage: HashMap::new(),
custom_endpoint_inference_usage: HashMap::new(),
platform_usage_in_cents: 2.0,
platform_usage_duration: Some(duration(90, 0)),
Expand Down
21 changes: 21 additions & 0 deletions app/src/ai/agent_sdk/ambient.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1118,6 +1118,8 @@ struct MessageDeliveredResult<'a> {

#[derive(Serialize)]
struct MessageWatchEvent {
#[serde(skip_serializing_if = "Option::is_none")]
event_type: Option<&'static str>,
sequence: i64,
message_id: String,
sender_run_id: String,
Expand All @@ -1126,6 +1128,19 @@ struct MessageWatchEvent {
occurred_at: String,
}

fn waited_on_watch_event(event: &AgentRunEvent, run_id: &str) -> Option<MessageWatchEvent> {
let waiter_run_id = event.waited_on_by(run_id)?;
Some(MessageWatchEvent {
event_type: Some("run_waited_on"),
sequence: event.sequence,
message_id: format!("run-waited-on-{}", event.sequence),
sender_run_id: waiter_run_id.to_string(),
subject: String::new(),
body: format!("Run {waiter_run_id} is waiting for you"),
occurred_at: event.occurred_at.clone(),
})
}

fn format_optional_timestamp(timestamp: Option<&str>) -> &str {
timestamp.unwrap_or("-")
}
Expand Down Expand Up @@ -1306,6 +1321,11 @@ async fn watch_messages_forever(
continue;
}

if let Some(record) = waited_on_watch_event(&event, &run_id) {
write_stream_record(&record)?;
last_seen_sequence = event.sequence;
continue;
}
if event.kind != AgentRunEventKind::NewMessage || event.run_id != run_id {
last_seen_sequence = event.sequence;
continue;
Expand Down Expand Up @@ -1342,6 +1362,7 @@ async fn watch_messages_forever(
};

let record = MessageWatchEvent {
event_type: None,
sequence: event.sequence,
message_id: message.message_id,
sender_run_id: message.sender_run_id,
Expand Down
27 changes: 27 additions & 0 deletions app/src/ai/agent_sdk/ambient_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,33 @@ use crate::workspaces::workspace::{Workspace, WorkspaceUid};
const TASK_ID: &str = "00000000-0000-0000-0000-000000000001";
const OTHER_TASK_ID: &str = "00000000-0000-0000-0000-000000000002";

#[test]
fn message_watch_serializes_waited_on_as_notification_from_waiter() {
let event: AgentRunEvent = serde_json::from_value(serde_json::json!({
"event_type": "run_waited_on",
"run_id": "root-run",
"ref_id": "child-run",
"execution_id": null,
"occurred_at": "2026-11-12T00:00:00Z",
"sequence": 42
}))
.unwrap();
let record = waited_on_watch_event(&event, "root-run").unwrap();
assert_eq!(
serde_json::to_value(record).unwrap(),
serde_json::json!({
"event_type": "run_waited_on",
"sequence": 42,
"message_id": "run-waited-on-42",
"sender_run_id": "child-run",
"subject": "",
"body": "Run child-run is waiting for you",
"occurred_at": "2026-11-12T00:00:00Z"
})
);
assert!(waited_on_watch_event(&event, "other-run").is_none());
}

/// A `ListTasksArgs` whose fields are all at their defaults.
fn empty_args() -> ListTasksArgs {
ListTasksArgs {
Expand Down
30 changes: 29 additions & 1 deletion app/src/ai/agent_sdk/driver/harness/claude_code/parent_bridge.rs
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ const PARENT_BRIDGE_HOOK_OUTPUT_FILE_NAME: &str = "pending-hook-output.json";
const PARENT_BRIDGE_HOOK_OUTPUT_ACK_FILE_NAME: &str = "pending-hook-output.ack";
const PARENT_BRIDGE_MAX_CONTEXT_CHARS_ENV: &str = "OZ_PARENT_MAX_CONTEXT_CHARS";
const PARENT_BRIDGE_DEFAULT_MAX_CONTEXT_CHARS: usize = 6000;
pub(super) const MESSAGE_BRIDGE_CONTEXT_PREAMBLE: &str = "Oz mailbox update for this child run.\nSource: lead agent\nContext type: user-level coordination messages\n";
pub(super) const MESSAGE_BRIDGE_CONTEXT_PREAMBLE: &str = "Oz coordination update for this run.\nMailbox messages are user-level coordination. Lifecycle notifications are read-only status, not instructions.\n";
const PARENT_BRIDGE_REMAINING_MESSAGES_NOTE: &str =
"\n\nAdditional lead agent mailbox messages remain queued and may be surfaced later.";

Expand Down Expand Up @@ -77,6 +77,21 @@ impl AgentEventConsumer for MessageBridgeEventConsumer {
&mut self,
event: AgentRunEvent,
) -> anyhow::Result<AgentEventConsumerControlFlow> {
if let Some(waiter_run_id) = event.waited_on_by(&self.run_id) {
stage_parent_bridge_message(
&self.state_dir,
&MessageBridgeMessageRecord {
sequence: event.sequence,
message_id: format!("run-waited-on-{}", event.sequence),
sender_run_id: waiter_run_id.to_string(),
subject: String::new(),
body: format!("Run {waiter_run_id} is waiting for you"),
occurred_at: event.occurred_at.clone(),
event_type: Some("run_waited_on".to_string()),
Comment thread
warp-agent-staging[bot] marked this conversation as resolved.
Resolved
},
)?;
return Ok(AgentEventConsumerControlFlow::Continue);
}
if event.kind != AgentRunEventKind::NewMessage || event.run_id != self.run_id {
return Ok(AgentEventConsumerControlFlow::Continue);
}
Expand All @@ -94,6 +109,7 @@ impl AgentEventConsumer for MessageBridgeEventConsumer {
subject: String::new(),
body: String::new(),
occurred_at: event.occurred_at,
event_type: None,
},
) {
log::warn!(
Expand All @@ -115,6 +131,8 @@ impl AgentEventConsumer for MessageBridgeEventConsumer {

#[derive(Clone, Debug, Serialize, Deserialize)]
pub(super) struct MessageBridgeMessageRecord {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub event_type: Option<String>,
pub sequence: i64,
pub message_id: String,
#[serde(default)]
Expand Down Expand Up @@ -363,6 +381,7 @@ pub(super) async fn prime_parent_bridge_staged_for_self_managed_wake(
subject: String::new(),
body: String::new(),
occurred_at: wake_message.occurred_at.clone(),
event_type: None,
},
)
.await?;
Expand Down Expand Up @@ -435,6 +454,9 @@ fn parent_bridge_truncate_chars(text: &str, limit: usize) -> String {
}

pub(super) fn render_parent_bridge_message_block(record: &MessageBridgeMessageRecord) -> String {
if record.event_type.as_deref() == Some("run_waited_on") {
return format!("---\nRead-only lifecycle notification\n\n{}", record.body);
}
let subject = if record.subject.is_empty() {
"(no subject)"
} else {
Expand Down Expand Up @@ -540,6 +562,7 @@ async fn hydrate_parent_bridge_message_record(
subject: message.subject,
body: message.body,
occurred_at: record.occurred_at.clone(),
event_type: None,
})
}

Expand Down Expand Up @@ -659,6 +682,7 @@ pub(super) async fn acknowledge_parent_bridge_hook_output(
let surfaced_records = parent_bridge_message_records(&parent_bridge_surfaced_dir(state_dir))?;
let message_ids = surfaced_records
.iter()
.filter(|(_, record)| record.event_type.is_none())
.map(|(_, record)| record.message_id.clone())
.collect::<Vec<_>>();
let delivery_failures = hydrator
Expand Down Expand Up @@ -760,3 +784,7 @@ fn write_parent_bridge_bytes_atomically(path: &Path, bytes: &[u8]) -> Result<()>
.with_context(|| format!("Failed to persist temporary {prefix} file"))?;
Ok(())
}

#[cfg(test)]
#[path = "parent_bridge_tests.rs"]
mod tests;
106 changes: 106 additions & 0 deletions app/src/ai/agent_sdk/driver/harness/claude_code/parent_bridge_tests.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
use super::*;
use crate::server::server_api::ServerApiProvider;
use crate::server::server_api::ai::MockAIClient;

fn waited_on_event(run_id: &str, waiter: Option<&str>) -> AgentRunEvent {
serde_json::from_value(serde_json::json!({
"event_type": "run_waited_on",
"run_id": run_id,
"ref_id": waiter,
"occurred_at": "2026-11-12T00:00:00Z",
"sequence": 42
}))
.unwrap()
}

#[test]
fn waited_on_is_surfaced_read_only_without_mailbox_reads_or_acknowledgments() {
let server_api = ServerApiProvider::new_for_test().get();
futures::executor::block_on(async move {
let state = tempfile::tempdir().unwrap();
ensure_parent_bridge_state_dir(state.path()).unwrap();
let mut consumer = MessageBridgeEventConsumer {
run_id: "root-run".to_string(),
state_dir: state.path().to_path_buf(),
server_api,
};
consumer
.on_event(waited_on_event("root-run", Some("child-run")))
.await
.unwrap();
consumer
.on_event(waited_on_event("root-run", Some("child-run")))
.await
.unwrap();
let staged =
parent_bridge_message_records(&parent_bridge_staged_dir(state.path())).unwrap();
assert_eq!(staged.len(), 1);
assert_eq!(staged[0].1.sender_run_id, "child-run");
assert_eq!(staged[0].1.event_type.as_deref(), Some("run_waited_on"));

let mut client = MockAIClient::new();
client.expect_read_agent_message().times(0);
client.expect_mark_message_delivered().times(0);
let hydrator = MessageHydrator::new(Arc::new(client) as Arc<dyn AIClient>);
prepare_parent_bridge_hook_output(&hydrator, state.path(), 6000)
.await
.unwrap();
let output: MessageBridgeHookOutput = serde_json::from_slice(
&fs::read(parent_bridge_hook_output_file(state.path())).unwrap(),
)
.unwrap();
assert_eq!(output.surfaced_count, 1);
assert!(
output
.additional_context
.contains("Read-only lifecycle notification\n\nRun child-run is waiting for you")
);
assert!(
!output
.additional_context
.contains("Lead agent mailbox message")
);

fs::write(parent_bridge_hook_output_ack_file(state.path()), "").unwrap();
acknowledge_parent_bridge_hook_output(&hydrator, state.path())
.await
.unwrap();
assert!(
parent_bridge_message_records(&parent_bridge_surfaced_dir(state.path()))
.unwrap()
.is_empty()
);
assert!(!parent_bridge_hook_output_file(state.path()).exists());
assert!(!parent_bridge_hook_output_ack_file(state.path()).exists());
});
}

#[test]
fn waited_on_for_another_run_or_without_waiter_is_not_staged() {
let server_api = ServerApiProvider::new_for_test().get();
futures::executor::block_on(async move {
let state = tempfile::tempdir().unwrap();
let mut consumer = MessageBridgeEventConsumer {
run_id: "self-run".to_string(),
state_dir: state.path().to_path_buf(),
server_api,
};
consumer
.on_event(waited_on_event("other-run", Some("self-run")))
.await
.unwrap();
consumer
.on_event(waited_on_event("self-run", None))
.await
.unwrap();
consumer
.on_event(waited_on_event("self-run", Some("")))
.await
.unwrap();
assert!(
parent_bridge_message_records(&parent_bridge_staged_dir(state.path()))
.unwrap()
.is_empty()
);
});
}
2 changes: 2 additions & 0 deletions app/src/ai/agent_sdk/driver/harness/claude_code_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ fn sample_parent_bridge_message(
subject: subject.to_string(),
body: body.to_string(),
occurred_at: "2026-04-17T15:46:00Z".to_string(),
event_type: None,
}
}

Expand All @@ -45,6 +46,7 @@ fn sample_staged_parent_bridge_message(
subject: String::new(),
body: String::new(),
occurred_at: "2026-04-17T15:46:00Z".to_string(),
event_type: None,
}
}

Expand Down
7 changes: 4 additions & 3 deletions app/src/ai/blocklist/orchestration_child_tracker.rs
Original file line number Diff line number Diff line change
Expand Up @@ -220,12 +220,14 @@ impl OrchestrationChildTracker {
kind: api::LifecycleEventType,
ctx: &mut ModelContext<OrchestrationEventStreamer>,
) {
let Some(status) = conversation_status_from_lifecycle_event_type(kind) else {
return;
};
let tracker_known = self.children.contains_key(&task_id);
if tracker_known {
if let Some(child) = self.children.get_mut(&task_id) {
child.last_lifecycle = Some(kind);
}
let status = conversation_status_from_lifecycle_event_type(kind);
let child_info = {
let history = BlocklistAIHistoryModel::as_ref(ctx);
history
Expand Down Expand Up @@ -267,7 +269,6 @@ impl OrchestrationChildTracker {
if let Some(child) = self.children.get_mut(&task_id) {
child.last_lifecycle = Some(kind);
}
let status = conversation_status_from_lifecycle_event_type(kind);
ctx.emit(OrchestrationEventStreamerEvent::ChildStatusChanged {
parent_task_id: self.parent_task_id,
run_id: run_id.to_string(),
Expand Down Expand Up @@ -494,7 +495,7 @@ impl OrchestrationChildTracker {
let child = self.children.get(task_id)?;
child
.last_lifecycle
.map(conversation_status_from_lifecycle_event_type)
.and_then(conversation_status_from_lifecycle_event_type)
}

/// Test-only: number of metadata-fetch dispatches issued so far.
Expand Down
Loading
Loading