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