Improve session abort handling, emit tool output deltas, and tighten validation

- Break out of streaming loop on abort and drop the stream before emitting
  SessionEnd to properly cancel the HTTP connection
- Emit ToolCallOutputDelta events for tool call results in both sequential
  and parallel execution paths
- Retry on StageStatus::Fail in addition to Retry in pipeline engine
- Set preferred_label on WaitHumanHandler choice outcomes
- Enforce exactly one terminal node in pipeline validation
- Downgrade unreachable node diagnostic from Error to Warning
- Update context window test to match 1M token limit

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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 { .. }))));
     }
-}
+}

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));
     }
-}
+}

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"))
         );
     }
-}
+}

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);
     }
-}
+}

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));
     }
-}
+}