1use std::collections::BTreeMap;
41use std::path::PathBuf;
42use std::sync::{Arc, OnceLock, RwLock};
43
44use rpi_ai::types::ThinkingLevel;
45use rpi_extensions::RuntimeActionHost;
46use rpi_harness::agent_harness::{
47 AgentHarness, HarnessRunOutcome, NavigationOutcome, SendMessageResult, WeakAgentHarness,
48};
49use rpi_harness::session::session::Session;
50use rpi_harness::types::SendMessageOptions;
51use tokio::runtime::Handle;
52
53use crate::session::{default_session_dir, open_session_by_id_in_dir};
54
55fn merge_command_list_reserved(
77 builtin: Vec<(String, String)>,
78 extension: Vec<(String, String)>,
79 templates: Vec<(String, String)>,
80 skills: Vec<(String, String)>,
81 reserved: Vec<String>,
82) -> Vec<serde_json::Value> {
83 let mut reserved_names = std::collections::BTreeSet::new();
89 for name in reserved {
90 let name = if !name.starts_with('/') {
91 format!("/{name}")
92 } else {
93 name
94 };
95 reserved_names.insert(name);
96 }
97 let mut seen = std::collections::BTreeSet::new();
98 let mut out: Vec<serde_json::Value> = Vec::new();
99 for (source, commands) in [
100 ("builtin", builtin),
101 ("extension", extension),
102 ("prompt", templates),
103 ("skill", skills),
104 ] {
105 for (name, description) in commands {
106 let name = if source != "builtin" && !name.starts_with('/') {
107 format!("/{name}")
108 } else {
109 name
110 };
111 if source != "builtin" && reserved_names.contains(&name) {
112 continue;
113 }
114 if !seen.insert(name.clone()) {
115 continue;
116 }
117 out.push(serde_json::json!({
118 "name": name,
119 "description": description,
120 "source": source,
121 }));
122 }
123 }
124 out
125}
126
127pub struct HarnessActionHost {
137 harness: Arc<OnceLock<WeakAgentHarness>>,
140 catalog: RwLock<Vec<rpi_ai::Model>>,
143 cwd: PathBuf,
146 session_dir: PathBuf,
147 fresh_session_only: bool,
148 cli_flags: BTreeMap<String, serde_json::Value>,
150 tool_context: rpi_extensions::ToolCallContext,
156 children: crate::child_agents::ChildAgents,
157 side_questions: crate::side_questions::SideQuestions,
158 #[allow(dead_code)]
159 runtime: Handle,
160 registry: std::sync::RwLock<Option<std::sync::Arc<rpi_extensions::RegistrySnapshot>>>,
168}
169
170impl HarnessActionHost {
171 pub fn new_empty(
178 catalog: Vec<rpi_ai::Model>,
179 cwd: PathBuf,
180 runtime: Handle,
181 cli_flags: BTreeMap<String, serde_json::Value>,
182 tool_context: rpi_extensions::ToolCallContext,
183 ) -> (Self, Arc<OnceLock<WeakAgentHarness>>) {
184 let harness = Arc::new(OnceLock::new());
185 (
186 Self {
187 harness: Arc::clone(&harness),
188 catalog: RwLock::new(catalog),
189 session_dir: default_session_dir(&cwd),
190 fresh_session_only: false,
191 cwd,
192 cli_flags,
193 tool_context,
194 children: crate::child_agents::ChildAgents::default(),
195 side_questions: crate::side_questions::SideQuestions::default(),
196 runtime,
197 registry: std::sync::RwLock::new(None),
198 },
199 harness,
200 )
201 }
202
203 pub fn set_catalog(&self, catalog: Vec<rpi_ai::Model>) {
204 *self
205 .catalog
206 .write()
207 .unwrap_or_else(|error| error.into_inner()) = catalog;
208 }
209
210 pub(crate) fn with_fresh_session_only(mut self, locked: bool) -> Self {
211 self.fresh_session_only = locked;
212 self
213 }
214 fn allow_session_routing(&self) -> Result<(), String> {
215 if self.fresh_session_only {
216 Err("experimental native print session forbids session routing actions".into())
217 } else {
218 Ok(())
219 }
220 }
221
222 pub(crate) fn with_session_dir(mut self, session_dir: PathBuf) -> Self {
223 self.session_dir = session_dir;
224 self
225 }
226
227 pub(crate) fn publish_registry_impl(
233 &self,
234 snapshot: Option<std::sync::Arc<rpi_extensions::RegistrySnapshot>>,
235 ) {
236 if let Ok(mut slot) = self.registry.write() {
237 *slot = snapshot;
238 }
239 }
240
241 fn registry_snapshot(&self) -> Option<std::sync::Arc<rpi_extensions::RegistrySnapshot>> {
242 self.registry.read().ok().and_then(|slot| slot.clone())
243 }
244
245 pub fn set_harness(cell: &Arc<OnceLock<WeakAgentHarness>>, harness: Arc<AgentHarness>) {
250 let _ = cell.set(harness.downgrade());
251 }
252
253 fn harness(&self) -> Result<AgentHarness, String> {
256 self.harness
257 .get()
258 .ok_or_else(|| "runtime action invoked before harness was built".to_string())?
259 .upgrade()
260 .ok_or_else(|| "runtime action invoked after harness was dropped".to_string())
261 }
262
263 fn resolve_model(&self, id: &str) -> Option<rpi_ai::Model> {
266 let catalog = self.catalog.read().ok()?;
267 catalog
268 .iter()
269 .find(|m| m.id.eq_ignore_ascii_case(id))
270 .cloned()
271 .or_else(|| {
272 catalog
273 .iter()
274 .find(|m| m.id.to_ascii_lowercase().contains(&id.to_ascii_lowercase()))
275 .cloned()
276 })
277 }
278}
279
280pub(crate) fn queued_message_payloads(
288 messages: &[rpi_agent::AgentMessage],
289) -> Vec<serde_json::Value> {
290 messages
291 .iter()
292 .map(|message| {
293 let (text, images) = match message {
294 rpi_agent::AgentMessage::User(user) => match &user.content {
295 rpi_ai::types::UserContent::Text(text) => (text.clone(), Vec::new()),
296 rpi_ai::types::UserContent::Blocks(blocks) => {
297 let text = blocks
298 .iter()
299 .filter_map(|block| match block {
300 rpi_ai::types::Content::Text(text) => Some(text.text.as_str()),
301 _ => None,
302 })
303 .collect::<Vec<_>>()
304 .join("\n");
305 let collected: Vec<_> = blocks
306 .iter()
307 .filter_map(|block| match block {
308 rpi_ai::types::Content::Image(image) => Some(image.clone()),
309 _ => None,
310 })
311 .collect();
312 (text, image_payloads(&collected))
313 }
314 },
315 _ => (String::new(), Vec::new()),
316 };
317 serde_json::json!({ "text": text, "images": images })
318 })
319 .filter(|entry| {
320 entry["text"].as_str().is_some_and(|t| !t.trim().is_empty())
321 || entry["images"].as_array().is_some_and(|i| !i.is_empty())
322 })
323 .collect()
324}
325
326fn image_payloads(images: &[rpi_ai::types::ImageContent]) -> Vec<serde_json::Value> {
328 images
329 .iter()
330 .map(|image| {
331 serde_json::json!({
332 "mediaType": image.mime_type,
333 "data": image.data,
334 })
335 })
336 .collect()
337}
338
339fn arg_str(args: &serde_json::Value, key: &str) -> Result<String, String> {
341 args.get(key)
342 .and_then(|v| v.as_str())
343 .map(|s| s.to_string())
344 .ok_or_else(|| format!("missing string field `{key}` in action args"))
345}
346
347fn arg_str_opt(args: &serde_json::Value, key: &str) -> Option<String> {
349 args.get(key)
350 .and_then(|v| v.as_str())
351 .map(|s| s.to_string())
352}
353
354fn arg_bool(args: &serde_json::Value, key: &str) -> bool {
356 args.get(key).and_then(|v| v.as_bool()).unwrap_or(false)
357}
358
359fn arg_str_array(args: &serde_json::Value, key: &str) -> Result<Vec<String>, String> {
361 args.get(key)
362 .and_then(|v| v.as_array())
363 .map(|arr| {
364 arr.iter()
365 .filter_map(|v| v.as_str().map(|s| s.to_string()))
366 .collect()
367 })
368 .ok_or_else(|| format!("missing string-array field `{key}` in action args"))
369}
370
371pub(crate) fn run_outcome_json(outcome: HarnessRunOutcome) -> serde_json::Value {
375 match outcome {
376 HarnessRunOutcome::Completed {
377 leaf_id,
378 final_entry_id,
379 final_message,
380 } => {
381 serde_json::json!({
382 "status": "completed",
383 "leafId": leaf_id,
384 "finalEntryId": final_entry_id,
385 "text": assistant_text(&final_message),
386 })
387 }
388 HarnessRunOutcome::Aborted {
389 leaf_id,
390 final_entry_id,
391 final_message,
392 } => {
393 serde_json::json!({
394 "status": "aborted",
395 "leafId": leaf_id,
396 "finalEntryId": final_entry_id,
397 "text": assistant_text(&final_message),
398 })
399 }
400 HarnessRunOutcome::Failed {
401 leaf_id,
402 error,
403 final_entry_id,
404 final_message,
405 } => {
406 serde_json::json!({
407 "status": "failed",
408 "leafId": leaf_id,
409 "error": format!("{error:?}"),
410 "finalEntryId": final_entry_id,
411 "text": final_message.map(|m| assistant_text(&m)).unwrap_or_default(),
412 })
413 }
414 HarnessRunOutcome::Suspended {
415 leaf_id,
416 final_entry_id,
417 ..
418 } => {
419 serde_json::json!({
420 "status": "suspended",
421 "leafId": leaf_id,
422 "finalEntryId": final_entry_id,
423 })
424 }
425 }
426}
427
428fn send_message_options(args: &serde_json::Value) -> Result<SendMessageOptions, String> {
431 let flat = args.get("triggerTurn").is_some() || args.get("deliverAs").is_some();
432 let value = if let Some(options) = args.get("options").filter(|value| !value.is_null()) {
433 if !options.is_object() {
434 return Err("send_message options must be an object".into());
435 }
436 if flat {
437 return Err(
438 "put triggerTurn/deliverAs either in options or at the top level, not both".into(),
439 );
440 }
441 options.clone()
442 } else {
443 let mut fields = serde_json::Map::new();
444 for key in ["triggerTurn", "deliverAs"] {
445 if let Some(value) = args.get(key) {
446 fields.insert(key.into(), value.clone());
447 }
448 }
449 serde_json::Value::Object(fields)
450 };
451 serde_json::from_value(value).map_err(|error| format!("invalid send_message options: {error}"))
452}
453
454fn send_message_input(args: &serde_json::Value) -> Result<rpi_agent::AgentMessage, String> {
455 let timestamp = std::time::SystemTime::now()
456 .duration_since(std::time::UNIX_EPOCH)
457 .unwrap_or_default()
458 .as_millis()
459 .min(i64::MAX as u128) as i64;
460 if let Some(message) = args.get("message") {
461 if message.get("kind").is_some() {
462 return serde_json::from_value(message.clone())
463 .map_err(|error| format!("invalid message: {error}"));
464 }
465 #[derive(serde::Deserialize)]
468 #[serde(rename_all = "camelCase")]
469 struct CustomInput {
470 custom_type: String,
471 #[serde(default)]
472 content: Option<rpi_ai::types::UserContent>,
473 #[serde(default)]
474 display: bool,
475 #[serde(default)]
476 details: Option<serde_json::Value>,
477 }
478 let input: CustomInput = serde_json::from_value(message.clone())
479 .map_err(|error| format!("invalid custom message: {error}"))?;
480 return Ok(rpi_harness::messages::create_custom_message(
481 &input.custom_type,
482 input
483 .content
484 .unwrap_or(rpi_ai::types::UserContent::Blocks(Vec::new())),
485 input.display,
486 input.details,
487 timestamp,
488 ));
489 }
490 let text =
491 arg_str(args, "text").map_err(|_| "missing `message` or `text` field".to_string())?;
492 Ok(rpi_agent::AgentMessage::User(
493 rpi_ai::types::UserMessage::new(text, timestamp),
494 ))
495}
496
497fn send_message_result(result: SendMessageResult) -> serde_json::Value {
498 match result {
499 SendMessageResult::Appended { entry_id } => {
500 serde_json::json!({"status":"appended", "entryId":entry_id})
501 }
502 SendMessageResult::Queued { entry_id, queue } => {
503 if queue == rpi_harness::session::types::QueueKind::Context {
504 return serde_json::json!({"status":"queued", "entryId":entry_id, "queue":"context"});
505 }
506 let delivery = match queue {
507 rpi_harness::session::types::QueueKind::NextRun => "nextTurn",
508 _ => queue.as_str(),
509 };
510 serde_json::json!({"status":"queued", "entryId":entry_id, "deliverAs":delivery})
511 }
512 SendMessageResult::Run(run) => run_outcome_json(run.outcome),
513 }
514}
515
516pub(crate) fn assistant_text(msg: &rpi_ai::types::AssistantMessage) -> String {
518 msg.content
519 .iter()
520 .filter_map(|b| match b {
521 rpi_ai::types::Content::Text(t) => Some(t.text.as_str()),
522 _ => None,
523 })
524 .collect::<Vec<_>>()
525 .join("")
526}
527
528#[async_trait::async_trait]
529impl RuntimeActionHost for HarnessActionHost {
530 fn cancel_side_questions(&self, owner: u64) {
531 self.side_questions.cancel_owner(owner);
532 }
533 fn side_question_revision(&self) -> u64 {
534 self.side_questions.revision()
535 }
536 fn side_question_views(&self) -> Vec<serde_json::Value> {
537 self.side_questions.views()
538 }
539 async fn side_question_ui(
540 &self,
541 owner: u64,
542 thread_id: String,
543 args: serde_json::Value,
544 ) -> Result<serde_json::Value, String> {
545 self.side_questions
546 .ui_handle(&self.harness()?, owner, thread_id, args)
547 .await
548 }
549 async fn side_question(
550 &self,
551 owner: u64,
552 args: serde_json::Value,
553 sink: rpi_extensions::SideQuestionSink,
554 ) -> Result<serde_json::Value, String> {
555 self.side_questions
556 .handle(&self.harness()?, owner, args, sink)
557 .await
558 }
559
560 fn cancel_child_agents(&self, owner: u64) {
561 self.children.cancel_owner(owner);
562 }
563 fn replace_model_catalog(&self, models: Vec<rpi_ai::Model>) {
564 self.set_catalog(models);
565 }
566 async fn list_commands(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
567 let builtin = crate::interactive_tui::build_builtin_commands_for_abi();
568 let extension: Vec<(String, String)> = self
569 .registry_snapshot()
570 .as_deref()
571 .map(|snapshot| {
572 snapshot
573 .commands()
574 .iter()
575 .map(|c| (c.name.clone(), c.description.clone()))
576 .collect()
577 })
578 .unwrap_or_default();
579 let resources = match self.harness() {
583 Ok(harness) => match harness.get_resources().await {
584 Ok(resources) => crate::resource_commands::resource_command_entries(&resources),
585 Err(_) => Default::default(),
586 },
587 Err(_) => Default::default(),
588 };
589 Ok(serde_json::json!({
590 "commands": merge_command_list_reserved(
591 builtin,
592 extension,
593 resources.templates,
594 resources.skills,
595 crate::interactive_tui::builtin_reserved_names(),
596 )
597 }))
598 }
599
600 fn publish_registry(&self, snapshot: std::sync::Arc<rpi_extensions::RegistrySnapshot>) {
601 self.publish_registry_impl(Some(snapshot));
602 }
603
604 fn publish_registry_opt(
605 &self,
606 snapshot: Option<std::sync::Arc<rpi_extensions::RegistrySnapshot>>,
607 ) {
608 self.publish_registry_impl(snapshot);
609 }
610
611 async fn abort(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
612 let harness = self.harness()?;
628 let result = harness
629 .abort_reclaiming(true)
630 .await
631 .map_err(|error| error.to_string())?;
632 Ok(serde_json::json!({
633 "aborted": !result.run_id.is_empty(),
637 "runId": if result.run_id.is_empty() {
638 serde_json::Value::Null
639 } else {
640 serde_json::json!(result.run_id)
641 },
642 "steer": queued_message_payloads(&result.steer),
643 "followUp": queued_message_payloads(&result.follow_up),
644 }))
645 }
646
647 fn stop_round(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
648 self.harness()?.request_stop_round();
649 Ok(serde_json::json!({ "requested": true }))
650 }
651
652 async fn get_execution_context(
653 &self,
654 _args: serde_json::Value,
655 ) -> Result<serde_json::Value, String> {
656 let harness = self.harness()?;
657 let mut context = harness.execution_context().map_err(|e| e.to_string())?;
658 if let Some(serde_json::Value::Object(child)) =
659 rpi_agent::tool_execution_scope::execution_context()
660 {
661 context
662 .as_object_mut()
663 .expect("execution context is an object")
664 .extend(child);
665 }
666 context["cwd"] = serde_json::json!(self.cwd);
667 context["sessionId"] = serde_json::json!(harness.session().storage().metadata().id);
668 Ok(context)
669 }
670
671 async fn child_agent(
672 &self,
673 owner: u64,
674 args: serde_json::Value,
675 ) -> Result<serde_json::Value, String> {
676 if rpi_agent::tool_execution_scope::is_child_execution() {
677 return Err("recursive child agents are not allowed".into());
678 }
679 if args.get("op").and_then(serde_json::Value::as_str) == Some("start") {
680 self.children.handle(&self.harness()?, owner, args).await
681 } else {
682 self.children.handle_existing(owner, args)
684 }
685 }
686
687 async fn get_cli_flag(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
688 let name = arg_str(&args, "name")?;
689 Ok(serde_json::json!({
690 "value": self.cli_flags.get(&name).cloned().unwrap_or(serde_json::Value::Null)
691 }))
692 }
693
694 async fn send_message(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
695 let options = send_message_options(&args)?;
696 let message = send_message_input(&args)?;
697 let lane = self.harness()?.lane("main");
698 let result = lane
699 .send_message(message, options)
700 .await
701 .map_err(|e| e.to_string())?;
702 Ok(send_message_result(result))
703 }
704
705 async fn send_user_message(
706 &self,
707 args: serde_json::Value,
708 ) -> Result<serde_json::Value, String> {
709 let text = arg_str(&args, "text")?;
710 let lane = self.harness()?.lane("main");
711 let mut options = send_message_options(&args)?;
727 if options.trigger_turn.is_none() {
728 options.trigger_turn = Some(true);
729 }
730 let message = rpi_agent::AgentMessage::User(rpi_ai::types::UserMessage::new(
731 text,
732 std::time::SystemTime::now()
733 .duration_since(std::time::UNIX_EPOCH)
734 .unwrap_or_default()
735 .as_millis()
736 .min(i64::MAX as u128) as i64,
737 ));
738 let result = lane
739 .send_message(message, options)
740 .await
741 .map_err(|e| e.to_string())?;
742 Ok(send_message_result(result))
743 }
744
745 async fn append_entry(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
746 if let Some(expected) = args.get("expected") {
752 let guard = serde_json::from_value(expected.clone())
753 .map_err(|error| format!("invalid AppendEntry expected guard: {error}"))?;
754 let entry = if let Some(custom_type) = arg_str_opt(&args, "customType") {
755 if args.get("message").is_some() || args.get("outboxKey").is_some() {
756 return Err("scoped custom entry cannot include message/outboxKey".into());
757 }
758 crate::resource_selection::validate_public_entry(&custom_type, args.get("data"))?;
759 rpi_harness::agent_harness::ScopedEntry::Custom {
760 custom_type,
761 data: args.get("data").cloned(),
762 }
763 } else {
764 let message =
765 serde_json::from_value(args.get("message").cloned().ok_or("missing message")?)
766 .map_err(|error| format!("invalid message: {error}"))?;
767 let key = arg_str(&args, "outboxKey")?;
768 rpi_harness::agent_harness::ScopedEntry::Notice { message, key }
769 };
770 let id = self
771 .harness()?
772 .append_scoped_entry(&guard, entry)
773 .await
774 .map_err(|error| error.to_string())?;
775 return Ok(serde_json::json!({"entryId":id, "scoped":true}));
776 }
777 if args.get("outboxKey").is_some() {
778 return Err("outboxKey requires an expected guard".into());
779 }
780 if let Some(custom_type) = arg_str_opt(&args, "customType") {
781 crate::resource_selection::validate_public_entry(&custom_type, args.get("data"))?;
782 let data = args.get("data").cloned();
783 let id = self
784 .harness()?
785 .append_custom_entry(&custom_type, data)
786 .await
787 .map_err(|e| e.to_string())?;
788 return Ok(serde_json::json!({ "entryId": id }));
789 }
790 let msg = args
791 .get("message")
792 .ok_or_else(|| "missing `message` or `customType` field".to_string())?;
793 let message: rpi_agent::AgentMessage =
794 serde_json::from_value(msg.clone()).map_err(|e| format!("invalid message: {e}"))?;
795 let id = self
796 .harness()?
797 .append_message(message)
798 .await
799 .map_err(|e| e.to_string())?;
800 Ok(serde_json::json!({ "entryId": id }))
801 }
802
803 async fn get_session_branch(
804 &self,
805 args: serde_json::Value,
806 ) -> Result<serde_json::Value, String> {
807 let request = crate::modes::branch_entry_query(&args)?;
808 let (entries, session_id, leaf_id) =
809 crate::modes::branch_snapshot(&self.harness()?, &request).await?;
810 let entries = entries
811 .iter()
812 .map(serde_json::to_value)
813 .collect::<Result<Vec<_>, _>>()
814 .map_err(|e| format!("serialize session branch: {e}"))?;
815 Ok(serde_json::json!({
816 "entries": entries,
817 "count": entries.len(),
818 "filtered": true,
819 "sessionId": session_id,
820 "leafId": leaf_id,
821 "scopedAppend": true,
822 }))
823 }
824
825 async fn set_session_name(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
826 let name = arg_str(&args, "name")?;
827 self.harness()?
828 .session()
829 .set_name(Some(&name))
830 .await
831 .map_err(|e| e.to_string())?;
832 Ok(serde_json::Value::Null)
833 }
834
835 async fn get_active_tools(
836 &self,
837 _args: serde_json::Value,
838 ) -> Result<serde_json::Value, String> {
839 let lane = self.harness()?.lane("main");
840 let tools = lane.get_active_tools().await.map_err(|e| e.to_string())?;
841 Ok(serde_json::json!({ "tools": tools }))
842 }
843
844 async fn set_active_tools(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
845 let tools = arg_str_array(&args, "tools")?;
846 let lane = self.harness()?.lane("main");
847 lane.set_active_tools(tools)
848 .await
849 .map_err(|e| e.to_string())?;
850 Ok(serde_json::Value::Null)
851 }
852
853 async fn set_model(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
854 let id = arg_str(&args, "model")?;
855 let model = self
856 .resolve_model(&id)
857 .ok_or_else(|| format!("model `{id}` not in catalog"))?;
858 let lane = self.harness()?.lane("main");
859 lane.set_model(model.clone())
860 .await
861 .map_err(|e| e.to_string())?;
862 Ok(serde_json::json!({ "model": model.id }))
863 }
864
865 async fn get_thinking_level(
866 &self,
867 _args: serde_json::Value,
868 ) -> Result<serde_json::Value, String> {
869 let lane = self.harness()?.lane("main");
870 let level = lane.get_thinking_level().await.map_err(|e| e.to_string())?;
871 Ok(serde_json::json!({ "level": level }))
872 }
873
874 async fn set_thinking_level(
875 &self,
876 args: serde_json::Value,
877 ) -> Result<serde_json::Value, String> {
878 let level_val = args
879 .get("level")
880 .ok_or_else(|| "missing `level` field".to_string())?;
881 let level: ThinkingLevel = if let Some(s) = level_val.as_str() {
882 serde_json::from_value(serde_json::Value::String(s.to_string()))
883 .map_err(|e| format!("invalid thinking level `{s}`: {e}"))?
884 } else {
885 serde_json::from_value(level_val.clone())
886 .map_err(|e| format!("invalid thinking level: {e}"))?
887 };
888 let lane = self.harness()?.lane("main");
889 lane.set_thinking_level(level)
890 .await
891 .map_err(|e| e.to_string())?;
892 Ok(serde_json::Value::Null)
893 }
894
895 async fn compact(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
896 let custom = arg_str_opt(&args, "customInstructions");
897 let lane = self.harness()?.lane("main");
898 let result = lane
899 .compact(custom.as_deref())
900 .await
901 .map_err(|e| e.to_string())?;
902 Ok(
903 serde_json::json!({ "runId": result.run_id, "outcome": format!("{:?}", result.outcome) }),
904 )
905 }
906
907 async fn get_system_prompt(
908 &self,
909 _args: serde_json::Value,
910 ) -> Result<serde_json::Value, String> {
911 let prompt = self
912 .harness()?
913 .get_system_prompt()
914 .await
915 .map_err(|e| e.to_string())?;
916 Ok(serde_json::json!({ "prompt": prompt }))
917 }
918
919 async fn new_session(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
920 self.allow_session_routing()?;
921 let harness = self.harness()?;
922 let cwd_str = self.cwd.to_string_lossy().to_string();
923 let dir = &self.session_dir;
924 std::fs::create_dir_all(dir)
925 .map_err(|e| format!("create session dir {}: {e}", dir.display()))?;
926 let session = crate::session::create_jsonl_session(dir, &cwd_str)
927 .await
928 .map_err(|e| format!("create session: {e}"))?;
929 let id = session.storage().metadata().id.clone();
930 crate::session::set_active_session(&harness, session, Some(&self.tool_context)).await?;
931 Ok(serde_json::json!({ "sessionId": id }))
932 }
933
934 async fn fork(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
935 self.allow_session_routing()?;
936 let harness = self.harness()?;
937 let cwd_str = self.cwd.to_string_lossy().to_string();
938 let new_session =
939 crate::session::fork_session_storage(&harness, &cwd_str, &self.session_dir)
940 .await
941 .map_err(|e| format!("fork session: {e}"))?;
942 let id = new_session.storage().metadata().id.clone();
943 crate::session::set_active_session(&harness, new_session, Some(&self.tool_context)).await?;
944 Ok(serde_json::json!({ "sessionId": id }))
945 }
946
947 async fn navigate_tree(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
948 self.allow_session_routing()?;
949 let target_id = arg_str_opt(&args, "targetId");
950 let summarize = arg_bool(&args, "summarize");
951 let custom = arg_str_opt(&args, "customInstructions");
952 let label = arg_str_opt(&args, "label");
953 let lane = self.harness()?.lane("main");
954 let result = lane
955 .navigate_tree(
956 target_id.as_deref(),
957 summarize,
958 custom.as_deref(),
959 label.as_deref(),
960 )
961 .await
962 .map_err(|e| e.to_string())?;
963 let status = match &result.outcome {
964 NavigationOutcome::Completed { .. } => "completed",
965 NavigationOutcome::Declined { .. } => "declined",
966 NavigationOutcome::Aborted { .. } => "aborted",
967 NavigationOutcome::Failed { .. } => "failed",
968 };
969 Ok(serde_json::json!({ "runId": result.run_id, "status": status }))
970 }
971
972 async fn switch_session(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
973 self.allow_session_routing()?;
974 let harness = self.harness()?;
975 let id = arg_str(&args, "id")?;
976 let cwd_str = self.cwd.to_string_lossy().to_string();
977 let new_session: Session = open_session_by_id_in_dir(&id, &cwd_str, &self.session_dir)
978 .await
979 .map(|restored| restored.session)
980 .map_err(|e| e.to_string())?;
981 let new_id = new_session.storage().metadata().id.clone();
982 crate::session::set_active_session(&harness, new_session, Some(&self.tool_context)).await?;
983 Ok(serde_json::json!({ "sessionId": new_id }))
984 }
985
986 async fn reload(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
987 Err("reload not configured (no /reload callback on this bridge)".to_string())
991 }
992}
993
994#[cfg(test)]
995mod lifetime_tests;
996
997#[cfg(test)]
998mod send_message_tests {
999 use super::*;
1000 use rpi_harness::types::MessageDelivery;
1001
1002 #[test]
1003 fn options_accept_flat_and_nested_forms_with_pi_defaults() {
1004 assert_eq!(
1005 send_message_options(&serde_json::json!({"text":"context"})).unwrap(),
1006 SendMessageOptions::default()
1007 );
1008 let expected = SendMessageOptions {
1009 trigger_turn: Some(false),
1010 deliver_as: MessageDelivery::FollowUp,
1011 };
1012 for args in [
1013 serde_json::json!({"text":"context", "triggerTurn":false, "deliverAs":"followUp"}),
1014 serde_json::json!({"text":"context", "options":{"triggerTurn":false, "deliverAs":"followUp"}}),
1015 ] {
1016 assert_eq!(send_message_options(&args).unwrap(), expected);
1017 }
1018 }
1019
1020 #[test]
1021 fn options_reject_invalid_or_ambiguous_scheduling_instructions() {
1022 for args in [
1023 serde_json::json!({"triggerTurn":"false"}),
1024 serde_json::json!({"deliverAs":"follow-up"}),
1025 serde_json::json!({"options":{"typo":true}}),
1026 serde_json::json!({"options":[], "text":"context"}),
1027 serde_json::json!({"options":{"triggerTurn":true}, "triggerTurn":false}),
1028 ] {
1029 assert!(send_message_options(&args).is_err(), "{args}");
1030 }
1031 }
1032
1033 #[test]
1034 fn pi_custom_input_preserves_content_and_display_metadata() {
1035 let message = send_message_input(&serde_json::json!({"message":{
1036 "customType":"goal", "content":"goal state", "display":true, "details":{"step":2}
1037 }}))
1038 .unwrap();
1039 let rpi_agent::AgentMessage::Custom(custom) = &message else {
1040 panic!("custom")
1041 };
1042 assert_eq!(
1043 custom.content,
1044 vec![rpi_ai::types::Content::text("goal state")]
1045 );
1046 let data = rpi_harness::messages::custom_data(&message).unwrap();
1047 assert_eq!(data.custom_type, "goal");
1048 assert!(data.display);
1049 assert_eq!(data.details, Some(serde_json::json!({"step":2})));
1050 let stored = serde_json::to_value(&message).unwrap();
1052 assert_eq!(
1053 send_message_input(&serde_json::json!({"message":stored})).unwrap(),
1054 message
1055 );
1056 }
1057
1058 #[test]
1059 fn missing_custom_content_is_empty_and_queued_results_use_pi_delivery_names() {
1060 let message = send_message_input(
1061 &serde_json::json!({"message":{"customType":"goal", "content":null}}),
1062 )
1063 .unwrap();
1064 assert!(
1065 matches!(message, rpi_agent::AgentMessage::Custom(custom) if custom.content.is_empty())
1066 );
1067 assert_eq!(
1068 send_message_result(SendMessageResult::Queued {
1069 entry_id: "reserved".into(),
1070 queue: rpi_harness::session::types::QueueKind::NextRun,
1071 }),
1072 serde_json::json!({"status":"queued", "entryId":"reserved", "deliverAs":"nextTurn"})
1073 );
1074 }
1075}
1076
1077#[cfg(test)]
1078mod list_commands_tests {
1079 use super::*;
1080 use rpi_harness::types::{AgentHarnessOptions, AgentHarnessResources, PromptTemplate, Skill};
1081
1082 #[tokio::test]
1083 async fn review_list_commands_must_include_pi_style_skill_names() {
1084 let root = tempfile::tempdir().unwrap();
1085 let harness = Arc::new(
1086 AgentHarness::create(AgentHarnessOptions {
1087 resources: AgentHarnessResources {
1088 skills: Some(vec![Skill {
1089 name: "manual".into(),
1090 description: "Explicit only".into(),
1091 content: "body".into(),
1092 file_path: "/skills/manual/SKILL.md".into(),
1093 disable_model_invocation: Some(true),
1094 }]),
1095 ..Default::default()
1096 },
1097 ..Default::default()
1098 })
1099 .await
1100 .unwrap(),
1101 );
1102 let (host, cell) = HarnessActionHost::new_empty(
1103 vec![],
1104 root.path().to_path_buf(),
1105 Handle::current(),
1106 Default::default(),
1107 rpi_extensions::ToolCallContext::new(root.path().to_string_lossy()),
1108 );
1109 HarnessActionHost::set_harness(&cell, harness.clone());
1110 let listing = host.list_commands(serde_json::json!({})).await.unwrap();
1111 let command = listing["commands"]
1112 .as_array()
1113 .unwrap()
1114 .iter()
1115 .find(|item| item["name"] == "/skill:manual")
1116 .expect("loaded explicit-only skill must be discoverable");
1117 assert_eq!(command["source"], "skill");
1118 assert_eq!(command["description"], "Explicit only");
1119 }
1120
1121 fn skill(name: &str, description: &str) -> Skill {
1122 Skill {
1123 name: name.into(),
1124 description: description.into(),
1125 content: "body".into(),
1126 file_path: format!("/skills/{name}/SKILL.md"),
1127 disable_model_invocation: Some(false),
1128 }
1129 }
1130
1131 fn template(name: &str) -> PromptTemplate {
1132 PromptTemplate {
1133 name: name.into(),
1134 description: None,
1135 content: "body".into(),
1136 }
1137 }
1138
1139 fn host(root: &std::path::Path) -> (HarnessActionHost, Arc<OnceLock<WeakAgentHarness>>) {
1140 let (host, cell) = HarnessActionHost::new_empty(
1141 vec![],
1142 root.to_path_buf(),
1143 Handle::current(),
1144 Default::default(),
1145 rpi_extensions::ToolCallContext::new(root.to_string_lossy()),
1146 );
1147 extern "C" fn unused_command(
1148 _: rpi_plugin_sdk::StbStringRef,
1149 _: *mut rpi_plugin_sdk::StbString,
1150 _: *mut std::ffi::c_void,
1151 ) -> i32 {
1152 0
1153 }
1154 let mut registry = rpi_extensions::ExtensionRegistry::new();
1155 for name in ["deploy", "help", "q", "skill:extension"] {
1156 registry.register_command(
1157 name.into(),
1158 "Extension command".into(),
1159 unused_command,
1160 std::ptr::null_mut(),
1161 );
1162 }
1163 host.publish_registry(Arc::new(registry.snapshot()));
1164 (host, cell)
1165 }
1166
1167 #[tokio::test]
1168 async fn list_commands_merges_sources_deduplicates_and_preserves_priority() {
1169 let root = tempfile::tempdir().unwrap();
1170 let harness = Arc::new(
1171 AgentHarness::create(AgentHarnessOptions {
1172 resources: AgentHarnessResources {
1173 skills: Some(vec![
1174 skill("extension", "shadowed"),
1175 skill("prompt", "shadowed"),
1176 skill("manual", "First skill"),
1177 skill("manual", "Duplicate skill"),
1178 ]),
1179 prompt_templates: Some(vec![
1180 template("help"),
1181 template("q"),
1182 template("deploy"),
1183 template("skill:prompt"),
1184 template("summarize"),
1185 template("/summarize"),
1186 ]),
1187 ..Default::default()
1188 },
1189 ..Default::default()
1190 })
1191 .await
1192 .unwrap(),
1193 );
1194 let (host, cell) = host(root.path());
1195 HarnessActionHost::set_harness(&cell, harness.clone());
1196 let listing = host.list_commands(serde_json::json!({})).await.unwrap();
1197 let commands = listing["commands"].as_array().unwrap();
1198 let find = |name| commands.iter().find(|c| c["name"] == name).unwrap();
1199 assert_eq!(find("/help")["source"], "builtin");
1200 assert_eq!(find("/deploy")["source"], "extension");
1201 assert_eq!(find("/skill:extension")["source"], "extension");
1202 assert_eq!(find("/skill:prompt")["source"], "prompt");
1203 assert_eq!(find("/summarize")["description"], "Expand prompt template");
1204 assert_eq!(find("/skill:manual")["source"], "skill");
1205 assert_eq!(find("/skill:manual")["description"], "First skill");
1206 assert!(!names(commands).contains(&"/q"));
1207 let unique: std::collections::BTreeSet<_> = names(commands).into_iter().collect();
1208 assert_eq!(unique.len(), commands.len());
1209 assert_eq!(commands.last().unwrap()["name"], "/skill:manual");
1210 }
1211
1212 #[tokio::test]
1213 async fn list_commands_reads_reloaded_resources_and_removes_stale_skills() {
1214 let root = tempfile::tempdir().unwrap();
1215 let harness = Arc::new(AgentHarness::create(Default::default()).await.unwrap());
1216 let (host, cell) = host(root.path());
1217 HarnessActionHost::set_harness(&cell, harness.clone());
1218 for name in ["oldskill", "freshskill"] {
1219 harness
1220 .set_resources(AgentHarnessResources {
1221 skills: Some(vec![skill(name, "Reloaded skill")]),
1222 ..Default::default()
1223 })
1224 .await
1225 .unwrap();
1226 let listing = host.list_commands(serde_json::json!({})).await.unwrap();
1227 let commands = listing["commands"].as_array().unwrap();
1228 assert!(names(commands).contains(&format!("/skill:{name}").as_str()));
1229 if name == "freshskill" {
1230 assert!(!names(commands).contains(&"/skill:oldskill"));
1231 }
1232 }
1233 harness.set_resources(Default::default()).await.unwrap();
1234 let listing = host.list_commands(serde_json::json!({})).await.unwrap();
1235 assert!(!names(listing["commands"].as_array().unwrap()).contains(&"/skill:freshskill"));
1236 }
1237
1238 #[tokio::test]
1239 async fn list_commands_without_live_harness_keeps_builtin_and_extension_entries() {
1240 let root = tempfile::tempdir().unwrap();
1241 let (host, cell) = host(root.path());
1242 let unbound = host.list_commands(serde_json::json!({})).await.unwrap();
1243 let commands = unbound["commands"].as_array().unwrap();
1244 assert!(commands
1245 .iter()
1246 .any(|c| c["name"] == "/help" && c["source"] == "builtin"));
1247 assert!(commands
1248 .iter()
1249 .any(|c| c["name"] == "/deploy" && c["source"] == "extension"));
1250 let harness = Arc::new(
1251 AgentHarness::create(AgentHarnessOptions {
1252 resources: AgentHarnessResources {
1253 skills: Some(vec![skill("manual", "Manual")]),
1254 prompt_templates: Some(vec![template("summarize")]),
1255 ..Default::default()
1256 },
1257 ..Default::default()
1258 })
1259 .await
1260 .unwrap(),
1261 );
1262 HarnessActionHost::set_harness(&cell, harness.clone());
1263 harness.close().await.unwrap();
1264 assert_eq!(
1265 host.list_commands(serde_json::json!({})).await.unwrap(),
1266 unbound
1267 );
1268 drop(harness);
1269 assert_eq!(
1270 host.list_commands(serde_json::json!({})).await.unwrap(),
1271 unbound
1272 );
1273 }
1274
1275 #[test]
1281 fn reserved_names_do_not_remove_the_builtins_themselves() {
1282 let builtin = vec![
1283 ("/help".to_string(), "help".to_string()),
1284 ("/model".to_string(), "model".to_string()),
1285 ];
1286 let out = merge_command_list_reserved(
1287 builtin,
1288 vec![("deploy".into(), "deploy".into())],
1289 vec![],
1290 vec![],
1291 vec![
1294 "/help".to_string(),
1295 "/model".to_string(),
1296 "/q".to_string(),
1297 "/context".to_string(),
1298 ],
1299 );
1300 assert_eq!(names(&out), vec!["/help", "/model", "/deploy"]);
1301 assert_eq!(out[0]["source"], "builtin");
1302 assert_eq!(out[1]["source"], "builtin");
1303 assert_eq!(out[2]["source"], "extension");
1304 }
1305
1306 #[test]
1310 fn reserved_builtin_aliases_hide_colliding_extension_entries() {
1311 let out = merge_command_list_reserved(
1312 vec![("/help".into(), "help".into())],
1313 vec![
1314 ("q".into(), "quit".into()),
1315 ("context".into(), "context".into()),
1316 ("deploy".into(), "deploy".into()),
1317 ],
1318 vec![],
1319 vec![],
1320 vec!["/q".into(), "/context".into()],
1321 );
1322 assert_eq!(names(&out), vec!["/help", "/deploy"]);
1324 assert_eq!(out[1]["source"], "extension");
1325 }
1326
1327 fn names(v: &[serde_json::Value]) -> Vec<&str> {
1328 v.iter().map(|c| c["name"].as_str().unwrap()).collect()
1329 }
1330
1331 #[test]
1332 fn builtin_wins_over_a_same_named_extension_command() {
1333 let out = merge_command_list_reserved(
1334 vec![("/model".into(), "built-in".into())],
1335 vec![("/model".into(), "extension".into())],
1336 vec![],
1337 vec![],
1338 Vec::new(),
1339 );
1340 assert_eq!(names(&out), vec!["/model"]);
1341 assert_eq!(out[0]["source"], "builtin");
1342 assert_eq!(out[0]["description"], "built-in");
1343 }
1344
1345 #[test]
1346 fn extension_and_template_names_get_a_leading_slash() {
1347 let out = merge_command_list_reserved(
1348 vec![],
1349 vec![("deploy".into(), "deploy".into())],
1350 vec![("review".into(), "review".into())],
1351 vec![],
1352 Vec::new(),
1353 );
1354 assert_eq!(names(&out), vec!["/deploy", "/review"]);
1355 assert_eq!(out[0]["source"], "extension");
1356 assert_eq!(out[1]["source"], "prompt");
1357 }
1358
1359 #[test]
1363 fn prompt_templates_are_listed_with_their_own_source() {
1364 let out = merge_command_list_reserved(
1365 vec![("/help".into(), "help".into())],
1366 vec![],
1367 vec![("review".into(), "review a diff".into())],
1368 vec![],
1369 Vec::new(),
1370 );
1371 assert_eq!(names(&out), vec!["/help", "/review"]);
1372 assert_eq!(out[1]["source"], "prompt");
1373 assert_eq!(out[1]["description"], "review a diff");
1374 }
1375
1376 #[test]
1379 fn a_builtin_name_is_not_reported_as_a_template() {
1380 let out = merge_command_list_reserved(
1381 vec![("/review".into(), "built-in".into())],
1382 vec![],
1383 vec![("review".into(), "template".into())],
1384 vec![],
1385 Vec::new(),
1386 );
1387 assert_eq!(names(&out), vec!["/review"]);
1388 assert_eq!(out[0]["source"], "builtin");
1389 }
1390
1391 #[test]
1392 fn no_commands_is_an_empty_list_not_an_error() {
1393 assert!(merge_command_list_reserved(vec![], vec![], vec![], vec![], Vec::new()).is_empty());
1394 }
1395}