fix: add native turn lifecycle and stabilize opencode session flow

server/packages/universal-agent-schema/src/agents/amp.rs

@@ -4,7 +4,7 @@ use serde_json::Value;
 
 use crate::amp as schema;
 use crate::{
-    turn_completed_event, ContentPart, ErrorData, EventConversion, ItemDeltaData, ItemEventData,
+    turn_ended_event, ContentPart, ErrorData, EventConversion, ItemDeltaData, ItemEventData,
     ItemKind, ItemRole, ItemStatus, SessionEndReason, SessionEndedData, TerminatedBy,
     UniversalEventData, UniversalEventType, UniversalItem,
 };
@@ -99,7 +99,7 @@ pub fn event_to_universal(
             ));
         }
         schema::StreamJsonMessageType::Done => {
-            events.push(turn_completed_event());
+            events.push(turn_ended_event(None, None).synthetic());
             events.push(
                 EventConversion::new(
                     UniversalEventType::SessionEnded,

server/packages/universal-agent-schema/src/agents/claude.rs

@@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use serde_json::Value;
 
 use crate::{
-    turn_completed_event, ContentPart, EventConversion, ItemDeltaData, ItemEventData, ItemKind,
+    turn_ended_event, ContentPart, EventConversion, ItemDeltaData, ItemEventData, ItemKind,
     ItemRole, ItemStatus, PermissionEventData, PermissionStatus, QuestionEventData, QuestionStatus,
     SessionStartedData, UniversalEventData, UniversalEventType, UniversalItem,
 };
@@ -425,7 +425,7 @@ fn result_event_to_universal(event: &Value, session_id: &str) -> Vec<EventConver
             UniversalEventType::ItemCompleted,
             UniversalEventData::Item(ItemEventData { item: message_item }),
         ),
-        turn_completed_event(),
+        turn_ended_event(None, None).synthetic(),
     ]
 }
 

server/packages/universal-agent-schema/src/agents/codex.rs

@@ -4,7 +4,7 @@ use crate::codex as schema;
 use crate::{
     ContentPart, ErrorData, EventConversion, ItemDeltaData, ItemEventData, ItemKind, ItemRole,
     ItemStatus, ReasoningVisibility, SessionEndReason, SessionEndedData, SessionStartedData,
-    TerminatedBy, UniversalEventData, UniversalEventType, UniversalItem,
+    TerminatedBy, TurnEventData, TurnPhase, UniversalEventData, UniversalEventType, UniversalItem,
 };
 
 /// Convert a Codex ServerNotification to universal events.
@@ -36,18 +36,26 @@ pub fn notification_to_universal(
             Some(params.thread_id.clone()),
             raw,
         )]),
