Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions crates/runtime/src/app/events.rs
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,7 @@ impl AppState {
}
}
AgentEvent::TurnCompleted { .. } => {
self.clear_native_subagent_work(session_id, cx);
if let Some(meta) = self.meta_mut(session_id) {
meta.updated_at = now_secs();
let meta = meta.clone();
Expand Down
3 changes: 3 additions & 0 deletions crates/runtime/src/app/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -337,6 +337,8 @@ pub struct AppState {
pending_native_rewinds: HashMap<String, (String, RewindMode)>,
/// Provider-native subagent item ids mapped to their read-only mirror sessions.
native_subagent_sessions: HashMap<(String, String), String>,
/// Synthetic turn state survives eviction; false remembers a finished child.
native_subagent_turns: HashMap<String, bool>,
pub settings: Settings,
pub providers: ProviderCatalog,
terminal_preferences_path: PathBuf,
Expand Down Expand Up @@ -475,6 +477,7 @@ impl AppState {
terminal_registry,
pending_native_rewinds: HashMap::new(),
native_subagent_sessions: HashMap::new(),
native_subagent_turns: HashMap::new(),
settings,
providers: ProviderCatalog::new(model_catalogs, provider_secret_names),
terminal_preferences_path,
Expand Down
30 changes: 19 additions & 11 deletions crates/runtime/src/app/subagents.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,14 @@ impl AppState {
);
return true;
};
// Late transcript items still need a turn for the chat view, but
// a finished child has no future lifecycle event to close it.
let finished = self.native_subagent_turns.get(&mirror_id) == Some(&false);
self.sync_mirror_turn(&mirror_id, true, parent_item_id, TurnStatus::Completed, cx);
self.record_event(&mirror_id, &strip_parent_item_id(event), cx);
if finished {
self.sync_mirror_turn(&mirror_id, false, parent_item_id, TurnStatus::Completed, cx);
}
return true;
}

Expand All @@ -48,7 +54,6 @@ impl AppState {
let in_progress = matches!(status, ItemStatus::InProgress);
let title = mirror_title(agent_type, description);
let meta = self.resident_mut(&mirror_id).map(|mirror| {
mirror.turn_in_flight = in_progress;
let title_changed = mirror.meta.title == "subagent" && mirror.meta.title != title;
if title_changed {
mirror.meta.title = title;
Expand Down Expand Up @@ -86,8 +91,12 @@ impl AppState {
cx: &mut HostCx,
) {
let open = self
.resident(mirror_id)
.is_some_and(|mirror| mirror.timeline.turn_running);
.native_subagent_turns
.insert(mirror_id.to_string(), running)
.unwrap_or(false);
if let Some(mirror) = self.resident_mut(mirror_id) {
mirror.turn_in_flight = running;
}
if running && !open {
self.record_event(
mirror_id,
Expand Down Expand Up @@ -171,7 +180,6 @@ impl AppState {
);
mirror.meta = meta;
mirror.draft = false;
mirror.turn_in_flight = true;
self.residents.parked.insert(id.clone(), mirror);
self.native_subagent_sessions.insert(key, id.clone());
Some(id)
Expand All @@ -188,20 +196,20 @@ impl AppState {
.map(|meta| (meta.id.clone(), meta.native_subagent.clone().unwrap()))
.collect();
for (mirror_id, subagent_item_id) in mirror_ids {
let was_running = self.native_subagent_turns.get(&mirror_id) == Some(&true);
self.sync_mirror_turn(
&mirror_id,
false,
&subagent_item_id,
TurnStatus::Completed,
cx,
);
let meta = self.resident_mut(&mirror_id).and_then(|mirror| {
if !mirror.turn_in_flight {
return None;
}
mirror.turn_in_flight = false;
mirror.meta.updated_at = now_secs();
Some(mirror.meta.clone())
if !was_running {
continue;
}
let meta = self.meta_mut(&mirror_id).map(|meta| {
meta.updated_at = now_secs();
meta.clone()
});
if let Some(meta) = meta {
self.persist_meta(&meta, cx);
Expand Down
190 changes: 190 additions & 0 deletions crates/runtime/src/app/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,196 @@ fn provider_native_subagent_events_create_and_feed_read_only_mirror_session() {
});
}

// Exercise provider routing and disk replay, including residency changes between events.
fn assert_native_mirror_turn_lifecycle(evict: bool, late: bool, parent_end: bool, reload: bool) {
let cx = &mut TestAppContext::default();
let test_store = TestStore::new("tcode-native-mirror-turn-lifecycle");
let state = cx.new_entity(|_| AppState::new((*test_store).clone()));
let mirror_id = state.update(cx, |state, cx| {
let mut meta = SessionMeta::new(ProviderKind::Codex, PathBuf::from("/tmp"), None);
meta.id = "parent".into();
state.sessions.push(meta.clone());
state.residents.active = Some(ActiveSession::new(meta, false, Vec::new()));
state.on_event(
"parent",
AgentEvent::TurnStarted {
turn_id: "parent-turn".into(),
},
cx,
);
state.on_event(
"parent",
native_mirror_parent_item(ItemStatus::InProgress),
cx,
);
let id = state
.sessions
.iter()
.find(|m| m.native_subagent.is_some())
.unwrap()
.id
.clone();
for i in 0..3 {
if evict && i == 1 {
assert!(state.residents.evict(&id).is_some());
assert!(state.resident(&id).is_none());
if reload {
// Force metadata lookup and its asynchronous timeline load.
state.native_subagent_sessions.clear();
}
}
state.on_event("parent", native_mirror_child_item(i), cx);
}
if parent_end {
state.on_event(
"parent",
AgentEvent::TurnCompleted {
turn_id: "parent-turn".into(),
status: TurnStatus::Completed,
usage: None,
},
cx,
);
} else {
state.on_event(
"parent",
native_mirror_parent_item(ItemStatus::Completed),
cx,
);
}
if late {
state.on_event("parent", native_mirror_child_item(3), cx);
}
id
});
cx.run_until_parked();
state.update(cx, |state, cx| {
let events = state.store.read_events(&mirror_id);
let starts = events
.iter()
.filter(|e| matches!(e.event, AgentEvent::TurnStarted { .. }))
.count();
let completions = events
.iter()
.filter(|e| matches!(e.event, AgentEvent::TurnCompleted { .. }))
.count();
assert_eq!(
starts, completions,
"persisted mirror boundaries must balance"
);
assert_eq!(starts, if late { 2 } else { 1 });
let mut open = false;
let mut items = 0;
for stored in &events {
match &stored.event {
AgentEvent::TurnStarted { .. } => {
assert!(!open);
open = true;
}
AgentEvent::TurnCompleted { .. } => {
assert!(open);
open = false;
}
AgentEvent::ItemCompleted(_) => {
assert!(open, "child must render inside a turn");
items += 1;
}
_ => {}
}
}
assert!(!open);
assert_eq!(items, if late { 4 } else { 3 });
assert!(!state.turn_running_for(&mirror_id));
if state.resident(&mirror_id).is_none() {
let meta = state.find_meta(&mirror_id).unwrap().clone();
state.load_background_session(meta, cx);
}
});
cx.run_until_parked();
state.update(cx, |state, _| {
let mirror = state.resident(&mirror_id).unwrap();
assert!(!mirror.timeline.turn_running);
assert_eq!(mirror.timeline.turns.len(), if late { 2 } else { 1 });
assert!(
mirror
.timeline
.turns
.iter()
.all(|turn| { !turn.running && turn.start_ts.is_some() && turn.end_ts.is_some() })
);
assert!(!mirror.turn_in_flight);
assert!(!mirror.has_work());
assert!(!state.session_status_snapshot(&mirror_id).unwrap().working);
assert_eq!(
mirror
.timeline
.entries
.iter()
.filter(|entry| matches!(
entry.content,
EntryContent::Item(ItemContent::AssistantMessage { .. })
))
.count(),
if late { 4 } else { 3 }
);
});
}

fn native_mirror_parent_item(status: ItemStatus) -> AgentEvent {
let item = ThreadItem {
id: "spawn-1".into(),
parent_item_id: None,
content: ItemContent::Subagent {
agent_type: "explorer".into(),
description: "Inspect routing".into(),
status,
summary: None,
},
};
if status == ItemStatus::InProgress {
AgentEvent::ItemStarted(item)
} else {
AgentEvent::ItemCompleted(item)
}
}

fn native_mirror_child_item(i: usize) -> AgentEvent {
AgentEvent::ItemCompleted(ThreadItem {
id: format!("child-{i}"),
parent_item_id: Some("spawn-1".into()),
content: ItemContent::AssistantMessage {
text: format!("answer {i}"),
},
})
}

#[test]
fn native_mirror_turn_lifecycle_resident() {
assert_native_mirror_turn_lifecycle(false, false, false, false);
}

#[test]
fn native_mirror_turn_lifecycle_evicted() {
assert_native_mirror_turn_lifecycle(true, false, false, false);
}

#[test]
fn native_mirror_turn_lifecycle_async_reload() {
assert_native_mirror_turn_lifecycle(true, false, false, true);
}

#[test]
fn native_mirror_turn_lifecycle_late_item() {
assert_native_mirror_turn_lifecycle(false, true, false, false);
assert_native_mirror_turn_lifecycle(true, true, false, false);
}

#[test]
fn native_mirror_turn_lifecycle_parent_end() {
assert_native_mirror_turn_lifecycle(false, true, true, false);
assert_native_mirror_turn_lifecycle(true, true, true, false);
}

#[test]
fn settings_patch_preserves_concurrently_changed_other_field() {
let cx = &mut TestAppContext::default();
Expand Down