Skip to content

fix(engine): keep turn and submission attribution on TurnComplete #7018

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Hmbown merged 3 commits into from
Oct 18, 2026
Merged

fix(engine): keep turn and submission attribution on TurnComplete #7018

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3f1b96f
fix(engine): keep turn and submission attribution on TurnComplete
asto18089 Oct 17, 2026
dffdfaa
Merge branch 'main' into up/turn-complete-attribution
Hmbown Oct 18, 2026
d9604d1
Merge branch 'main' into up/turn-complete-attribution
Hmbown Oct 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 58 additions & 2 deletions crates/protocol/src/event_msg.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -450,10 +450,19 @@ pub enum EventMsg {
TurnComplete {
thread_id: ThreadId,
session_id: SessionId,
/// The engine's `TurnComplete` carries no turn id; the emitter fills
/// it from the envelope when it knows it.
/// Actual started turn; absent when a submitted operation fails
/// before starting.
#[serde(default, skip_serializing_if = "Option::is_none")]
turn_id: Option<String>,
/// Host submission correlation, including preflight rejection. Runtime
/// self-starts do not carry a host token. Older producers default to
/// absent.
#[serde(default, skip_serializing_if = "Option::is_none")]
submission_id: Option<String>,
/// Manual compaction operation identity, never the surrounding host
/// submission.
#[serde(default, skip_serializing_if = "Option::is_none")]
compaction_id: Option<String>,
status: TurnOutcomeStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
error: Option<String>,
Expand Down Expand Up @@ -1195,6 +1204,8 @@ mod tests {
route,
},
EventMsg::TurnComplete {
submission_id: None,
compaction_id: None,
thread_id: t.clone(),
session_id: s.clone(),
turn_id: None,
Expand Down Expand Up @@ -1584,6 +1595,8 @@ mod tests {
#[test]
fn event_msg_roundtrip() {
let msg = EventMsg::TurnComplete {
submission_id: None,
compaction_id: None,
thread_id: ThreadId::new(),
session_id: SessionId::new(),
turn_id: Some("turn-1".into()),
Expand All @@ -1601,6 +1614,49 @@ mod tests {
assert!(json.contains(r#""status":"completed""#));
}

#[test]
fn turn_complete_ownership_is_additive_and_round_trips() {
for (turn_id, submission_id, compaction_id) in [
(Some("turn-1"), Some("sub-host-1"), None),
(None, Some("sub-rejected"), None),
(Some("autonomous-turn"), None, None),
(None, None, Some("compact-1")),
(None, None, None),
] {
let msg = EventMsg::TurnComplete {
thread_id: ThreadId::new(),
session_id: SessionId::new(),
turn_id: turn_id.map(str::to_string),
submission_id: submission_id.map(str::to_string),
compaction_id: compaction_id.map(str::to_string),
status: TurnOutcomeStatus::Failed,
error: Some("fixture rejection".into()),
usage: TokenUsage::default(),
parent_route_usage: None,
routed_usage_dropped_records: 0,
tool_catalog: None,
base_url: None,
};
let value = serde_json::to_value(&msg).unwrap();
for (key, expected) in [
("turn_id", turn_id),
("submission_id", submission_id),
("compaction_id", compaction_id),
] {
if let Some(expected) = expected {
assert_eq!(value[key], expected);
} else {
assert!(value.get(key).is_none(), "{key} must stay absent");
}
}
let back: EventMsg = serde_json::from_value(value).unwrap();
assert_eq!(
back, msg,
"legacy absent fields and owned receipts round-trip"
);
}
}

#[test]
fn text_channel_is_elided_on_the_wire() {
let msg = EventMsg::ResponseDelta {
Expand Down
3 changes: 3 additions & 0 deletions crates/tui/src/acp_server/runtime.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -665,6 +665,9 @@ mod tests {
engine
.tx_event
.send(Event::TurnComplete {
turn_id: None,
submission_id: None,
compaction_id: None,
usage: codewhale_models::Usage::default(),
parent_route_usage: codewhale_models::Usage::default(),
routed_usage_dropped_records: 0,
Expand Down
144 changes: 109 additions & 35 deletions crates/tui/src/core/engine.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2722,6 +2722,11 @@ impl Engine {
let pending_post_turn = self.reserve_post_turn_snapshot();
let status = terminal_turn_status_at_settlement(status, self.cancel_token.is_cancelled());
terminal_permit.send(Event::TurnComplete {
// The composer shell turn started above; it never carries a host
// submission envelope or compaction id.
turn_id: Some(turn_id.clone()),
submission_id: None,
compaction_id: None,
usage: Usage::default(),
parent_route_usage: Usage::default(),
routed_usage_dropped_records: 0,
Expand Down Expand Up @@ -4314,15 +4319,18 @@ impl Engine {
let route = match self.current_runtime_route() {
Ok(route) => route,
Err(err) => {
self.reject_edit_last_turn(ErrorEnvelope::new(
ErrorCategory::Authentication,
ErrorSeverity::Critical,
false,
"edit_last_turn_invalid_route",
format!(
"Cannot edit the last turn because its provider route is no longer valid: {err}"
self.reject_edit_last_turn(
submission_id.clone(),
ErrorEnvelope::new(
ErrorCategory::Authentication,
ErrorSeverity::Critical,
false,
"edit_last_turn_invalid_route",
format!(
"Cannot edit the last turn because its provider route is no longer valid: {err}"
),
),
))
)
.await;
continue;
}
Expand All @@ -4339,24 +4347,30 @@ impl Engine {
) {
crate::runtime_handoff::EditLastTurnTarget::Editable(idx) => idx,
crate::runtime_handoff::EditLastTurnTarget::Unsupported => {
self.reject_edit_last_turn(ErrorEnvelope::new(
ErrorCategory::InvalidInput,
ErrorSeverity::Error,
false,
"edit_last_turn_unsupported_user_content",
"Cannot edit the last turn because the latest user message has no editable text content.",
))
self.reject_edit_last_turn(
submission_id.clone(),
ErrorEnvelope::new(
ErrorCategory::InvalidInput,
ErrorSeverity::Error,
false,
"edit_last_turn_unsupported_user_content",
"Cannot edit the last turn because the latest user message has no editable text content.",
),
)
.await;
continue;
}
crate::runtime_handoff::EditLastTurnTarget::Missing => {
self.reject_edit_last_turn(ErrorEnvelope::new(
ErrorCategory::State,
ErrorSeverity::Error,
false,
"edit_last_turn_no_user_prompt",
"Cannot edit the last turn because the session history has no user message to replace.",
))
self.reject_edit_last_turn(
submission_id.clone(),
ErrorEnvelope::new(
ErrorCategory::State,
ErrorSeverity::Error,
false,
"edit_last_turn_no_user_prompt",
"Cannot edit the last turn because the session history has no user message to replace.",
),
)
.await;
continue;
}
Expand Down Expand Up @@ -5247,24 +5261,47 @@ impl Engine {
}
}

/// Reject an edit operation before model dispatch while still completing
/// the submitted host lifecycle. `Event::Error` is advisory to embedded
/// hosts; `TurnComplete(Failed)` is the authoritative terminal signal that
/// releases their busy state and closes the admitted operation.
async fn reject_edit_last_turn(&mut self, envelope: ErrorEnvelope) {
/// Reject a submitted operation before model dispatch while still
/// completing the submitted host lifecycle. `Event::Error` is advisory to
/// embedded hosts; `TurnComplete(Failed)` is the authoritative terminal
/// signal that releases their busy state and closes the admitted
/// operation. The terminal carries the rejected operation's
/// `submission_id` and no `turn_id`: nothing started, so no turn may be
/// claimed — and no earlier turn may be inherited either.
async fn reject_unstarted_submission(
&mut self,
submission_id: Option<String>,
envelope: ErrorEnvelope,
) {
let message = envelope.message.clone();
let _ = self.send_event(Event::error(envelope)).await;
let _ = self
.send_event(Event::TurnComplete {
turn_id: None,
submission_id,
compaction_id: None,
usage: Usage::default(),
parent_route_usage: Usage::default(),
routed_usage_dropped_records: 0,
status: TurnOutcomeStatus::Failed,
error: Some(message.clone()),
error: Some(message),
tool_catalog: None,
base_url: None,
})
.await;
}

/// Editing an existing conversation preserves its established goal-failure
/// reconciliation. Other callers only emit the rejected operation's
/// receipt.
async fn reject_edit_last_turn(
&mut self,
submission_id: Option<String>,
envelope: ErrorEnvelope,
) {
let message = envelope.message.clone();
self.reject_unstarted_submission(submission_id, envelope)
.await;
let outcome = SendMessageOutcome::NotStarted {
error: Some(message),
};
Expand Down Expand Up @@ -6327,16 +6364,30 @@ impl Engine {
initial_usage_owner.as_deref(),
&initial_routed_usage,
);
let _ = self
.send_event(Event::error(ErrorEnvelope::fatal_auth(format!(
"Cannot start the turn because its provider route is not ready: {err}"
))))
.await;
self.sync_unstarted_goal_for_terminal_projection(
goal_objective.as_deref(),
goal_token_budget,
goal_status,
);
// The turn never started, so the terminal names only the rejected
// host submission — no `turn_id`, and no inherited turn either.
let message =
format!("Cannot start the turn because its provider route is not ready: {err}");
let _ = self
.send_event(Event::error(ErrorEnvelope::fatal_auth(message.clone())))
.await;
terminal_permit.send(Event::TurnComplete {
turn_id: None,
submission_id,
compaction_id: None,
usage: Usage::default(),
parent_route_usage: Usage::default(),
routed_usage_dropped_records: 0,
status: TurnOutcomeStatus::Failed,
error: Some(message),
tool_catalog: None,
base_url: None,
});
let outcome = SendMessageOutcome::NotStarted { error: Some(err) };
self.reconcile_non_completed_goal_turn(&outcome).await;
return outcome;
Expand Down Expand Up @@ -6527,8 +6578,11 @@ impl Engine {
turn_id: turn.id.clone(),
created_at: turn_started_at,
route: Some(turn_route),
// Bind submit-window actions to the turn admitted by this permit.
submission_id,
// Echo the host's correlation token (`None` when this turn was
// self-started without one) so the host can bind its submit-window
// actions to the turn that actually started. The terminal
// `TurnComplete` echoes the same token again.
submission_id: submission_id.clone(),
});

// Auto's classifier completed before this parent turn was admitted.
Expand Down Expand Up @@ -6629,6 +6683,11 @@ impl Engine {
);
let error = (status == TurnOutcomeStatus::Failed).then_some(message.clone());
terminal_permit.send(Event::TurnComplete {
// `TurnStarted` already named this turn; the failed terminal
// echoes it and the submission that started it.
turn_id: Some(turn.id.clone()),
submission_id: submission_id.clone(),
compaction_id: None,
usage: turn.usage.clone(),
parent_route_usage: turn.parent_route_usage.clone(),
routed_usage_dropped_records: turn.routed_usage_dropped_records,
Expand Down Expand Up @@ -7013,6 +7072,11 @@ impl Engine {
// name was read as user-visible delivery on Interrupted turns that
// rendered nothing.)
let completion_sender = terminal_permit.send(Event::TurnComplete {
// The authoritative terminal names the turn that actually ran and
// the host submission that started it (`None` for self-starts).
turn_id: Some(turn.id.clone()),
submission_id,
compaction_id: None,
usage: turn.usage,
parent_route_usage: turn.parent_route_usage,
routed_usage_dropped_records: turn.routed_usage_dropped_records,
Expand Down Expand Up @@ -7155,6 +7219,11 @@ impl Engine {
.await;
let _ = self
.send_event(Event::TurnComplete {
// Purge carries none of the turn, submission or compaction
// identities: it never starts a model turn.
turn_id: None,
submission_id: None,
compaction_id: None,
usage: zero_usage,
parent_route_usage: Usage::default(),
routed_usage_dropped_records: 0,
Expand Down Expand Up @@ -7214,6 +7283,11 @@ impl Engine {

let _ = self
.send_event(Event::TurnComplete {
// Purge carries none of the turn, submission or compaction
// identities: it never starts a model turn.
turn_id: None,
submission_id: None,
compaction_id: None,
usage: zero_usage,
parent_route_usage: Usage::default(),
routed_usage_dropped_records: 0,
Expand Down
Loading
Loading