codewhale-hq/Codewhale
1
0 Fork 0
mirror of https://github.com/codewhale-hq/Codewhale.git synced 2026-11-08 08:04:25 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

fix: retain completed parallel tool results before cancellation projection

Browse Source 
An actual registry tool finished successfully, but biased cancellation dropped its known result while artifact/media processing yielded inside FuturesUnordered and fabricated a cancelled fallback. Return the existing plan, start time and raw RichToolResult directly to the existing outcomes collector before projection. Preserve every output artifact/media guard, original digest, route identity and plan-index ordering; emit through the existing receipt primitive and settle completed data before cancelling remaining tasks. This deletes the parallel sender consumer and reuses the Engine event authority (11 reservation callers: 8 Strict, 3 Receipt). Strengthen the actual fixture to verify cancelled-peer metadata. Prior f877 run: 34 passed / 1 failed in 8.85s; all 13 unchanged goldens and exact activity outcome witness passed. Formatted/diff checked; combined requalification pending.

Signed-off-by: CodeWhale Bot <[email protected]>
This commit is contained in:
CodeWhale Bot committed 2026-09-30 05:50:26 -07:00
1 parent a0e9082391
commit f0a00cd367
2 files changed
+73 -72
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/tui/src/core/engine/tests.rs
+10
View File
@@ -759,6 +759,16 @@ async fn event_capacity_cancelled_parallel_tool_keeps_completed_span_and_call_wh
.filter(|(_, _, result)| result.as_ref().is_ok_and(|result| result.success))
.collect();
assert_eq!(succeeded.len(), 1);
let cancelled_peer = calls
.iter()
.find(|(_, _, result)| !result.as_ref().is_ok_and(|result| result.success))
.expect("cancelled peer has its own completion");
let peer = cancelled_peer
.2
.as_ref()
.expect("legacy cancelled peer result");
assert_eq!(peer.metadata.as_ref().unwrap()["cancelled"], true);
assert_eq!(peer.metadata.as_ref().unwrap()["cleanup_confirmed"], false);
assert_eq!(
succeeded[0].2.as_ref().unwrap().content,
"completed before cancellation"
crates/tui/src/core/engine/turn_loop.rs
+63 -72
View File
@@ -4399,6 +4399,10 @@ impl Engine {
.collect();
let mut tool_tasks = FuturesUnordered::new();
let shell_permits = Arc::new(tokio::sync::Semaphore::new(MAX_PARALLEL_SHELL_EXEC));
let session_id = self.session.id.clone();
let provider = self.api_provider;
let model = self.session.model.clone();
let route_limits = self.active_route_limits;
for plan in plans {
if let Some(result) = plan.guard_result.clone() {
let result = Ok(result);
@@ -4449,10 +4453,6 @@ impl Engine {
let lock = tool_exec_lock.clone();
let mcp_pool = mcp_pool.clone();
let tx_event = self.tx_event.clone();
let session_id = self.session.id.clone();
let provider = self.api_provider;
let model = self.session.model.clone();
let route_limits = self.active_route_limits;
let started_at = Instant::now();
let shell_permits = shell_permits.clone();
let workspace = self.session.workspace.clone();
@@ -4482,71 +4482,11 @@ impl Engine {
context_override,
)
.await;
let original_content_digest = result
.as_ref()
.ok()
.filter(|_| collect_fleet_evidence)
.and_then(|result| {
FleetDenialGuard::original_content_digest(
&plan.name,
&plan.input,
&result.result,
)
});
let result = preserve_tool_output_before_fanout(
result,
provider,
&model,
route_limits,
&session_id,
&plan.id,
&plan.name,
)
.await;
let result = match result {
Ok(rich) => Ok(super::tool_media::project(
rich,
&session_id,
&plan.id,
&plan.name,
)
.await),
Err(error) => Err(error),
};
let content_blocks = result
.as_ref()
.map(|result| result.content_blocks.clone())
.unwrap_or_default();
let legacy_result = result.map(RichToolResult::into_result);
if let Ok(permit) = super::streaming::reserve_event_capacity(
&tx_event,
Some(&cancel_token),
super::streaming::EventReservationPolicy::Receipt,
)
.await
{
permit.send(Event::ToolCallComplete {
model_call: plan.model_call.clone(),
id: plan.id.clone(),
name: plan.name.clone(),
result: legacy_result.clone(),
});
}
ToolExecOutcome {
model_call: plan.model_call.clone(),
index: plan.index,
id: plan.id,
name: plan.name,
input: plan.input,
started_at,
terminal: ToolExecutionOutcome::from_legacy(legacy_result),
content_blocks,
original_content_digest,
}
// Hand the known result to the existing collector
// before artifact/media processing can yield. Once
// execution finished, cancellation must not drop its
// success and fabricate a cancelled fallback.
(plan, started_at, result)
});
}
@@ -4559,9 +4499,60 @@ impl Engine {
break;
}
outcome = tool_tasks.next() => {
let Some(outcome) = outcome else { break; };
let index = outcome.index;
outcomes[index] = Some(outcome);
let Some((plan, started_at, result)) = outcome else { break; };
let original_content_digest = result
.as_ref()
.ok()
.filter(|_| collect_fleet_evidence)
.and_then(|result| {
FleetDenialGuard::original_content_digest(
&plan.name,
&plan.input,
&result.result,
)
});
let result = preserve_tool_output_before_fanout(
result,
provider,
&model,
route_limits,
&session_id,
&plan.id,
&plan.name,
)
.await;
let result = match result {
Ok(rich) => Ok(super::tool_media::project(
rich,
&session_id,
&plan.id,
&plan.name,
)
.await),
Err(error) => Err(error),
};
let content_blocks = result
.as_ref()
.map(|result| result.content_blocks.clone())
.unwrap_or_default();
let legacy_result = result.map(RichToolResult::into_result);
let _ = self.send_event(Event::ToolCallComplete {
model_call: plan.model_call.clone(),
id: plan.id.clone(),
name: plan.name.clone(),
result: legacy_result.clone(),
}).await;
outcomes[plan.index] = Some(ToolExecOutcome {
model_call: plan.model_call.clone(),
index: plan.index,
id: plan.id,
name: plan.name,
input: plan.input,
started_at,
terminal: ToolExecutionOutcome::from_legacy(legacy_result),
content_blocks,
original_content_digest,
});
}
}
}