mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-07 08:27:12 +00:00
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>
This commit is contained in:
parent
834b7f6160
commit
ceaa9bf560
5 changed files with 49 additions and 14 deletions
|
|
@ -317,13 +317,20 @@ impl Session {
|
||||||
|
|
||||||
// Check abort flag between chunks
|
// Check abort flag between chunks
|
||||||
if self.abort_flag.load(Ordering::SeqCst) {
|
if self.abort_flag.load(Ordering::SeqCst) {
|
||||||
self.state = SessionState::Closed;
|
break;
|
||||||
self.event_emitter
|
|
||||||
.emit(EventKind::SessionEnd, self.id.clone(), EventData::Empty);
|
|
||||||
return Err(AgentError::Aborted);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 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(|| {
|
let response = accumulator.response().cloned().ok_or_else(|| {
|
||||||
AgentError::Llm(SdkError::Stream {
|
AgentError::Llm(SdkError::Stream {
|
||||||
message: "Stream ended without a Finish event".into(),
|
message: "Stream ended without a Finish event".into(),
|
||||||
|
|
@ -494,6 +501,14 @@ impl Session {
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
|
self.event_emitter.emit(
|
||||||
|
EventKind::ToolCallOutputDelta,
|
||||||
|
self.id.clone(),
|
||||||
|
EventData::TextDelta {
|
||||||
|
delta: result.content.to_string(),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
self.event_emitter.emit(
|
self.event_emitter.emit(
|
||||||
EventKind::ToolCallEnd,
|
EventKind::ToolCallEnd,
|
||||||
self.id.clone(),
|
self.id.clone(),
|
||||||
|
|
@ -551,6 +566,14 @@ impl Session {
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
|
emitter.emit(
|
||||||
|
EventKind::ToolCallOutputDelta,
|
||||||
|
session_id.clone(),
|
||||||
|
EventData::TextDelta {
|
||||||
|
delta: result.content.to_string(),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
emitter.emit(
|
emitter.emit(
|
||||||
EventKind::ToolCallEnd,
|
EventKind::ToolCallEnd,
|
||||||
session_id,
|
session_id,
|
||||||
|
|
@ -1671,4 +1694,4 @@ mod tests {
|
||||||
let result = session.process_input("Hello").await;
|
let result = session.process_input("Hello").await;
|
||||||
assert!(matches!(result, Err(AgentError::Llm(SdkError::Stream { .. }))));
|
assert!(matches!(result, Err(AgentError::Llm(SdkError::Stream { .. }))));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -583,11 +583,10 @@ impl PipelineEngine {
|
||||||
match outcome.status {
|
match outcome.status {
|
||||||
StageStatus::Success
|
StageStatus::Success
|
||||||
| StageStatus::PartialSuccess
|
| StageStatus::PartialSuccess
|
||||||
| StageStatus::Fail
|
|
||||||
| StageStatus::Skipped => {
|
| StageStatus::Skipped => {
|
||||||
return Ok((outcome, attempt));
|
return Ok((outcome, attempt));
|
||||||
}
|
}
|
||||||
StageStatus::Retry => {
|
StageStatus::Fail | StageStatus::Retry => {
|
||||||
if attempt < policy.max_attempts {
|
if attempt < policy.max_attempts {
|
||||||
let delay = policy.backoff.delay_for_attempt(attempt);
|
let delay = policy.backoff.delay_for_attempt(attempt);
|
||||||
self.emitter.emit(&PipelineEvent::StageRetrying {
|
self.emitter.emit(&PipelineEvent::StageRetrying {
|
||||||
|
|
@ -2378,4 +2377,4 @@ mod tests {
|
||||||
assert!(result.is_err());
|
assert!(result.is_err());
|
||||||
assert!(matches!(result.unwrap_err(), AttractorError::Cancelled));
|
assert!(matches!(result.unwrap_err(), AttractorError::Cancelled));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -231,6 +231,7 @@ impl Handler for WaitHumanHandler {
|
||||||
|
|
||||||
fn make_choice_outcome(key: &str, label: &str, to: &str) -> Outcome {
|
fn make_choice_outcome(key: &str, label: &str, to: &str) -> Outcome {
|
||||||
let mut outcome = Outcome::success();
|
let mut outcome = Outcome::success();
|
||||||
|
outcome.preferred_label = Some(label.to_string());
|
||||||
outcome.suggested_next_ids = vec![to.to_string()];
|
outcome.suggested_next_ids = vec![to.to_string()];
|
||||||
outcome.context_updates.insert(
|
outcome.context_updates.insert(
|
||||||
"human.gate.selected".to_string(),
|
"human.gate.selected".to_string(),
|
||||||
|
|
@ -410,4 +411,4 @@ mod tests {
|
||||||
Some(&serde_json::json!("custom input"))
|
Some(&serde_json::json!("custom input"))
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -94,12 +94,24 @@ impl LintRule for TerminalNodeRule {
|
||||||
return vec![Diagnostic {
|
return vec![Diagnostic {
|
||||||
rule: self.name().to_string(),
|
rule: self.name().to_string(),
|
||||||
severity: Severity::Error,
|
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,
|
node_id: None,
|
||||||
edge: None,
|
edge: None,
|
||||||
fix: Some("Add a node with shape=Msquare or id 'exit'/'end'".to_string()),
|
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()
|
Vec::new()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -143,7 +155,7 @@ impl LintRule for ReachabilityRule {
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|node_id| Diagnostic {
|
.map(|node_id| Diagnostic {
|
||||||
rule: self.name().to_string(),
|
rule: self.name().to_string(),
|
||||||
severity: Severity::Error,
|
severity: Severity::Warning,
|
||||||
message: format!("Node '{node_id}' is not reachable from the start node"),
|
message: format!("Node '{node_id}' is not reachable from the start node"),
|
||||||
node_id: Some(node_id.to_string()),
|
node_id: Some(node_id.to_string()),
|
||||||
edge: None,
|
edge: None,
|
||||||
|
|
@ -1222,4 +1234,4 @@ mod tests {
|
||||||
assert_eq!(d.len(), 1);
|
assert_eq!(d.len(), 1);
|
||||||
assert_eq!(d[0].severity, Severity::Error);
|
assert_eq!(d[0].severity, Severity::Error);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -51,7 +51,7 @@ mod tests {
|
||||||
assert!(info.supports_tools);
|
assert!(info.supports_tools);
|
||||||
assert!(info.supports_vision);
|
assert!(info.supports_vision);
|
||||||
assert!(info.supports_reasoning);
|
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));
|
assert_eq!(info.max_output, Some(128_000));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -132,4 +132,4 @@ mod tests {
|
||||||
let sonnet = get_model_info("claude-sonnet-4-5").unwrap();
|
let sonnet = get_model_info("claude-sonnet-4-5").unwrap();
|
||||||
assert_eq!(sonnet.input_cost_per_million, Some(3.0));
|
assert_eq!(sonnet.input_cost_per_million, Some(3.0));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Loading…
Add table
Reference in a new issue