orchestral_runtime/orchestrator/
entry.rs

1use super::*;
2
3impl Orchestrator {
4    /// Handle an event end-to-end (intent -> plan -> normalize -> execute)
5    pub async fn handle_event(
6        &self,
7        event: Event,
8    ) -> Result<OrchestratorResult, OrchestratorError> {
9        if let Some(interaction_id) = self.thread_runtime.find_resume_interaction(&event).await {
10            return self.resume_interaction(interaction_id, event).await;
11        }
12
13        self.start_new_interaction(event).await
14    }
15
16    async fn start_new_interaction(
17        &self,
18        event: Event,
19    ) -> Result<OrchestratorResult, OrchestratorError> {
20        let event_clone = event.clone();
21        tracing::info!(
22            thread_id = %event_clone.thread_id(),
23            event_type = %event_type_label(&event_clone),
24            "orchestrator received event"
25        );
26        let decision = self.thread_runtime.handle_event(event).await?;
27
28        let (interaction_id, started_kind) = match &decision {
29            HandleEventResult::Started { interaction_id } => (interaction_id.clone(), "started"),
30            HandleEventResult::Merged { interaction_id } => {
31                // Event is already persisted to the active interaction by thread_runtime.
32                // Don't start a new pipeline — the running pipeline's agent loop will
33                // see the new message in history on its next iteration.
34                self.emit_lifecycle_event(
35                    "turn_merged",
36                    Some(interaction_id.as_str()),
37                    None,
38                    Some("event merged into active interaction"),
39                    serde_json::json!({
40                        "event_type": event_type_label(&event_clone),
41                    }),
42                )
43                .await;
44                return Ok(OrchestratorResult::Merged {
45                    interaction_id: interaction_id.clone(),
46                    task_id: "".into(), // no new task created
47                    result: ExecutionResult::Completed,
48                });
49            }
50            HandleEventResult::Rejected { reason } => {
51                self.emit_lifecycle_event(
52                    "turn_rejected",
53                    None,
54                    None,
55                    Some(reason),
56                    serde_json::json!({ "reason": reason }),
57                )
58                .await;
59                return Ok(OrchestratorResult::Rejected {
60                    reason: reason.clone(),
61                });
62            }
63            HandleEventResult::Queued => {
64                self.emit_lifecycle_event(
65                    "turn_queued",
66                    None,
67                    None,
68                    Some("event queued"),
69                    Value::Null,
70                )
71                .await;
72                return Ok(OrchestratorResult::Queued);
73            }
74        };
75
76        let intent = intent_from_event(&event_clone, Some(interaction_id.to_string()))?;
77        let task = Task::new(intent);
78        self.task_store.save(&task).await?;
79        self.thread_runtime
80            .add_task_to_interaction(interaction_id.as_str(), task.id.clone())
81            .await?;
82        self.emit_lifecycle_event(
83            "turn_started",
84            Some(interaction_id.as_str()),
85            Some(task.id.as_str()),
86            Some("turn started"),
87            serde_json::json!({
88                "started_kind": started_kind,
89                "event_type": event_type_label(&event_clone),
90            }),
91        )
92        .await;
93
94        self.run_planning_pipeline(interaction_id, started_kind, task)
95            .await
96    }
97
98    async fn resume_interaction(
99        &self,
100        interaction_id: InteractionId,
101        event: Event,
102    ) -> Result<OrchestratorResult, OrchestratorError> {
103        self.thread_runtime
104            .append_event_to_interaction(interaction_id.as_str(), event.clone())
105            .await?;
106        self.thread_runtime
107            .resume_interaction(interaction_id.as_str())
108            .await?;
109
110        let interaction = self
111            .thread_runtime
112            .get_interaction(interaction_id.as_str())
113            .await
114            .ok_or_else(|| OrchestratorError::ResumeError("interaction missing".to_string()))?;
115        let task_id =
116            interaction.task_ids.last().cloned().ok_or_else(|| {
117                OrchestratorError::ResumeError("interaction has no task".to_string())
118            })?;
119
120        let mut task = self
121            .task_store
122            .load(task_id.as_str())
123            .await?
124            .ok_or_else(|| OrchestratorError::TaskNotFound(task_id.to_string()))?;
125        self.emit_lifecycle_event(
126            "turn_resumed",
127            Some(interaction_id.as_str()),
128            Some(task.id.as_str()),
129            Some("resuming waiting turn"),
130            serde_json::json!({
131                "event_type": event_type_label(&event),
132            }),
133        )
134        .await;
135        if task.plan.is_none() {
136            let intent = intent_from_event(&event, Some(interaction_id.to_string()))?;
137            let new_task = Task::new(intent);
138            self.task_store.save(&new_task).await?;
139            self.thread_runtime
140                .add_task_to_interaction(interaction_id.as_str(), new_task.id.clone())
141                .await?;
142            self.emit_lifecycle_event(
143                "turn_started",
144                Some(interaction_id.as_str()),
145                Some(new_task.id.as_str()),
146                Some("turn started from clarification follow-up"),
147                serde_json::json!({
148                    "started_kind": "merged",
149                    "event_type": event_type_label(&event),
150                    "resumed_from_task_id": task.id.clone(),
151                }),
152            )
153            .await;
154            return self
155                .run_planning_pipeline(interaction_id, "merged", new_task)
156                .await;
157        }
158        task.start_executing();
159        self.task_store.save(&task).await?;
160
161        self.emit_lifecycle_event(
162            "execution_started",
163            Some(interaction_id.as_str()),
164            Some(task.id.as_str()),
165            Some("execution resumed"),
166            serde_json::json!({ "resume": true }),
167        )
168        .await;
169        let result = self
170            .execute_existing_task(&mut task, interaction_id.as_str(), Some(&event))
171            .await?;
172        self.emit_lifecycle_event(
173            "execution_completed",
174            Some(interaction_id.as_str()),
175            Some(task.id.as_str()),
176            Some("execution completed"),
177            execution_result_metadata(&result),
178        )
179        .await;
180        self.emit_interpreted_output(interaction_id.as_str(), &task, &result)
181            .await;
182
183        Ok(OrchestratorResult::Merged {
184            interaction_id,
185            task_id: task.id.clone(),
186            result,
187        })
188    }
189}
orchestral_runtime/orchestrator/entry.rs

orchestral_runtime/orchestrator/
entry.rs