-        schema::ServerNotification::TurnStarted(params) => Ok(vec![status_event(
-            "turn.started",
-            serde_json::to_string(&params.turn).ok(),
-            Some(params.thread_id.clone()),
-            raw,
-        )]),
-        schema::ServerNotification::TurnCompleted(params) => Ok(vec![status_event(
-            "turn.completed",
-            serde_json::to_string(&params.turn).ok(),
-            Some(params.thread_id.clone()),
-            raw,
-        )]),
+        schema::ServerNotification::TurnStarted(params) => Ok(vec![EventConversion::new(
+            UniversalEventType::TurnStarted,
+            UniversalEventData::Turn(TurnEventData {
+                phase: TurnPhase::Started,
+                turn_id: Some(params.turn.id.clone()),
+                metadata: serde_json::to_value(&params.turn).ok(),
+            }),
+        )
+        .with_native_session(Some(params.thread_id.clone()))
+        .with_raw(raw)]),
+        schema::ServerNotification::TurnCompleted(params) => Ok(vec![EventConversion::new(
+            UniversalEventType::TurnEnded,
+            UniversalEventData::Turn(TurnEventData {
+                phase: TurnPhase::Ended,
+                turn_id: Some(params.turn.id.clone()),
+                metadata: serde_json::to_value(&params.turn).ok(),
+            }),
+        )
+        .with_native_session(Some(params.thread_id.clone()))
+        .with_raw(raw)]),
         schema::ServerNotification::TurnDiffUpdated(params) => Ok(vec![status_event(
             "turn.diff.updated",
             serde_json::to_string(params).ok(),

server/packages/universal-agent-schema/src/agents/opencode.rs

@@ -3,8 +3,9 @@ use serde_json::Value;
 use crate::opencode as schema;
 use crate::{
     ContentPart, EventConversion, ItemDeltaData, ItemEventData, ItemKind, ItemRole, ItemStatus,
-    PermissionEventData, PermissionStatus, QuestionEventData, QuestionStatus, SessionStartedData,
-    UniversalEventData, UniversalEventType, UniversalItem,
+    PermissionEventData, PermissionStatus, QuestionEventData, QuestionStatus, ReasoningVisibility,
+    SessionStartedData, TurnEventData, TurnPhase, UniversalEventData, UniversalEventType,
+    UniversalItem,
 };
 
 pub fn event_to_universal(event: &schema::Event) -> Result<Vec<EventConversion>, String> {
@@ -69,27 +70,37 @@ pub fn event_to_universal(event: &schema::Event) -> Result<Vec<EventConversion>,
                     );
                 }
                 schema::Part::ReasoningPart(reasoning_part) => {
-                    let delta_text = delta
+                    let reasoning_text = delta
                         .as_ref()
                         .cloned()
                         .unwrap_or_else(|| reasoning_part.text.clone());
-                    let stub = stub_message_item(&message_id, ItemRole::Assistant);
+                    let reasoning_id = reasoning_part.id.clone();
+                    let mut started = stub_message_item(&reasoning_id, ItemRole::Assistant);
+                    started.parent_id = Some(message_id.clone());
+                    let completed = UniversalItem {
+                        item_id: String::new(),
+                        native_item_id: Some(reasoning_id),
+                        parent_id: Some(message_id.clone()),
+                        kind: ItemKind::Message,
+                        role: Some(ItemRole::Assistant),
+                        content: vec![ContentPart::Reasoning {
+                            text: reasoning_text,
+                            visibility: ReasoningVisibility::Public,
+                        }],
+                        status: ItemStatus::Completed,
+                    };
                     events.push(
                         EventConversion::new(
                             UniversalEventType::ItemStarted,
-                            UniversalEventData::Item(ItemEventData { item: stub }),
+                            UniversalEventData::Item(ItemEventData { item: started }),
                         )
                         .synthetic()
                         .with_raw(raw.clone()),
                     );
                     events.push(
                         EventConversion::new(
-                            UniversalEventType::ItemDelta,
-                            UniversalEventData::ItemDelta(ItemDeltaData {
-                                item_id: String::new(),
-                                native_item_id: Some(message_id.clone()),
-                                delta: delta_text,
-                            }),
+                            UniversalEventType::ItemCompleted,
+                            UniversalEventData::Item(ItemEventData { item: completed }),
                         )
                         .with_native_session(session_id.clone())
                         .with_raw(raw.clone()),
@@ -207,26 +218,59 @@ pub fn event_to_universal(event: &schema::Event) -> Result<Vec<EventConversion>,
                 properties,
                 type_: _,
             } = status;
+            let status_type = serde_json::to_value(&properties.status)
+                .ok()
+                .and_then(|value| {
+                    value
+                        .get("type")
+                        .and_then(Value::as_str)
+                        .map(str::to_string)
+                });
             let detail =
                 serde_json::to_string(&properties.status).unwrap_or_else(|_| "status".to_string());
             let item = status_item("session.status", Some(detail));
-            let conversion = EventConversion::new(
+            let mut events = vec![EventConversion::new(
                 UniversalEventType::ItemCompleted,
                 UniversalEventData::Item(ItemEventData { item }),
             )
             .with_native_session(Some(properties.session_id.clone()))
-            .with_raw(raw);
-            Ok(vec![conversion])
+            .with_raw(raw.clone())];
+
+            if matches!(status_type.as_deref(), Some("busy" | "idle")) {
+                let (event_type, phase) = if status_type.as_deref() == Some("busy") {
+                    (UniversalEventType::TurnStarted, TurnPhase::Started)
+                } else {
+                    (UniversalEventType::TurnEnded, TurnPhase::Ended)
+                };
+                events.push(
+                    EventConversion::new(
+                        event_type,
+                        UniversalEventData::Turn(TurnEventData {
+                            phase,
+                            turn_id: None,
+                            metadata: Some(
+                                serde_json::to_value(&properties.status).unwrap_or(Value::Null),
+                            ),
+                        }),
+                    )
+                    .with_native_session(Some(properties.session_id.clone()))
+                    .with_raw(raw),
+                );
+            }
+            Ok(events)
         }
         schema::Event::SessionIdle(idle) => {
             let schema::EventSessionIdle {
                 properties,
                 type_: _,
             } = idle;
-            let item = status_item("session.idle", None);
             let conversion = EventConversion::new(
-                UniversalEventType::ItemCompleted,
-                UniversalEventData::Item(ItemEventData { item }),
+                UniversalEventType::TurnEnded,
+                UniversalEventData::Turn(TurnEventData {
+                    phase: TurnPhase::Ended,
+                    turn_id: None,
+                    metadata: None,
+                }),
             )
             .with_native_session(Some(properties.session_id.clone()))
             .with_raw(raw);
@@ -528,3 +572,50 @@ fn permission_from_opencode(request: &schema::PermissionRequest) -> PermissionEv
         metadata: serde_json::to_value(request).ok(),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn reasoning_part_updates_stay_typed_not_text_delta() {
+        let event = schema::Event::MessagePartUpdated(schema::EventMessagePartUpdated {
+            properties: schema::EventMessagePartUpdatedProperties {
+                delta: Some("Preparing friendly brief response".to_string()),
+                part: schema::Part::ReasoningPart(schema::ReasoningPart {
+                    id: "part_reason_1".to_string(),
+                    message_id: "msg_1".to_string(),
+                    metadata: serde_json::Map::new(),
+                    session_id: "ses_1".to_string(),
+                    text: "Preparing".to_string(),
+                    time: schema::ReasoningPartTime {
+                        end: None,
+                        start: 0.0,
+                    },
+                    type_: "reasoning".to_string(),
+                }),
+            },
+            type_: "message.part.updated".to_string(),
+        });
+
+        let converted = event_to_universal(&event).expect("conversion succeeds");
+        assert_eq!(converted.len(), 2);
+        assert!(converted
+            .iter()
+            .all(|entry| entry.event_type != UniversalEventType::ItemDelta));
+
+        let completed = converted
+            .iter()
+            .find(|entry| entry.event_type == UniversalEventType::ItemCompleted)
+            .expect("item.completed exists");
+        let UniversalEventData::Item(ItemEventData { item }) = &completed.data else {
+            panic!("expected item payload");
+        };
+        assert_eq!(item.native_item_id.as_deref(), Some("part_reason_1"));
+        assert!(matches!(
+            item.content.first(),
+            Some(ContentPart::Reasoning { text, .. })
+                if text == "Preparing friendly brief response"
+        ));
+    }
+}

server/packages/universal-agent-schema/src/lib.rs

@@ -40,6 +40,10 @@ pub enum UniversalEventType {
     SessionStarted,
     #[serde(rename = "session.ended")]
     SessionEnded,
+    #[serde(rename = "turn.started")]
+    TurnStarted,
+    #[serde(rename = "turn.ended")]
+    TurnEnded,
     #[serde(rename = "item.started")]
     ItemStarted,
     #[serde(rename = "item.delta")]
@@ -63,6 +67,7 @@ pub enum UniversalEventType {
 #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, ToSchema)]
 #[serde(untagged)]
 pub enum UniversalEventData {
+    Turn(TurnEventData),
     SessionStarted(SessionStartedData),
     SessionEnded(SessionEndedData),
     Item(ItemEventData),
@@ -93,6 +98,22 @@ pub struct SessionEndedData {
     pub stderr: Option<StderrOutput>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, ToSchema)]
+pub struct TurnEventData {
+    pub phase: TurnPhase,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub turn_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub metadata: Option<Value>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum TurnPhase {
+    Started,
+    Ended,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, ToSchema)]
 pub struct StderrOutput {
     /// First N lines of stderr (if truncated) or full stderr (if not truncated)
@@ -318,25 +339,26 @@ impl EventConversion {
     }
 }
 
-pub fn turn_completed_event() -> EventConversion {
+pub fn turn_started_event(turn_id: Option<String>, metadata: Option<Value>) -> EventConversion {
     EventConversion::new(
-        UniversalEventType::ItemCompleted,
-        UniversalEventData::Item(ItemEventData {
-            item: UniversalItem {
-                item_id: String::new(),
-                native_item_id: None,
-                parent_id: None,
-                kind: ItemKind::Status,
-                role: Some(ItemRole::System),
-                content: vec![ContentPart::Status {
-                    label: "turn.completed".to_string(),
-                    detail: None,
-                }],
-                status: ItemStatus::Completed,
-            },
+        UniversalEventType::TurnStarted,
+        UniversalEventData::Turn(TurnEventData {
+            phase: TurnPhase::Started,
+            turn_id,
+            metadata,
+        }),
+    )
+}
+
+pub fn turn_ended_event(turn_id: Option<String>, metadata: Option<Value>) -> EventConversion {
+    EventConversion::new(
+        UniversalEventType::TurnEnded,
+        UniversalEventData::Turn(TurnEventData {
+            phase: TurnPhase::Ended,
+            turn_id,
+            metadata,
         }),
     )
-    .synthetic()
 }
 
 pub fn item_from_text(role: ItemRole, text: String) -> UniversalItem {