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

feat(engine): echo host submission id on TurnStarted

Browse Source 
A host submitting a turn through the in-process engine
(Op::SendMessage / Op::EditLastTurn) can attach an optional
process-local correlation token, submission_id, and the engine echoes
it verbatim on that turn's TurnStarted event. An embedder that defers
submit-window actions (e.g. a stop replay) binds them to the turn
that actually started instead of the first TurnStarted to arrive:
every engine self-started turn (idle sub-agent continuation,
background shell wake, goal continuation) and the composer shell
command turn emit None, so their start can never be mistaken for the
host's pending submission even when it overtakes it in the event
stream.

The change is additive and wire-compatible both ways:
EventMsg::TurnStarted gains submission_id: Option<String> with
serde(default) + skip_serializing_if, so old consumers parse new
events and vice versa. The token stays on the internal TurnSpec only:
the wire Op projection deliberately drops it and the durable-runtime
submission path carries none, matching the in-process handle scope.
The protocol-parity guard now asserts a stamped token crosses the
Event -> wire projection verbatim; engine tests pin the echo
contract, the edit replay's own token, and a real self-start
staying None.

Signed-off-by: asto18089 <[email protected]>
This commit is contained in:
asto18089 committed 2026-09-29 16:36:25 +08:00
1 parent 5f7ca50162
commit bf7df01a99
18 files changed
+361 -3
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/protocol/src/event_msg.rs
+48
View File
@@ -392,6 +392,14 @@ pub enum EventMsg {
created_at: DateTime<Utc>,
#[serde(default, skip_serializing_if = "Option::is_none")]
route: Option<TurnRoute>,
/// Host submission correlation echo: the token an in-process host
/// stamped on its `SendMessage`/`EditLastTurn` op, echoed verbatim by
/// that turn's start; `None` for every engine self-started turn.
/// Additive and default-absent on the wire. Only in-process engine
/// handles can stamp a token — the wire `Op` carries no correlation
/// field — so wire submitters only ever observe `None` here.
#[serde(default, skip_serializing_if = "Option::is_none")]
submission_id: Option<String>,
},
/// Bounded tool-field projection from a prepared model-client request
/// (`ToolInspectionSnapshot` serialized).
@@ -1113,6 +1121,7 @@ mod tests {
turn_id: "turn-1".into(),
created_at: DateTime::<Utc>::from_timestamp(1, 0).unwrap(),
route: Some(route.clone()),
submission_id: None,
},
EventMsg::ToolRequestSnapshot {
thread_id: t.clone(),
@@ -1546,4 +1555,43 @@ mod tests {
let back: EventMsg = serde_json::from_str(&json).unwrap();
assert_eq!(back, msg);
}
/// A host-stamped `submission_id` crosses the wire verbatim, a `None`
/// token stays absent, and a payload from a producer that predates the
/// field still deserializes (`serde(default)`).
#[test]
fn turn_started_submission_id_is_additive_and_default_absent() {
let msg = EventMsg::TurnStarted {
thread_id: ThreadId::new(),
session_id: SessionId::new(),
turn_id: "turn-1".into(),
created_at: DateTime::<Utc>::from_timestamp(1, 0).unwrap(),
route: None,
submission_id: Some("sub-host-1".into()),
};
let json = serde_json::to_string(&msg).unwrap();
assert!(
json.contains(r#""submission_id":"sub-host-1""#),
"a host-stamped token must be serialized: {json}"
);
let back: EventMsg = serde_json::from_str(&json).unwrap();
assert_eq!(back, msg);
let none = EventMsg::TurnStarted {
thread_id: ThreadId::new(),
session_id: SessionId::new(),
turn_id: "turn-2".into(),
created_at: DateTime::<Utc>::from_timestamp(2, 0).unwrap(),
route: None,
submission_id: None,
};
let value = serde_json::to_value(&none).unwrap();
assert!(
value.get("submission_id").is_none(),
"a self-started turn's None token must stay absent on the wire: {value}"
);
// An older producer that predates the field: the absent key defaults.
let back: EventMsg = serde_json::from_value(value).unwrap();
assert_eq!(back, none);
}
}
crates/tui/src/context_report/pressure_fixture_tests.rs
+1
View File
@@ -92,6 +92,7 @@ fn turn_op(content: &str, route: &ResolvedRuntimeRoute) -> Op {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
})
}
crates/tui/src/core/engine.rs
+22 -1
View File
@@ -2050,6 +2050,9 @@ impl Engine {
turn_id: turn_id.clone(),
created_at: chrono::Utc::now(),
route: None,
// A composer shell command has no host submission envelope to
// correlate with.
submission_id: None,
})
.await;
@@ -2917,6 +2920,9 @@ impl Engine {
provenance: UserInputProvenance::Runtime,
images: Vec::new(),
max_output_tokens: None,
// Background shell completion wake: no host submission to
// correlate with.
submission_id: None,
})
.await;
}
@@ -3096,6 +3102,9 @@ impl Engine {
provenance: UserInputProvenance::Runtime,
images: Vec::new(),
max_output_tokens: None,
// Engine-scheduled goal continuation: no host
// submission to correlate with.
submission_id: None,
})
.await;
}
@@ -3635,7 +3644,10 @@ impl Engine {
}
self.handle_purge().await;
}
Op::EditLastTurn { new_message } => {
Op::EditLastTurn {
new_message,
submission_id,
} => {
let route = match self.current_runtime_route() {
Ok(route) => route,
Err(err) => {
@@ -3717,6 +3729,7 @@ impl Engine {
provenance: UserInputProvenance::ExternalUser,
images: Vec::new(),
max_output_tokens: None,
submission_id,
})
.await;
}
@@ -4400,6 +4413,9 @@ impl Engine {
provenance: UserInputProvenance::SubAgentHandoff,
images: Vec::new(),
max_output_tokens: None,
// Idle sub-agent completion resume: no host submission to
// correlate with.
submission_id: None,
})
.await;
if !outcome.started() {
@@ -5248,6 +5264,7 @@ impl Engine {
hook_executor,
verbosity,
provenance,
submission_id,
} = spec;
let route = *route;
let compaction = *compaction;
@@ -5501,6 +5518,10 @@ impl Engine {
turn_id: turn.id.clone(),
created_at: turn_started_at,
route: Some(turn_route),
// 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.
submission_id,
})
.await;
crates/tui/src/core/engine/preview/tests.rs
+2
View File
@@ -981,6 +981,7 @@ async fn assert_preview_matches_first_wire_body(
provenance: UserInputProvenance::ExternalUser,
images: Vec::new(),
max_output_tokens: None,
submission_id: None,
})
.await;
@@ -1948,6 +1949,7 @@ async fn provider_reported_usage_is_unavailable_until_a_response_reports_it() {
provenance: UserInputProvenance::ExternalUser,
images: Vec::new(),
max_output_tokens: None,
submission_id: None,
})
.await;
crates/tui/src/core/engine/tests.rs
+175
View File
@@ -1079,6 +1079,7 @@ async fn exact_turn_snapshot_restores_custom_endpoint_and_turn_receipt_after_bui
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send exact custom turn");
@@ -1255,6 +1256,7 @@ async fn main_turn_dispatch_freezes_declared_custom_model_rate() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send declared custom turn");
@@ -1570,6 +1572,7 @@ async fn goal_continuation_preserves_goal_and_resolves_updated_authoritative_rou
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send first goal turn");
@@ -1852,6 +1855,7 @@ async fn saturated_mailbox_does_not_deadlock_goal_continuation_self_dispatch() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send saturated goal turn");
@@ -1984,6 +1988,7 @@ async fn queued_ordinary_turn_does_not_multiply_engine_goal_continuations() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
})
};
@@ -3271,6 +3276,7 @@ async fn cross_turn_token_budget_exhaustion_does_not_pause_goal() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send budgeted goal turn");
@@ -3731,6 +3737,7 @@ async fn ordinary_prose_never_activates_a_goal() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send explicit natural goal turn");
@@ -3831,6 +3838,7 @@ async fn operate_goal_probe(mode: AppMode, prompt: &str) -> (Option<String>, boo
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send probe turn");
@@ -4016,6 +4024,7 @@ async fn operate_contract_is_appended_once_and_an_existing_goal_is_never_replace
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
})
};
@@ -4800,6 +4809,7 @@ async fn host_managed_engine_does_not_self_dispatch_goal_continuation() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send host-owned goal turn");
@@ -5076,6 +5086,7 @@ async fn host_managed_engine_defers_idle_subagent_completion_to_explicit_turn()
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send explicit host turn");
@@ -5108,6 +5119,127 @@ async fn host_managed_engine_defers_idle_subagent_completion_to_explicit_turn()
run_task.await.expect("engine task");
}
/// A host-submitted turn echoes the correlation token it was submitted with,
/// while a runtime self-started turn (idle sub-agent completion resume) never
/// carries one. With this contract a host can tell "my pending submission
/// started" apart from "an autonomous follow-up overtook it in the event
/// stream" and only ever consume a deferred submit-window action on the
/// former.
#[tokio::test]
async fn turn_started_echoes_submission_id_and_self_starts_stay_none() {
use crate::llm_client::mock::{MockLlmClient, canned};
use crate::tools::subagent::SubAgentCompletion;
let workspace = tempdir().unwrap();
let config = Config::default();
let mock = Arc::new(MockLlmClient::new(vec![
canned::simple_text_turn("submitted turn finished."),
canned::simple_text_turn("self-started continuation finished."),
]));
let (engine, handle) = Engine::new_with_model_client(
deterministic_engine_config(workspace.path()),
&config,
mock.clone(),
);
let owner_session_id = engine.session.id.clone();
let completion_tx = engine.tx_subagent_completion.clone();
let mut op = external_user_message_op("Watch the child agent.", AppMode::Agent, &config);
if let Op::SendMessage(TurnSpec { submission_id, .. }) = &mut op {
*submission_id = Some("sub-host-1".to_string());
}
let run_task = tokio::spawn(engine.run());
handle.send(op).await.expect("send correlated turn");
// The submitted turn's start echoes the token verbatim.
let submitted_turn_id = {
let mut rx = handle.rx_event.write().await;
loop {
let event = tokio::time::timeout(model_turn_event_timeout(), rx.recv())
.await
.expect("timed out waiting for the submitted turn")
.expect("engine event");
if let Event::TurnStarted {
turn_id,
submission_id,
..
} = event
{
assert_eq!(
submission_id.as_deref(),
Some("sub-host-1"),
"the submitted turn's TurnStarted must echo its correlation token"
);
break turn_id;
}
}
};
{
let mut rx = handle.rx_event.write().await;
loop {
let event = tokio::time::timeout(model_turn_event_timeout(), rx.recv())
.await
.expect("timed out waiting for the submitted turn to complete")
.expect("engine event");
if let Event::TurnComplete { status, error, .. } = event {
assert_eq!(status, TurnOutcomeStatus::Completed, "{error:?}");
break;
}
}
}
// An idle child completion self-starts the follow-up without any host
// submission; its start must not present a token.
completion_tx
.try_send(SubAgentCompletion {
owner_session_id,
agent_id: "idle-child".to_string(),
payload: "child finished its work".to_string(),
})
.expect("inject idle sub-agent completion");
{
let mut rx = handle.rx_event.write().await;
loop {
let event = tokio::time::timeout(model_turn_event_timeout(), rx.recv())
.await
.expect("timed out waiting for the self-started turn")
.expect("engine event");
if let Event::TurnStarted {
turn_id,
submission_id,
..
} = event
{
assert_ne!(
turn_id, submitted_turn_id,
"the idle child completion must self-start a new turn"
);
assert!(
submission_id.is_none(),
"a runtime self-started turn must not present a submission id"
);
break;
}
}
}
{
let mut rx = handle.rx_event.write().await;
loop {
let event = tokio::time::timeout(model_turn_event_timeout(), rx.recv())
.await
.expect("timed out waiting for the self-started turn to complete")
.expect("engine event");
if let Event::TurnComplete { status, error, .. } = event {
assert_eq!(status, TurnOutcomeStatus::Completed, "{error:?}");
break;
}
}
}
handle.send(Op::Shutdown).await.expect("shutdown engine");
run_task.await.expect("engine task");
assert_eq!(mock.call_count(), 2);
}
#[test]
fn idle_and_in_turn_subagent_delivery_claim_each_completion_once() {
use crate::tools::subagent::SubAgentCompletion;
@@ -6616,6 +6748,7 @@ fn active_goal_message_op(
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
})
}
@@ -6655,6 +6788,7 @@ fn external_user_message_op(content: &str, mode: AppMode, config: &Config) -> Op
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
})
}
@@ -6683,6 +6817,7 @@ fn auto_review_message_op(content: &str, config: &Config) -> Op {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
})
}
@@ -13082,6 +13217,7 @@ async fn deferred_tool_first_use_does_not_emit_a_retry_status() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send model turn");
@@ -13853,6 +13989,7 @@ async fn operate_model_shell_uses_normal_approval_and_workspace_sandbox() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send Operate model turn");
@@ -14008,6 +14145,7 @@ async fn posture_change_during_approval_wait(
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send model turn");
@@ -14224,6 +14362,7 @@ async fn full_access_subagent_handoff_keeps_model_shell_free_of_approval_prompts
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::SubAgentHandoff,
submission_id: None,
}))
.await
.expect("send model turn");
@@ -14360,6 +14499,7 @@ async fn assert_full_access_model_tool_batch_is_blocked(
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send Full Access model turn");
@@ -14567,6 +14707,7 @@ async fn assert_full_access_model_tool_batch_runs(
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send Full Access model turn");
@@ -14843,6 +14984,7 @@ async fn auto_review_asks_the_user_and_returns_the_answer() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send Auto-Review model turn");
@@ -15029,6 +15171,7 @@ async fn full_access_permission_allow_cannot_bypass_background_catastrophic_floo
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send model turn");
@@ -15171,6 +15314,7 @@ async fn yolo_mode_does_not_prompt_for_background_shell() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send model turn");
@@ -15309,6 +15453,7 @@ async fn yolo_mode_executes_publish_like_shell_without_prompt() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send model turn");
@@ -15451,6 +15596,7 @@ async fn yolo_mode_does_not_prompt_for_mcp_action() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send model turn");
@@ -17994,9 +18140,29 @@ async fn edit_last_turn_preserves_current_mode() {
handle
.send(Op::EditLastTurn {
new_message: "revise this in plan mode".to_string(),
submission_id: Some("sub-edit-1".to_string()),
})
.await
.expect("send edit");
// The replacement turn the edit replays must echo the edit's own
// correlation token, not the one of any earlier turn.
{
let mut rx = handle.rx_event.write().await;
loop {
let event = tokio::time::timeout(model_turn_event_timeout(), rx.recv())
.await
.expect("timed out waiting for the edited turn")
.expect("engine event");
if let Event::TurnStarted { submission_id, .. } = event {
assert_eq!(
submission_id.as_deref(),
Some("sub-edit-1"),
"the edit's replacement turn must echo its correlation token"
);
break;
}
}
}
let (tx, rx) = tokio::sync::oneshot::channel();
handle
@@ -18119,6 +18285,7 @@ async fn edit_last_turn_cuts_at_user_prompt_before_tool_results() {
handle
.send(Op::EditLastTurn {
new_message: "edited prompt".to_string(),
submission_id: None,
})
.await
.expect("send edit");
@@ -18230,6 +18397,7 @@ async fn edit_last_turn_without_user_prompt_errors_and_sends_nothing() {
handle
.send(Op::EditLastTurn {
new_message: "edited prompt".to_string(),
submission_id: None,
})
.await
.expect("send edit");
@@ -18332,6 +18500,7 @@ async fn edit_last_turn_without_user_prompt_errors_and_sends_nothing() {
handle
.send(Op::EditLastTurn {
new_message: "must not replace the older prompt".to_string(),
submission_id: None,
})
.await
.expect("send unsupported edit");
@@ -22901,6 +23070,7 @@ async fn run_headless_turn_with_flaky_network(
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send flaky-network turn");
@@ -23082,6 +23252,7 @@ async fn terminal_output_limit_followed_by_stream_error_is_charged_and_not_retri
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send terminal-then-drop turn");
@@ -23189,6 +23360,7 @@ async fn midstream_error_frame_stops_the_stream_and_drops_trailing_deltas() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send midstream-error turn");
@@ -23440,6 +23612,7 @@ async fn run_interactive_turn_with_flaky_network(
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send interactive flaky-network turn");
@@ -23671,6 +23844,7 @@ async fn interactive_thinking_only_drop_preserves_nothing_and_never_claims_it_di
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send thinking-only drop turn");
@@ -23901,6 +24075,7 @@ async fn run_reasoning_only_turn_with_reprompts(
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send reasoning-only turn");
crates/tui/src/core/events.rs
+10
View File
@@ -273,6 +273,16 @@ pub enum Event {
/// Legacy/non-model hosts may still attach a route at start. Model
/// turns emit it separately at the application dispatch boundary.
route: Option<TurnRoute>,
/// Correlation with the host submission that produced this turn: the
/// verbatim echo of the `submission_id` the host stamped on its
/// `SendMessage`/`EditLastTurn` op, `None` for every runtime
/// self-started turn (idle sub-agent completion, background shell
/// wake, goal continuation) and for the composer shell command turn.
/// Hosts use the echo to tell their own pending submission's
/// `TurnStarted` apart from an autonomous follow-up whose start event
/// overtook it in the stream, so a deferred submit-window action is
/// only ever consumed by the turn it targets.
submission_id: Option<String>,
},
/// Bounded tool-field projection from a prepared model-client request.
crates/tui/src/core/ops.rs
+18 -1
View File
@@ -216,6 +216,18 @@ pub struct TurnSpec {
/// Structural input origin. This gates whether the turn may inherit
/// YOLO/auto-approval authority; user-shaped text is not enough.
pub provenance: UserInputProvenance,
/// Host-supplied correlation token for this submission, echoed verbatim on
/// the turn's `Event::TurnStarted`. Hosts that arm submit→`TurnStarted`
/// window actions (e.g. a deferred stop replay) use the echo to bind those
/// actions to the turn that actually started: every engine self-started
/// turn (idle sub-agent completion, background shell wake, goal
/// continuation) and the composer shell command turn carry no token, so
/// their start cannot be mistaken for a pending submission even when it
/// overtakes it in the event stream. `None` for callers that do not
/// correlate. The token only survives the in-process engine path: the
/// wire op projection and the durable-runtime submission path carry none,
/// so hosts submitting over those channels cannot correlate.
pub submission_id: Option<String>,
}
/// Operations that can be submitted to the engine.
@@ -432,7 +444,12 @@ pub enum Op {
/// Edit the last user message: remove the last user+assistant exchange
/// from the session, then re-send with the new content.
#[cfg_attr(not(test), expect(dead_code))]
EditLastTurn { new_message: String },
EditLastTurn {
new_message: String,
/// Host-supplied correlation token, echoed on the replayed turn's
/// `Event::TurnStarted` (see `TurnSpec::submission_id`).
submission_id: Option<String>,
},
/// Enable or disable the background advisor watcher for this session.
/// When enabled, a fire-and-forget background task runs after each turn
crates/tui/src/core/protocol_parity.rs
+21 -1
View File
@@ -385,6 +385,10 @@ fn compaction_to_wire(config: &CompactionConfig) -> wire_op::CompactionPolicy {
/// fields (`initial_routed_usage`, `hook_executor`) and resolved routes are
/// stripped; only their non-secret receipts cross.
fn turn_spec_to_wire(spec: &crate::core::ops::TurnSpec) -> wire_op::TurnSpec {
// `spec.submission_id` is deliberately not projected: the token is
// host-process-local by design and the wire op has no correlation twin,
// so wire submitters observe `TurnStarted.submission_id` always absent
// and cannot correlate submissions on that channel.
wire_op::TurnSpec {
max_output_tokens: spec.max_output_tokens,
content: spec.content.clone(),
@@ -537,12 +541,14 @@ pub fn event_to_protocol(event: &Event, ids: &ProtocolIds) -> wire::EventMsg {
turn_id,
created_at,
route,
submission_id,
} => wire::EventMsg::TurnStarted {
thread_id,
session_id,
turn_id: turn_id.clone(),
created_at: *created_at,
route: route.as_ref().map(route_to_wire),
submission_id: submission_id.clone(),
},
Event::ToolRequestSnapshot { snapshot } => wire::EventMsg::ToolRequestSnapshot {
thread_id,
@@ -1179,7 +1185,10 @@ pub fn op_to_protocol(op: &Op) -> wire_op::Op {
config_path: config_path.clone(),
},
Op::PurgeContext => wire_op::Op::PurgeContext,
Op::EditLastTurn { new_message } => wire_op::Op::EditLastTurn {
Op::EditLastTurn {
new_message,
submission_id: _,
} => wire_op::Op::EditLastTurn {
new_message: new_message.clone(),
},
Op::SetAdvisorEnabled { enabled } => wire_op::Op::SetAdvisorEnabled { enabled: *enabled },
@@ -1436,6 +1445,9 @@ mod tests {
turn_id: "turn-1".into(),
created_at: chrono::Utc::now(),
route: None,
// A host-stamped token, not `None`: the projection must carry
// it verbatim or the assertion on `events[7]` below fails.
submission_id: Some("sub-host-1".into()),
},
Event::TurnComplete {
usage: usage.clone(),
@@ -1534,6 +1546,13 @@ mod tests {
serde_json::to_value(events[6].to_protocol(&ids)).unwrap()["result"],
json!({"outcome": "err", "error": {"kind": "timeout", "seconds": 9}})
);
// The host correlation token must cross the projection verbatim: the
// app forwarder binds its submit-window actions to this echo, so a
// silently dropped mapping would defeat the contract.
assert_eq!(
serde_json::to_value(events[7].to_protocol(&ids)).unwrap()["submission_id"],
json!("sub-host-1")
);
assert_eq!(
serde_json::to_value(events[8].to_protocol(&ids)).unwrap()["status"],
"interrupted"
@@ -1603,6 +1622,7 @@ mod tests {
Op::PurgeContext,
Op::EditLastTurn {
new_message: "again".into(),
submission_id: None,
},
Op::SetAdvisorEnabled { enabled: true },
Op::GetSubAgentSettlement {
crates/tui/src/core/queued_approval_tests.rs
+1
View File
@@ -144,6 +144,7 @@ async fn approving_the_first_of_three_queued_calls_cancels_none_of_them() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("send model turn");
crates/tui/src/exec_agent.rs
+2
View File
@@ -762,6 +762,8 @@ pub(crate) async fn run_exec_agent(
},
verbosity: execution_config.verbosity.clone(),
provenance: crate::core::ops::UserInputProvenance::ExternalUser,
// Headless exec does not correlate submissions.
submission_id: None,
}))
.await?;
crates/tui/src/remote_control.rs
+1
View File
@@ -6510,6 +6510,7 @@ mod tests {
turn_id: "turn_started_later".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
});
let WorkerCommand::Upload { envelopes, .. } =
worker_rx.try_recv().expect("one typed turn start")
crates/tui/src/runtime_api/tests.rs
+9
View File
@@ -3194,6 +3194,7 @@ async fn compatibility_stream_closes_losslessly_across_replay_live_handoff() ->
turn_id: "mock_compat_handoff".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -3400,6 +3401,7 @@ async fn compatibility_stream_exposes_and_resolves_user_input_without_answer_ech
turn_id: "mock_compat_input".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await?;
let request = crate::tools::user_input::UserInputRequest {
@@ -3817,6 +3819,7 @@ async fn thread_endpoints_expose_lifecycle_contract() -> Result<()> {
turn_id: "mock_lifecycle".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -4208,6 +4211,7 @@ async fn turn_operation_lookup_is_authenticated_read_only_and_survives_restart()
turn_id: "lookup-fixture-engine-turn".into(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
},
EngineEvent::MessageStarted { index: 0 },
EngineEvent::MessageDelta {
@@ -4342,6 +4346,7 @@ async fn events_endpoint_respects_since_seq_cursor() -> Result<()> {
turn_id: "mock_cursor".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -5273,6 +5278,7 @@ async fn steer_and_interrupt_endpoints_work_on_active_turn() -> Result<()> {
turn_id: "engine_turn_api".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -5739,6 +5745,7 @@ async fn stream_endpoint_remains_backward_compatible() -> Result<()> {
turn_id: "mock_stream".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -8299,6 +8306,7 @@ async fn session_create_from_thread_rejects_active_turn() -> Result<()> {
turn_id: "mock_active_session_save".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -18213,6 +18221,7 @@ async fn runtime_image_http_rejects_before_dispatch_and_accepts_large_canonical_
turn_id: "image-http-fixture".into(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
harness
crates/tui/src/runtime_threads.rs
+3
View File
@@ -12253,6 +12253,8 @@ impl RuntimeThreadManager {
approval_mode: policy.permission,
verbosity,
provenance: input_source.provenance(),
// Durable-runtime submissions carry no host correlation token.
submission_id: None,
});
// Reserve mailbox capacity before claiming or persisting anything.
@@ -14042,6 +14044,7 @@ impl RuntimeThreadManager {
turn_id: started_turn_id,
created_at,
route,
submission_id: _,
} => {
saw_turn_started = true;
engine_turn_id = Some(started_turn_id);
crates/tui/src/runtime_threads/tests.rs
+34
View File
@@ -8350,6 +8350,7 @@ async fn thread_lifecycle_persists_across_restart() -> Result<()> {
turn_id: "engine_turn_1".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -8515,6 +8516,7 @@ async fn initial_classifier_usage_is_persisted_before_terminal_and_merged_exactl
turn_id: "engine_classifier_receipt".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
harness
@@ -9092,6 +9094,7 @@ async fn monitor_deduplicates_sink_and_metadata_and_persists_metadata_only_missi
turn_id: "engine_metadata_receipt".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
for _ in 0..2 {
@@ -9174,6 +9177,7 @@ async fn monitor_separates_lifecycle_start_from_billing_dispatch_and_child_usage
turn_id: "engine_route_receipt".to_string(),
created_at: started_at,
route: None,
submission_id: None,
})
.await?;
harness
@@ -9342,6 +9346,7 @@ async fn monitor_separates_lifecycle_start_from_billing_dispatch_and_child_usage
turn_id: second_engine_turn.to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
harness
@@ -9421,6 +9426,7 @@ async fn monitor_persists_request_snapshots_and_matching_terminal_diagnostics()
turn_id: engine_turn_id.to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
let mut tool = catalog_tool("mcp_computer_get_app_state");
@@ -9588,6 +9594,7 @@ async fn monitor_persists_request_snapshots_and_matching_terminal_diagnostics()
turn_id: second_engine_turn_id.to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
let pre_request = crate::tool_inspection::ToolInspectionSnapshot::from_prepared_request(
@@ -9679,6 +9686,7 @@ async fn completed_turn_without_engine_output_fails() -> Result<()> {
turn_id: "engine_empty_turn".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -9770,6 +9778,7 @@ async fn worker_lifecycle_receipts_preserve_owner_outcome_and_durable_replay() -
turn_id: "engine_worker_lifecycle".into(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await;
for owner in [thread_id.clone(), foreign_id] {
@@ -9968,6 +9977,7 @@ async fn preturn_control_status_does_not_make_empty_turn_succeed() -> Result<()>
turn_id: "engine_empty_after_control_status".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -10025,6 +10035,7 @@ async fn engine_error_remains_failed_after_nominal_turn_complete() -> Result<()>
turn_id: "engine_error_then_complete".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -10836,6 +10847,7 @@ async fn multi_turn_continuity_same_thread() -> Result<()> {
turn_id: format!("engine_turn_{turn_index}"),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -11077,6 +11089,7 @@ async fn host_goal_loop_kickoff_arms_one_continuation_and_parks_at_engine_cap()
turn_id: format!("engine_goal_{pass}"),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -11195,6 +11208,7 @@ async fn host_goal_loop_skips_rearm_without_update_goal_and_after_failed_pass()
turn_id: "engine_goal_no_update".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -11285,6 +11299,7 @@ async fn host_goal_loop_skips_rearm_without_update_goal_and_after_failed_pass()
turn_id: "engine_goal_failed".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = failed_tx_event
@@ -11382,6 +11397,7 @@ async fn host_goal_loop_mirrors_terminal_snapshot_and_does_not_rearm() -> Result
turn_id: format!("engine_{status}"),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -11487,6 +11503,7 @@ async fn model_created_goal_persists_through_adopted_revision() -> Result<()> {
turn_id: "engine_model_created".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -11570,6 +11587,7 @@ async fn model_created_goal_never_overwrites_concurrent_explicit_goal() -> Resul
turn_id: "engine_concurrent".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -11896,6 +11914,7 @@ async fn interrupt_turn_marks_interrupted_after_cleanup() -> Result<()> {
turn_id: "engine_turn_interrupt".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -12690,6 +12709,7 @@ async fn thread_detail_cursor_precedes_projection_reads_at_terminal_boundary() -
turn_id: "snapshot_terminal".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
harness
@@ -12871,6 +12891,7 @@ async fn thread_detail_materializes_stream_prefixes_before_their_delta_cursor()
turn_id: "delta_snapshot".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
harness
@@ -12994,6 +13015,7 @@ async fn thread_detail_delta_boundary_is_replay_idempotent() -> Result<()> {
turn_id: "delta_boundary".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
harness
@@ -13337,6 +13359,7 @@ async fn dynamic_tool_result_settles_snapshot_and_emits_one_safe_resolution() ->
turn_id: "dynamic_result".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
@@ -13498,6 +13521,7 @@ async fn dynamic_tool_result_receipt_outlives_canceled_delivery_future() -> Resu
turn_id: "dynamic_detached_settlement".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
@@ -14440,6 +14464,7 @@ async fn dynamic_tool_timeout_clears_snapshot_and_emits_once() -> Result<()> {
turn_id: "dynamic_timeout".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
@@ -14514,6 +14539,7 @@ async fn terminal_turn_cancels_pending_dynamic_tool_exactly_once() -> Result<()>
turn_id: "dynamic_cancel".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
@@ -14623,6 +14649,7 @@ async fn approval_wait_heartbeat_is_never_sequenced_after_the_decision() -> Resu
turn_id: "engine_turn_wait".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
// The Responses client joins call and item ids with `|`.
@@ -15828,6 +15855,7 @@ async fn steer_turn_on_active_turn_records_item_and_event() -> Result<()> {
turn_id: "engine_turn_steer".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
if let Some(steer) = rx_steer.recv().await {
@@ -16477,6 +16505,7 @@ async fn compaction_lifecycle_emits_item_events_with_compaction_counts() -> Resu
turn_id: "engine_turn_auto".to_string(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
})
.await;
let _ = tx_event
@@ -18933,6 +18962,7 @@ async fn notices_raise_from_engine_events_and_clear_on_settle_or_ack() -> Result
turn_id: turn.id.clone(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
@@ -19514,6 +19544,7 @@ async fn engine_plumbing_items_are_tagged_internal_and_retry_hints_are_dropped()
turn_id: "engine_plumbing_visibility".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
for status in [
@@ -20166,6 +20197,7 @@ async fn runtime_receipts_mask_configured_secrets() -> Result<()> {
turn_id: turn.id.clone(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
harness
@@ -20401,6 +20433,7 @@ async fn tool_completion_items_carry_typed_artifact_refs() -> Result<()> {
turn_id: turn.id.clone(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
for (call, name) in [("call_write", "apply_patch"), ("call_shell", "exec_shell")] {
@@ -20633,6 +20666,7 @@ async fn run_workspace_turn(
turn_id: "engine-turn".to_string(),
created_at: Utc::now(),
route: None,
submission_id: None,
})
.await?;
let mut sent = Vec::new();
crates/tui/src/tui/pet_watch/mod.rs
+6
View File
@@ -741,6 +741,7 @@ mod tests {
turn_id: "preview-turn".into(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
},
Instant::now(),
);
@@ -817,6 +818,7 @@ mod tests {
turn_id: "turn".into(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
},
Instant::now(),
);
@@ -885,6 +887,7 @@ mod tests {
turn_id: "turn".into(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
},
Instant::now(),
);
@@ -923,6 +926,7 @@ mod tests {
turn_id: format!("turn-{:?}", status),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
},
Instant::now(),
);
@@ -936,6 +940,7 @@ mod tests {
turn_id: "turn-completed".into(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
},
Instant::now(),
);
@@ -961,6 +966,7 @@ mod tests {
turn_id: "turn-b".into(),
created_at: chrono::Utc::now(),
route: None,
submission_id: None,
},
Some("session-b"),
Instant::now(),
crates/tui/src/tui/ui/dispatch.rs
+2
View File
@@ -909,6 +909,8 @@ pub(crate) async fn spawned_dispatch_inner(
hook_executor: prepare.hook_executor.clone(),
verbosity: prepare.verbosity.clone(),
provenance: prepare.provenance,
// Interactive TUI submissions do not correlate submissions.
submission_id: None,
});
// Reserve capacity off the render thread, but do not let Engine start
// until the completion callback has installed the UI's acceptance state.
crates/tui/src/tui/ui/session_state.rs
+1
View File
@@ -585,6 +585,7 @@ pub(crate) fn capture_turn_started_metadata(app: &mut App, event: &EngineEvent)
turn_id,
created_at,
route,
submission_id: _,
} => {
app.ocean_completion_started_at = None;
let auto_route_receipt = if route.as_ref().is_some_and(|route| route.auto_model) {
crates/tui/src/tui/ui/tests.rs
+5
View File
@@ -17676,6 +17676,7 @@ fn turn_started_route_is_captured_before_cancel_suppression() {
let event = EngineEvent::TurnStarted {
turn_id: "turn_cancel_race".to_string(),
created_at,
submission_id: None,
route: Some(crate::core::events::TurnRoute {
provider: ApiProvider::Openai,
provider_identity: "openai".to_string(),
@@ -17759,6 +17760,7 @@ fn turn_started_suggestion_authority_comes_from_the_route_receipt_not_config() {
base_url: String::new(),
billing_product: crate::route_billing::RouteProduct::Unproven,
}),
submission_id: None,
};
capture_turn_started_metadata(&mut app, &event);
@@ -17803,6 +17805,7 @@ fn turn_started_without_a_route_receipt_captures_no_suggestion_authority() {
base_url: String::new(),
billing_product: crate::route_billing::RouteProduct::Unproven,
}),
submission_id: None,
};
capture_turn_started_metadata(&mut app, &event);
@@ -17841,6 +17844,7 @@ fn engine_error_health_accounting_uses_active_turn_route() {
base_url: String::new(),
billing_product: crate::route_billing::RouteProduct::Unproven,
}),
submission_id: None,
};
capture_turn_started_metadata(&mut app, &event);
@@ -26833,6 +26837,7 @@ async fn keyless_engine_error_stays_visible_after_a_config_ack() {
hook_executor: None,
verbosity: None,
provenance: UserInputProvenance::ExternalUser,
submission_id: None,
}))
.await
.expect("submit to the real Engine");