diff --git a/crates/agent/src/session.rs b/crates/agent/src/session.rs index fc92aad31..53d93f835 100644 --- a/crates/agent/src/session.rs +++ b/crates/agent/src/session.rs @@ -317,13 +317,20 @@ impl Session { // Check abort flag between chunks if self.abort_flag.load(Ordering::SeqCst) { - self.state = SessionState::Closed; - self.event_emitter - .emit(EventKind::SessionEnd, self.id.clone(), EventData::Empty); - return Err(AgentError::Aborted); + break; } } + // If aborted during streaming, drop the stream to cancel the HTTP + // connection, then emit SessionEnd before returning. + if self.abort_flag.load(Ordering::SeqCst) { + drop(event_stream); + self.state = SessionState::Closed; + self.event_emitter + .emit(EventKind::SessionEnd, self.id.clone(), EventData::Empty); + return Err(AgentError::Aborted); + } + let response = accumulator.response().cloned().ok_or_else(|| { AgentError::Llm(SdkError::Stream { message: "Stream ended without a Finish event".into(), @@ -494,6 +501,14 @@ impl Session { ) .await; + self.event_emitter.emit( + EventKind::ToolCallOutputDelta, + self.id.clone(), + EventData::TextDelta { + delta: result.content.to_string(), + }, + ); + self.event_emitter.emit( EventKind::ToolCallEnd, self.id.clone(), @@ -551,6 +566,14 @@ impl Session { ) .await; + emitter.emit( + EventKind::ToolCallOutputDelta, + session_id.clone(), + EventData::TextDelta { + delta: result.content.to_string(), + }, + ); + emitter.emit( EventKind::ToolCallEnd, session_id, @@ -1671,4 +1694,4 @@ mod tests { let result = session.process_input("Hello").await; assert!(matches!(result, Err(AgentError::Llm(SdkError::Stream { .. })))); } -} +} \ No newline at end of file diff --git a/crates/attractor/src/engine.rs b/crates/attractor/src/engine.rs index ea9ebbcb3..68316bc88 100644 --- a/crates/attractor/src/engine.rs +++ b/crates/attractor/src/engine.rs @@ -583,11 +583,10 @@ impl PipelineEngine { match outcome.status { StageStatus::Success | StageStatus::PartialSuccess - | StageStatus::Fail | StageStatus::Skipped => { return Ok((outcome, attempt)); } - StageStatus::Retry => { + StageStatus::Fail | StageStatus::Retry => { if attempt < policy.max_attempts { let delay = policy.backoff.delay_for_attempt(attempt); self.emitter.emit(&PipelineEvent::StageRetrying { @@ -2378,4 +2377,4 @@ mod tests { assert!(result.is_err()); assert!(matches!(result.unwrap_err(), AttractorError::Cancelled)); } -} +} \ No newline at end of file diff --git a/crates/attractor/src/handler/wait_human.rs b/crates/attractor/src/handler/wait_human.rs index 4dbd60ea0..2750db2fa 100644 --- a/crates/attractor/src/handler/wait_human.rs +++ b/crates/attractor/src/handler/wait_human.rs @@ -231,6 +231,7 @@ impl Handler for WaitHumanHandler { fn make_choice_outcome(key: &str, label: &str, to: &str) -> Outcome { let mut outcome = Outcome::success(); + outcome.preferred_label = Some(label.to_string()); outcome.suggested_next_ids = vec![to.to_string()]; outcome.context_updates.insert( "human.gate.selected".to_string(), @@ -410,4 +411,4 @@ mod tests { Some(&serde_json::json!("custom input")) ); } -} +} \ No newline at end of file diff --git a/crates/attractor/src/validation/rules.rs b/crates/attractor/src/validation/rules.rs index a902ebc48..971eb34fe 100644 --- a/crates/attractor/src/validation/rules.rs +++ b/crates/attractor/src/validation/rules.rs @@ -94,12 +94,24 @@ impl LintRule for TerminalNodeRule { return vec![Diagnostic { rule: self.name().to_string(), severity: Severity::Error, - message: "Pipeline must have at least one terminal node (shape=Msquare or id exit/end)".to_string(), + message: "Pipeline must have exactly one terminal node (shape=Msquare or id exit/end)".to_string(), node_id: None, edge: None, fix: Some("Add a node with shape=Msquare or id 'exit'/'end'".to_string()), }]; } + if terminal_count > 1 { + return vec![Diagnostic { + rule: self.name().to_string(), + severity: Severity::Error, + message: format!( + "Pipeline must have exactly one terminal node, found {terminal_count}" + ), + node_id: None, + edge: None, + fix: Some("Remove extra terminal nodes so exactly one remains".to_string()), + }]; + } Vec::new() } } @@ -143,7 +155,7 @@ impl LintRule for ReachabilityRule { .into_iter() .map(|node_id| Diagnostic { rule: self.name().to_string(), - severity: Severity::Error, + severity: Severity::Warning, message: format!("Node '{node_id}' is not reachable from the start node"), node_id: Some(node_id.to_string()), edge: None, @@ -1222,4 +1234,4 @@ mod tests { assert_eq!(d.len(), 1); assert_eq!(d[0].severity, Severity::Error); } -} +} \ No newline at end of file diff --git a/crates/llm/src/catalog.rs b/crates/llm/src/catalog.rs index 34d8316e6..912b7477b 100644 --- a/crates/llm/src/catalog.rs +++ b/crates/llm/src/catalog.rs @@ -51,7 +51,7 @@ mod tests { assert!(info.supports_tools); assert!(info.supports_vision); assert!(info.supports_reasoning); - assert_eq!(info.context_window, 200_000); + assert_eq!(info.context_window, 1_000_000); assert_eq!(info.max_output, Some(128_000)); } @@ -132,4 +132,4 @@ mod tests { let sonnet = get_model_info("claude-sonnet-4-5").unwrap(); assert_eq!(sonnet.input_cost_per_million, Some(3.0)); } -} +} \ No newline at end of file