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 async fn tool_program(
531 &self,
532 capability: Arc<rpi_agent::tool_program::ToolProgram>,
533 args: serde_json::Value,
534 ) -> Result<serde_json::Value, String> {
535 capability.handle(args).await
537 }
538 fn cancel_side_questions(&self, owner: u64) {
539 self.side_questions.cancel_owner(owner);
540 }
541 fn side_question_revision(&self) -> u64 {
542 self.side_questions.revision()
543 }
544 fn side_question_views(&self) -> Vec<serde_json::Value> {
545 self.side_questions.views()
546 }
547 async fn side_question_ui(
548 &self,
549 owner: u64,
550 thread_id: String,
551 args: serde_json::Value,
552 ) -> Result<serde_json::Value, String> {
553 self.side_questions
554 .ui_handle(&self.harness()?, owner, thread_id, args)
555 .await
556 }
557 async fn side_question(
558 &self,
559 owner: u64,
560 args: serde_json::Value,
561 sink: rpi_extensions::SideQuestionSink,
562 ) -> Result<serde_json::Value, String> {
563 self.side_questions
564 .handle(&self.harness()?, owner, args, sink)
565 .await
566 }
567
568 fn cancel_child_agents(&self, owner: u64) {
569 self.children.cancel_owner(owner);
570 }
571 fn replace_model_catalog(&self, models: Vec<rpi_ai::Model>) {
572 self.set_catalog(models);
573 }
574 async fn list_commands(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
575 let builtin = crate::interactive_tui::build_builtin_commands_for_abi();
576 let extension: Vec<(String, String)> = self
577 .registry_snapshot()
578 .as_deref()
579 .map(|snapshot| {
580 snapshot
581 .commands()
582 .iter()
583 .map(|c| (c.name.clone(), c.description.clone()))
584 .collect()
585 })
586 .unwrap_or_default();
587 let resources = match self.harness() {
591 Ok(harness) => match harness.get_resources().await {
592 Ok(resources) => crate::resource_commands::resource_command_entries(&resources),
593 Err(_) => Default::default(),
594 },
595 Err(_) => Default::default(),
596 };
597 Ok(serde_json::json!({
598 "commands": merge_command_list_reserved(
599 builtin,
600 extension,
601 resources.templates,
602 resources.skills,
603 crate::interactive_tui::builtin_reserved_names(),
604 )
605 }))
606 }
607
608 fn publish_registry(&self, snapshot: std::sync::Arc<rpi_extensions::RegistrySnapshot>) {
609 self.publish_registry_impl(Some(snapshot));
610 }
611
612 fn publish_registry_opt(
613 &self,
614 snapshot: Option<std::sync::Arc<rpi_extensions::RegistrySnapshot>>,
615 ) {
616 self.publish_registry_impl(snapshot);
617 }
618
619 async fn abort(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
620 let harness = self.harness()?;
636 let result = harness
637 .abort_reclaiming(true)
638 .await
639 .map_err(|error| error.to_string())?;
640 Ok(serde_json::json!({
641 "aborted": !result.run_id.is_empty(),
645 "runId": if result.run_id.is_empty() {
646 serde_json::Value::Null
647 } else {
648 serde_json::json!(result.run_id)
649 },
650 "steer": queued_message_payloads(&result.steer),
651 "followUp": queued_message_payloads(&result.follow_up),
652 }))
653 }
654
655 fn stop_round(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
656 self.harness()?.request_stop_round();
657 Ok(serde_json::json!({ "requested": true }))
658 }
659
660 async fn get_execution_context(
661 &self,
662 _args: serde_json::Value,
663 ) -> Result<serde_json::Value, String> {
664 let harness = self.harness()?;
665 let mut context = harness.execution_context().map_err(|e| e.to_string())?;
666 if let Some(serde_json::Value::Object(child)) =
667 rpi_agent::tool_execution_scope::execution_context()
668 {
669 context
670 .as_object_mut()
671 .expect("execution context is an object")
672 .extend(child);
673 }
674 context["cwd"] = serde_json::json!(self.cwd);
675 context["sessionId"] = serde_json::json!(harness.session().storage().metadata().id);
676 Ok(context)
677 }
678
679 async fn child_agent(
680 &self,
681 owner: u64,
682 args: serde_json::Value,
683 ) -> Result<serde_json::Value, String> {
684 if rpi_agent::tool_execution_scope::is_child_execution() {
685 return Err("recursive child agents are not allowed".into());
686 }
687 if args.get("op").and_then(serde_json::Value::as_str) == Some("start") {
688 self.children.handle(&self.harness()?, owner, args).await
689 } else {
690 self.children.handle_existing(owner, args)
692 }
693 }
694
695 async fn get_cli_flag(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
696 let name = arg_str(&args, "name")?;
697 Ok(serde_json::json!({
698 "value": self.cli_flags.get(&name).cloned().unwrap_or(serde_json::Value::Null)
699 }))
700 }
701
702 async fn send_message(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
703 let options = send_message_options(&args)?;
704 let message = send_message_input(&args)?;
705 let lane = self.harness()?.lane("main");
706 let result = lane
707 .send_message(message, options)
708 .await
709 .map_err(|e| e.to_string())?;
710 Ok(send_message_result(result))
711 }
712
713 async fn send_user_message(
714 &self,
715 args: serde_json::Value,
716 ) -> Result<serde_json::Value, String> {
717 let text = arg_str(&args, "text")?;
718 let lane = self.harness()?.lane("main");
719 let mut options = send_message_options(&args)?;
735 if options.trigger_turn.is_none() {
736 options.trigger_turn = Some(true);
737 }
738 let message = rpi_agent::AgentMessage::User(rpi_ai::types::UserMessage::new(
739 text,
740 std::time::SystemTime::now()
741 .duration_since(std::time::UNIX_EPOCH)
742 .unwrap_or_default()
743 .as_millis()
744 .min(i64::MAX as u128) as i64,
745 ));
746 let result = lane
747 .send_message(message, options)
748 .await
749 .map_err(|e| e.to_string())?;
750 Ok(send_message_result(result))
751 }
752
753 async fn append_entry(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
754 if let Some(expected) = args.get("expected") {
760 let guard = serde_json::from_value(expected.clone())
761 .map_err(|error| format!("invalid AppendEntry expected guard: {error}"))?;
762 let entry = if let Some(custom_type) = arg_str_opt(&args, "customType") {
763 if args.get("message").is_some() || args.get("outboxKey").is_some() {
764 return Err("scoped custom entry cannot include message/outboxKey".into());
765 }
766 crate::resource_selection::validate_public_entry(&custom_type, args.get("data"))?;
767 rpi_harness::agent_harness::ScopedEntry::Custom {
768 custom_type,
769 data: args.get("data").cloned(),
770 }
771 } else {
772 let message =
773 serde_json::from_value(args.get("message").cloned().ok_or("missing message")?)
774 .map_err(|error| format!("invalid message: {error}"))?;
775 let key = arg_str(&args, "outboxKey")?;
776 rpi_harness::agent_harness::ScopedEntry::Notice { message, key }
777 };
778 let id = self
779 .harness()?
780 .append_scoped_entry(&guard, entry)
781 .await
782 .map_err(|error| error.to_string())?;
783 return Ok(serde_json::json!({"entryId":id, "scoped":true}));
784 }
785 if args.get("outboxKey").is_some() {
786 return Err("outboxKey requires an expected guard".into());
787 }
788 if let Some(custom_type) = arg_str_opt(&args, "customType") {
789 crate::resource_selection::validate_public_entry(&custom_type, args.get("data"))?;
790 let data = args.get("data").cloned();
791 let id = self
792 .harness()?
793 .append_custom_entry(&custom_type, data)
794 .await
795 .map_err(|e| e.to_string())?;
796 return Ok(serde_json::json!({ "entryId": id }));
797 }
798 let msg = args
799 .get("message")
800 .ok_or_else(|| "missing `message` or `customType` field".to_string())?;
801 let message: rpi_agent::AgentMessage =
802 serde_json::from_value(msg.clone()).map_err(|e| format!("invalid message: {e}"))?;
803 let id = self
804 .harness()?
805 .append_message(message)
806 .await
807 .map_err(|e| e.to_string())?;
808 Ok(serde_json::json!({ "entryId": id }))
809 }
810
811 async fn get_session_branch(
812 &self,
813 args: serde_json::Value,
814 ) -> Result<serde_json::Value, String> {
815 let request = crate::modes::branch_entry_query(&args)?;
816 let (entries, session_id, leaf_id) =
817 crate::modes::branch_snapshot(&self.harness()?, &request).await?;
818 let entries = entries
819 .iter()
820 .map(serde_json::to_value)
821 .collect::<Result<Vec<_>, _>>()
822 .map_err(|e| format!("serialize session branch: {e}"))?;
823 Ok(serde_json::json!({
824 "entries": entries,
825 "count": entries.len(),
826 "filtered": true,
827 "sessionId": session_id,
828 "leafId": leaf_id,
829 "scopedAppend": true,
830 }))
831 }
832
833 async fn set_session_name(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
834 let name = arg_str(&args, "name")?;
835 self.harness()?
836 .session()
837 .set_name(Some(&name))
838 .await
839 .map_err(|e| e.to_string())?;
840 Ok(serde_json::Value::Null)
841 }
842
843 async fn get_active_tools(
844 &self,
845 _args: serde_json::Value,
846 ) -> Result<serde_json::Value, String> {
847 let lane = self.harness()?.lane("main");
848 let tools = lane.get_active_tools().await.map_err(|e| e.to_string())?;
849 Ok(serde_json::json!({ "tools": tools }))
850 }
851
852 async fn set_active_tools(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
853 let tools = arg_str_array(&args, "tools")?;
854 let lane = self.harness()?.lane("main");
855 lane.set_active_tools(tools)
856 .await
857 .map_err(|e| e.to_string())?;
858 Ok(serde_json::Value::Null)
859 }
860
861 async fn set_model(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
862 let id = arg_str(&args, "model")?;
863 let model = self
864 .resolve_model(&id)
865 .ok_or_else(|| format!("model `{id}` not in catalog"))?;
866 let lane = self.harness()?.lane("main");
867 lane.set_model(model.clone())
868 .await
869 .map_err(|e| e.to_string())?;
870 Ok(serde_json::json!({ "model": model.id }))
871 }
872
873 async fn get_thinking_level(
874 &self,
875 _args: serde_json::Value,
876 ) -> Result<serde_json::Value, String> {
877 let lane = self.harness()?.lane("main");
878 let level = lane.get_thinking_level().await.map_err(|e| e.to_string())?;
879 Ok(serde_json::json!({ "level": level }))
880 }
881
882 async fn set_thinking_level(
883 &self,
884 args: serde_json::Value,
885 ) -> Result<serde_json::Value, String> {
886 let level_val = args
887 .get("level")
888 .ok_or_else(|| "missing `level` field".to_string())?;
889 let level: ThinkingLevel = if let Some(s) = level_val.as_str() {
890 serde_json::from_value(serde_json::Value::String(s.to_string()))
891 .map_err(|e| format!("invalid thinking level `{s}`: {e}"))?
892 } else {
893 serde_json::from_value(level_val.clone())
894 .map_err(|e| format!("invalid thinking level: {e}"))?
895 };
896 let lane = self.harness()?.lane("main");
897 lane.set_thinking_level(level)
898 .await
899 .map_err(|e| e.to_string())?;
900 Ok(serde_json::Value::Null)
901 }
902
903 async fn compact(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
904 let custom = arg_str_opt(&args, "customInstructions");
905 let lane = self.harness()?.lane("main");
906 let result = lane
907 .compact(custom.as_deref())
908 .await
909 .map_err(|e| e.to_string())?;
910 Ok(
911 serde_json::json!({ "runId": result.run_id, "outcome": format!("{:?}", result.outcome) }),
912 )
913 }
914
915 async fn get_system_prompt(
916 &self,
917 _args: serde_json::Value,
918 ) -> Result<serde_json::Value, String> {
919 let prompt = self
920 .harness()?
921 .get_system_prompt()
922 .await
923 .map_err(|e| e.to_string())?;
924 Ok(serde_json::json!({ "prompt": prompt }))
925 }
926
927 async fn new_session(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
928 self.allow_session_routing()?;
929 let harness = self.harness()?;
930 let cwd_str = self.cwd.to_string_lossy().to_string();
931 let dir = &self.session_dir;
932 std::fs::create_dir_all(dir)
933 .map_err(|e| format!("create session dir {}: {e}", dir.display()))?;
934 let session = crate::session::create_jsonl_session(dir, &cwd_str)
935 .await
936 .map_err(|e| format!("create session: {e}"))?;
937 let id = session.storage().metadata().id.clone();
938 crate::session::set_active_session(&harness, session, Some(&self.tool_context)).await?;
939 Ok(serde_json::json!({ "sessionId": id }))
940 }
941
942 async fn fork(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
943 self.allow_session_routing()?;
944 let harness = self.harness()?;
945 let cwd_str = self.cwd.to_string_lossy().to_string();
946 let new_session =
947 crate::session::fork_session_storage(&harness, &cwd_str, &self.session_dir)
948 .await
949 .map_err(|e| format!("fork session: {e}"))?;
950 let id = new_session.storage().metadata().id.clone();
951 crate::session::set_active_session(&harness, new_session, Some(&self.tool_context)).await?;
952 Ok(serde_json::json!({ "sessionId": id }))
953 }
954
955 async fn navigate_tree(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
956 self.allow_session_routing()?;
957 let target_id = arg_str_opt(&args, "targetId");
958 let summarize = arg_bool(&args, "summarize");
959 let custom = arg_str_opt(&args, "customInstructions");
960 let label = arg_str_opt(&args, "label");
961 let lane = self.harness()?.lane("main");
962 let result = lane
963 .navigate_tree(
964 target_id.as_deref(),
965 summarize,
966 custom.as_deref(),
967 label.as_deref(),
968 )
969 .await
970 .map_err(|e| e.to_string())?;
971 let status = match &result.outcome {
972 NavigationOutcome::Completed { .. } => "completed",
973 NavigationOutcome::Declined { .. } => "declined",
974 NavigationOutcome::Aborted { .. } => "aborted",
975 NavigationOutcome::Failed { .. } => "failed",
976 };
977 Ok(serde_json::json!({ "runId": result.run_id, "status": status }))
978 }
979
980 async fn switch_session(&self, args: serde_json::Value) -> Result<serde_json::Value, String> {
981 self.allow_session_routing()?;
982 let harness = self.harness()?;
983 let id = arg_str(&args, "id")?;
984 let cwd_str = self.cwd.to_string_lossy().to_string();
985 let new_session: Session = open_session_by_id_in_dir(&id, &cwd_str, &self.session_dir)
986 .await
987 .map(|restored| restored.session)
988 .map_err(|e| e.to_string())?;
989 let new_id = new_session.storage().metadata().id.clone();
990 crate::session::set_active_session(&harness, new_session, Some(&self.tool_context)).await?;
991 Ok(serde_json::json!({ "sessionId": new_id }))
992 }
993
994 async fn reload(&self, _args: serde_json::Value) -> Result<serde_json::Value, String> {
995 Err("reload not configured (no /reload callback on this bridge)".to_string())
999 }
1000}
1001
1002#[cfg(test)]
1003mod lifetime_tests;
1004
1005#[cfg(test)]
1006mod send_message_tests {
1007 use super::*;
1008 use rpi_harness::types::MessageDelivery;
1009
1010 #[test]
1011 fn options_accept_flat_and_nested_forms_with_pi_defaults() {
1012 assert_eq!(
1013 send_message_options(&serde_json::json!({"text":"context"})).unwrap(),
1014 SendMessageOptions::default()
1015 );
1016 let expected = SendMessageOptions {
1017 trigger_turn: Some(false),
1018 deliver_as: MessageDelivery::FollowUp,
1019 };
1020 for args in [
1021 serde_json::json!({"text":"context", "triggerTurn":false, "deliverAs":"followUp"}),
1022 serde_json::json!({"text":"context", "options":{"triggerTurn":false, "deliverAs":"followUp"}}),
1023 ] {
1024 assert_eq!(send_message_options(&args).unwrap(), expected);
1025 }
1026 }
1027
1028 #[test]
1029 fn options_reject_invalid_or_ambiguous_scheduling_instructions() {
1030 for args in [
1031 serde_json::json!({"triggerTurn":"false"}),
1032 serde_json::json!({"deliverAs":"follow-up"}),
1033 serde_json::json!({"options":{"typo":true}}),
1034 serde_json::json!({"options":[], "text":"context"}),
1035 serde_json::json!({"options":{"triggerTurn":true}, "triggerTurn":false}),
1036 ] {
1037 assert!(send_message_options(&args).is_err(), "{args}");
1038 }
1039 }
1040
1041 #[test]
1042 fn pi_custom_input_preserves_content_and_display_metadata() {
1043 let message = send_message_input(&serde_json::json!({"message":{
1044 "customType":"goal", "content":"goal state", "display":true, "details":{"step":2}
1045 }}))
1046 .unwrap();
1047 let rpi_agent::AgentMessage::Custom(custom) = &message else {
1048 panic!("custom")
1049 };
1050 assert_eq!(
1051 custom.content,
1052 vec![rpi_ai::types::Content::text("goal state")]
1053 );
1054 let data = rpi_harness::messages::custom_data(&message).unwrap();
1055 assert_eq!(data.custom_type, "goal");
1056 assert!(data.display);
1057 assert_eq!(data.details, Some(serde_json::json!({"step":2})));
1058 let stored = serde_json::to_value(&message).unwrap();
1060 assert_eq!(
1061 send_message_input(&serde_json::json!({"message":stored})).unwrap(),
1062 message
1063 );
1064 }
1065
1066 #[test]
1067 fn missing_custom_content_is_empty_and_queued_results_use_pi_delivery_names() {
1068 let message = send_message_input(
1069 &serde_json::json!({"message":{"customType":"goal", "content":null}}),
1070 )
1071 .unwrap();
1072 assert!(
1073 matches!(message, rpi_agent::AgentMessage::Custom(custom) if custom.content.is_empty())
1074 );
1075 assert_eq!(
1076 send_message_result(SendMessageResult::Queued {
1077 entry_id: "reserved".into(),
1078 queue: rpi_harness::session::types::QueueKind::NextRun,
1079 }),
1080 serde_json::json!({"status":"queued", "entryId":"reserved", "deliverAs":"nextTurn"})
1081 );
1082 }
1083}
1084
1085#[cfg(test)]
1086mod list_commands_tests {
1087 use super::*;
1088 use rpi_harness::types::{AgentHarnessOptions, AgentHarnessResources, PromptTemplate, Skill};
1089
1090 #[tokio::test]
1091 async fn review_list_commands_must_include_pi_style_skill_names() {
1092 let root = tempfile::tempdir().unwrap();
1093 let harness = Arc::new(
1094 AgentHarness::create(AgentHarnessOptions {
1095 resources: AgentHarnessResources {
1096 skills: Some(vec![Skill {
1097 name: "manual".into(),
1098 description: "Explicit only".into(),
1099 content: "body".into(),
1100 file_path: "/skills/manual/SKILL.md".into(),
1101 disable_model_invocation: Some(true),
1102 }]),
1103 ..Default::default()
1104 },
1105 ..Default::default()
1106 })
1107 .await
1108 .unwrap(),
1109 );
1110 let (host, cell) = HarnessActionHost::new_empty(
1111 vec![],
1112 root.path().to_path_buf(),
1113 Handle::current(),
1114 Default::default(),
1115 rpi_extensions::ToolCallContext::new(root.path().to_string_lossy()),
1116 );
1117 HarnessActionHost::set_harness(&cell, harness.clone());
1118 let listing = host.list_commands(serde_json::json!({})).await.unwrap();
1119 let command = listing["commands"]
1120 .as_array()
1121 .unwrap()
1122 .iter()
1123 .find(|item| item["name"] == "/skill:manual")
1124 .expect("loaded explicit-only skill must be discoverable");
1125 assert_eq!(command["source"], "skill");
1126 assert_eq!(command["description"], "Explicit only");
1127 }
1128
1129 fn skill(name: &str, description: &str) -> Skill {
1130 Skill {
1131 name: name.into(),
1132 description: description.into(),
1133 content: "body".into(),
1134 file_path: format!("/skills/{name}/SKILL.md"),
1135 disable_model_invocation: Some(false),
1136 }
1137 }
1138
1139 fn template(name: &str) -> PromptTemplate {
1140 PromptTemplate {
1141 name: name.into(),
1142 description: None,
1143 content: "body".into(),
1144 }
1145 }
1146
1147 fn host(root: &std::path::Path) -> (HarnessActionHost, Arc<OnceLock<WeakAgentHarness>>) {
1148 let (host, cell) = HarnessActionHost::new_empty(
1149 vec![],
1150 root.to_path_buf(),
1151 Handle::current(),
1152 Default::default(),
1153 rpi_extensions::ToolCallContext::new(root.to_string_lossy()),
1154 );
1155 extern "C" fn unused_command(
1156 _: rpi_plugin_sdk::StbStringRef,
1157 _: *mut rpi_plugin_sdk::StbString,
1158 _: *mut std::ffi::c_void,
1159 ) -> i32 {
1160 0
1161 }
1162 let mut registry = rpi_extensions::ExtensionRegistry::new();
1163 for name in ["deploy", "help", "q", "skill:extension"] {
1164 registry.register_command(
1165 name.into(),
1166 "Extension command".into(),
1167 unused_command,
1168 std::ptr::null_mut(),
1169 );
1170 }
1171 host.publish_registry(Arc::new(registry.snapshot()));
1172 (host, cell)
1173 }
1174
1175 #[tokio::test]
1176 async fn list_commands_merges_sources_deduplicates_and_preserves_priority() {
1177 let root = tempfile::tempdir().unwrap();
1178 let harness = Arc::new(
1179 AgentHarness::create(AgentHarnessOptions {
1180 resources: AgentHarnessResources {
1181 skills: Some(vec![
1182 skill("extension", "shadowed"),
1183 skill("prompt", "shadowed"),
1184 skill("manual", "First skill"),
1185 skill("manual", "Duplicate skill"),
1186 ]),
1187 prompt_templates: Some(vec![
1188 template("help"),
1189 template("q"),
1190 template("deploy"),
1191 template("skill:prompt"),
1192 template("summarize"),
1193 template("/summarize"),
1194 ]),
1195 ..Default::default()
1196 },
1197 ..Default::default()
1198 })
1199 .await
1200 .unwrap(),
1201 );
1202 let (host, cell) = host(root.path());
1203 HarnessActionHost::set_harness(&cell, harness.clone());
1204 let listing = host.list_commands(serde_json::json!({})).await.unwrap();
1205 let commands = listing["commands"].as_array().unwrap();
1206 let find = |name| commands.iter().find(|c| c["name"] == name).unwrap();
1207 assert_eq!(find("/help")["source"], "builtin");
1208 assert_eq!(find("/deploy")["source"], "extension");
1209 assert_eq!(find("/skill:extension")["source"], "extension");
1210 assert_eq!(find("/skill:prompt")["source"], "prompt");
1211 assert_eq!(find("/summarize")["description"], "Expand prompt template");
1212 assert_eq!(find("/skill:manual")["source"], "skill");
1213 assert_eq!(find("/skill:manual")["description"], "First skill");
1214 assert!(!names(commands).contains(&"/q"));
1215 let unique: std::collections::BTreeSet<_> = names(commands).into_iter().collect();
1216 assert_eq!(unique.len(), commands.len());
1217 assert_eq!(commands.last().unwrap()["name"], "/skill:manual");
1218 }
1219
1220 #[tokio::test]
1221 async fn list_commands_reads_reloaded_resources_and_removes_stale_skills() {
1222 let root = tempfile::tempdir().unwrap();
1223 let harness = Arc::new(AgentHarness::create(Default::default()).await.unwrap());
1224 let (host, cell) = host(root.path());
1225 HarnessActionHost::set_harness(&cell, harness.clone());
1226 for name in ["oldskill", "freshskill"] {
1227 harness
1228 .set_resources(AgentHarnessResources {
1229 skills: Some(vec![skill(name, "Reloaded skill")]),
1230 ..Default::default()
1231 })
1232 .await
1233 .unwrap();
1234 let listing = host.list_commands(serde_json::json!({})).await.unwrap();
1235 let commands = listing["commands"].as_array().unwrap();
1236 assert!(names(commands).contains(&format!("/skill:{name}").as_str()));
1237 if name == "freshskill" {
1238 assert!(!names(commands).contains(&"/skill:oldskill"));
1239 }
1240 }
1241 harness.set_resources(Default::default()).await.unwrap();
1242 let listing = host.list_commands(serde_json::json!({})).await.unwrap();
1243 assert!(!names(listing["commands"].as_array().unwrap()).contains(&"/skill:freshskill"));
1244 }
1245
1246 #[tokio::test]
1247 async fn list_commands_without_live_harness_keeps_builtin_and_extension_entries() {
1248 let root = tempfile::tempdir().unwrap();
1249 let (host, cell) = host(root.path());
1250 let unbound = host.list_commands(serde_json::json!({})).await.unwrap();
1251 let commands = unbound["commands"].as_array().unwrap();
1252 assert!(commands
1253 .iter()
1254 .any(|c| c["name"] == "/help" && c["source"] == "builtin"));
1255 assert!(commands
1256 .iter()
1257 .any(|c| c["name"] == "/deploy" && c["source"] == "extension"));
1258 let harness = Arc::new(
1259 AgentHarness::create(AgentHarnessOptions {
1260 resources: AgentHarnessResources {
1261 skills: Some(vec![skill("manual", "Manual")]),
1262 prompt_templates: Some(vec![template("summarize")]),
1263 ..Default::default()
1264 },
1265 ..Default::default()
1266 })
1267 .await
1268 .unwrap(),
1269 );
1270 HarnessActionHost::set_harness(&cell, harness.clone());
1271 harness.close().await.unwrap();
1272 assert_eq!(
1273 host.list_commands(serde_json::json!({})).await.unwrap(),
1274 unbound
1275 );
1276 drop(harness);
1277 assert_eq!(
1278 host.list_commands(serde_json::json!({})).await.unwrap(),
1279 unbound
1280 );
1281 }
1282
1283 #[test]
1289 fn reserved_names_do_not_remove_the_builtins_themselves() {
1290 let builtin = vec![
1291 ("/help".to_string(), "help".to_string()),
1292 ("/model".to_string(), "model".to_string()),
1293 ];
1294 let out = merge_command_list_reserved(
1295 builtin,
1296 vec![("deploy".into(), "deploy".into())],
1297 vec![],
1298 vec![],
1299 vec![
1302 "/help".to_string(),
1303 "/model".to_string(),
1304 "/q".to_string(),
1305 "/context".to_string(),
1306 ],
1307 );
1308 assert_eq!(names(&out), vec!["/help", "/model", "/deploy"]);
1309 assert_eq!(out[0]["source"], "builtin");
1310 assert_eq!(out[1]["source"], "builtin");
1311 assert_eq!(out[2]["source"], "extension");
1312 }
1313
1314 #[test]
1318 fn reserved_builtin_aliases_hide_colliding_extension_entries() {
1319 let out = merge_command_list_reserved(
1320 vec![("/help".into(), "help".into())],
1321 vec![
1322 ("q".into(), "quit".into()),
1323 ("context".into(), "context".into()),
1324 ("deploy".into(), "deploy".into()),
1325 ],
1326 vec![],
1327 vec![],
1328 vec!["/q".into(), "/context".into()],
1329 );
1330 assert_eq!(names(&out), vec!["/help", "/deploy"]);
1332 assert_eq!(out[1]["source"], "extension");
1333 }
1334
1335 fn names(v: &[serde_json::Value]) -> Vec<&str> {
1336 v.iter().map(|c| c["name"].as_str().unwrap()).collect()
1337 }
1338
1339 #[test]
1340 fn builtin_wins_over_a_same_named_extension_command() {
1341 let out = merge_command_list_reserved(
1342 vec![("/model".into(), "built-in".into())],
1343 vec![("/model".into(), "extension".into())],
1344 vec![],
1345 vec![],
1346 Vec::new(),
1347 );
1348 assert_eq!(names(&out), vec!["/model"]);
1349 assert_eq!(out[0]["source"], "builtin");
1350 assert_eq!(out[0]["description"], "built-in");
1351 }
1352
1353 #[test]
1354 fn extension_and_template_names_get_a_leading_slash() {
1355 let out = merge_command_list_reserved(
1356 vec![],
1357 vec![("deploy".into(), "deploy".into())],
1358 vec![("review".into(), "review".into())],
1359 vec![],
1360 Vec::new(),
1361 );
1362 assert_eq!(names(&out), vec!["/deploy", "/review"]);
1363 assert_eq!(out[0]["source"], "extension");
1364 assert_eq!(out[1]["source"], "prompt");
1365 }
1366
1367 #[test]
1371 fn prompt_templates_are_listed_with_their_own_source() {
1372 let out = merge_command_list_reserved(
1373 vec![("/help".into(), "help".into())],
1374 vec![],
1375 vec![("review".into(), "review a diff".into())],
1376 vec![],
1377 Vec::new(),
1378 );
1379 assert_eq!(names(&out), vec!["/help", "/review"]);
1380 assert_eq!(out[1]["source"], "prompt");
1381 assert_eq!(out[1]["description"], "review a diff");
1382 }
1383
1384 #[test]
1387 fn a_builtin_name_is_not_reported_as_a_template() {
1388 let out = merge_command_list_reserved(
1389 vec![("/review".into(), "built-in".into())],
1390 vec![],
1391 vec![("review".into(), "template".into())],
1392 vec![],
1393 Vec::new(),
1394 );
1395 assert_eq!(names(&out), vec!["/review"]);
1396 assert_eq!(out[0]["source"], "builtin");
1397 }
1398
1399 #[test]
1400 fn no_commands_is_an_empty_list_not_an_error() {
1401 assert!(merge_command_list_reserved(vec![], vec![], vec![], vec![], Vec::new()).is_empty());
1402 }
1403}