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,12 +317,19 @@ impl Session {
 
                 // Check abort flag between chunks
                 if self.abort_flag.load(Ordering::SeqCst) {
+                    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 {
@@ -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,

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 {

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(),

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,

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