lash_plugin_process_controls/
lib.rs1use std::sync::Arc;
8
9use serde_json::Value;
10
11use lash_core::plugin::{
12 PluginError, PluginFactory, PluginSessionContext, PluginSpec, SessionPlugin,
13 StaticPluginFactory,
14};
15use lash_core::{ToolCall, ToolDefinition, ToolProvider, ToolResult};
16use lash_tool_support::{
17 LashlangToolBinding, StaticToolExecute, StaticToolProvider, ToolDefinitionLashlangExt,
18};
19
20pub struct SessionProcessAdminPluginFactory {
26 inner: StaticPluginFactory,
27}
28
29impl SessionProcessAdminPluginFactory {
30 pub fn new() -> Self {
31 Self::with_cancel_process(true)
32 }
33
34 pub fn without_cancel_process() -> Self {
35 Self::with_cancel_process(false)
36 }
37
38 fn with_cancel_process(include_cancel_process: bool) -> Self {
39 let provider = StaticToolProvider::new(
40 processes_tool_definitions(include_cancel_process),
41 SessionProcessAdminTools {
42 include_cancel_process,
43 },
44 );
45 let spec =
46 PluginSpec::new().with_tool_provider(Arc::new(provider) as Arc<dyn ToolProvider>);
47 Self {
48 inner: StaticPluginFactory::new("processes", spec),
49 }
50 }
51}
52
53impl Default for SessionProcessAdminPluginFactory {
54 fn default() -> Self {
55 Self::new()
56 }
57}
58
59impl PluginFactory for SessionProcessAdminPluginFactory {
60 fn id(&self) -> &'static str {
61 self.inner.id()
62 }
63
64 fn build(&self, ctx: &PluginSessionContext) -> Result<Arc<dyn SessionPlugin>, PluginError> {
65 self.inner.build(ctx)
66 }
67}
68
69struct SessionProcessAdminTools {
70 include_cancel_process: bool,
71}
72
73#[async_trait::async_trait]
74impl StaticToolExecute for SessionProcessAdminTools {
75 async fn execute(&self, call: ToolCall<'_>) -> ToolResult {
76 match call.name {
77 "list_process_handles" => execute_process_list_tool_call(call.context, call.args).await,
78 "cancel_process" if self.include_cancel_process => {
79 execute_process_cancel_tool_call(call.context, call.args).await
80 }
81 _ => ToolResult::err_fmt(format_args!("Unknown tool: {}", call.name)),
82 }
83 }
84}
85
86pub fn process_list_tool_definition() -> ToolDefinition {
87 ToolDefinition::raw(
88 "tool:list_process_handles",
89 "list_process_handles",
90 "List process runs visible to this session, including `shell.start` runs, with process id, descriptor, optional definition name, and lifecycle status. Filters are optional; the default returns running runs.",
91 serde_json::json!({
92 "type": "object",
93 "properties": {
94 "status": {
95 "type": "string",
96 "enum": ["running", "completed", "failed", "cancelled", "any"],
97 "description": "Lifecycle status to list. The default is `running`; `any` includes historical runs."
98 },
99 "definition": {
100 "type": "object",
101 "description": "A Lashlang process definition value, for example `on_button`."
102 }
103 },
104 "additionalProperties": false
105 }),
106 process_list_output_schema(),
107 )
108 .with_examples(vec![
109 "await processes.list({})?".into(),
110 r#"await processes.list({ status: "any" })?"#.into(),
111 "await processes.list({ definition: on_button })?".into(),
112 ])
113 .with_lashlang_binding(LashlangToolBinding::new(["processes"], "list"))
114}
115
116fn processes_tool_definitions(include_cancel_process: bool) -> Vec<ToolDefinition> {
117 let mut definitions = vec![process_list_tool_definition()];
118 if include_cancel_process {
119 definitions.push(process_cancel_tool_definition());
120 }
121 definitions
122}
123
124pub fn process_cancel_tool_definition() -> ToolDefinition {
125 ToolDefinition::raw(
126 "tool:cancel_process",
127 "cancel_process",
128 "Request cancellation for a durable process, including a running `shell.start` process, by `process_id`.",
129 serde_json::json!({
130 "type": "object",
131 "properties": {
132 "process_id": {
133 "type": "string",
134 "description": "Process id returned by a process handle or `processes.list(...)`."
135 }
136 },
137 "required": ["process_id"],
138 "additionalProperties": false
139 }),
140 serde_json::json!({
141 "type": "object",
142 "properties": {
143 "process_id": { "type": "string" },
144 "status": {
145 "type": "string",
146 "enum": ["running", "completed", "failed", "cancelled"]
147 }
148 },
149 "required": ["process_id", "status"],
150 "additionalProperties": false
151 }),
152 )
153 .with_examples(vec![
154 r#"await processes.cancel({ process_id: "tool:call-01JZK7G4QP9Q4J7W3Q2E1H6M9C" })?"#.into(),
155 r#"await processes.cancel({ process_id: "subagent:session-01JZK7G4QP9Q4J7W3Q2E1H6M9C" })?"#.into(),
156 ])
157 .with_lashlang_binding(LashlangToolBinding::new(["processes"], "cancel"))
158}
159
160pub async fn execute_process_list_tool_call(
161 context: &lash_core::ToolContext<'_>,
162 args: &Value,
163) -> ToolResult {
164 let filter = match lash_core::ProcessListFilter::decode(args) {
165 Ok(filter) => filter,
166 Err(err) => return ToolResult::err_fmt(err),
167 };
168 let processes = context.processes();
169 let result = processes.list_handles_filtered(&filter).await;
170 match result {
171 Ok(entries) => ToolResult::ok(serde_json::json!(entries)),
172 Err(err) => ToolResult::err_fmt(err.to_string()),
173 }
174}
175
176fn process_list_output_schema() -> Value {
177 serde_json::json!({
178 "type": "array",
179 "items": {
180 "type": "object",
181 "properties": {
182 "__handle__": {
183 "type": "string",
184 "enum": ["process"],
185 "description": "Handle marker; pass the whole record where a process handle is needed."
186 },
187 "id": {
188 "type": "string",
189 "description": "Process handle id."
190 },
191 "process_id": {
192 "type": "string",
193 "description": "Same process id, repeated for tools that ask for process_id."
194 },
195 "descriptor": {
196 "type": "object",
197 "properties": {
198 "kind": { "type": "string" },
199 "label": { "type": "string" }
200 },
201 "additionalProperties": false
202 },
203 "definition": {
204 "type": "object",
205 "properties": {
206 "name": { "type": "string" }
207 },
208 "required": ["name"],
209 "additionalProperties": false
210 },
211 "status": {
212 "type": "string",
213 "enum": ["running", "completed", "failed", "cancelled"]
214 }
215 },
216 "required": ["__handle__", "id", "process_id", "descriptor", "status"],
217 "additionalProperties": false
218 }
219 })
220}
221
222pub async fn execute_process_cancel_tool_call(
223 context: &lash_core::ToolContext<'_>,
224 args: &Value,
225) -> ToolResult {
226 let Some(id) = args
227 .get("process_id")
228 .and_then(|value| value.as_str())
229 .map(str::trim)
230 .filter(|value| !value.is_empty())
231 else {
232 return ToolResult::err_fmt("cancel_process requires `process_id`");
233 };
234 let processes = context.processes();
235 match processes.cancel(id).await {
236 Ok(summary) => ToolResult::ok(serde_json::json!(summary)),
237 Err(err) => ToolResult::err_fmt(err.to_string()),
238 }
239}
240
241#[cfg(test)]
242mod tests {
243 use super::*;
244 use std::sync::Mutex;
245
246 #[derive(Default)]
247 struct DenyCancelAbility {
248 calls: Mutex<Vec<(lash_core::ProcessCancelSource, String)>>,
249 }
250
251 impl DenyCancelAbility {
252 fn calls(&self) -> Vec<(lash_core::ProcessCancelSource, String)> {
253 self.calls.lock().expect("cancel calls").clone()
254 }
255 }
256
257 #[async_trait::async_trait]
258 impl lash_core::ProcessCancelAbility for DenyCancelAbility {
259 async fn cancel(
260 &self,
261 _processes: &dyn lash_core::ProcessService,
262 request: lash_core::ProcessCancelRequest<'_>,
263 ) -> Result<lash_core::ProcessRecord, PluginError> {
264 self.calls
265 .lock()
266 .expect("cancel calls")
267 .push((request.source, request.process_id.to_string()));
268 Err(PluginError::Session("denied by host".to_string()))
269 }
270 }
271
272 fn context_with_cancel_ability(
273 ability: Arc<dyn lash_core::ProcessCancelAbility>,
274 ) -> lash_core::ToolContext<'static> {
275 let manager = Arc::new(lash_core::testing::MockSessionManager::default());
276 lash_core::ToolContext::__for_testing_with_process_cancel_ability(
277 "session".to_string(),
278 manager.clone(),
279 manager.clone(),
280 manager,
281 Arc::new(lash_core::UnavailableProcessService),
282 ability,
283 Arc::new(lash_core::SessionAttachmentStore::in_memory()),
284 lash_core::DirectCompletionClient::from_fn(|_, _| {
285 Err(PluginError::Session(
286 "direct completions are unavailable in this test context".to_string(),
287 ))
288 }),
289 None,
290 )
291 }
292
293 #[test]
294 fn tool_definitions_expose_processes_tools() {
295 let definitions = processes_tool_definitions(true);
296 let names = definitions
297 .iter()
298 .map(|tool| tool.name().to_string())
299 .collect::<Vec<_>>();
300
301 assert_eq!(names, vec!["list_process_handles", "cancel_process"]);
302 #[cfg(not(feature = "lashlang"))]
303 assert!(
304 definitions
305 .iter()
306 .all(|tool| tool.manifest.bindings.is_empty())
307 );
308 #[cfg(feature = "lashlang")]
309 assert!(definitions.iter().all(|tool| {
310 tool.manifest
311 .bindings
312 .contains_key(lash_lashlang_runtime::LASHLANG_TOOL_BINDING_KEY)
313 }));
314 }
315
316 #[test]
317 fn cancel_process_definition_renders_contract() {
318 let definition = process_cancel_tool_definition();
319 let rendered = definition.compact_contract().render_signature();
320 assert!(rendered.contains("status: enum["), "{rendered}");
321 assert!(!rendered.contains("terminal:"), "{rendered}");
322 }
323
324 #[test]
325 fn list_process_contract_returns_handle_array() {
326 let definition = process_list_tool_definition();
327
328 assert_eq!(
329 definition.contract.output_schema.canonical["type"],
330 serde_json::json!("array")
331 );
332 let rendered = definition.compact_contract().render_signature();
333 assert!(rendered.contains("-> list[record{"), "{rendered}");
334 assert!(rendered.contains("__handle__"), "{rendered}");
335 assert!(rendered.contains("process_id"), "{rendered}");
336 assert!(rendered.contains("definition"), "{rendered}");
337 assert!(rendered.contains("status: enum["), "{rendered}");
338 assert!(rendered.contains("status?: enum["), "{rendered}");
339 assert!(rendered.contains("definition?: record"), "{rendered}");
340 assert!(!rendered.contains("history"), "{rendered}");
341 assert!(!rendered.contains("terminal:"), "{rendered}");
342 }
343
344 #[test]
345 fn plugin_registers_cancel_when_configured_and_omits_it_otherwise() {
346 let standard_session = lash_core::PluginHost::new(
347 std::iter::once(
348 Arc::new(SessionProcessAdminPluginFactory::new()) as Arc<dyn PluginFactory>
349 )
350 .chain(lash_core::testing::test_standard_protocol_factories())
351 .collect(),
352 )
353 .build_session("standard", None)
354 .expect("standard session");
355 let standard_names = standard_session
356 .resolved_tool_catalog("standard")
357 .expect("standard tool catalog")
358 .tool_names()
359 .as_ref()
360 .clone();
361
362 let rlm_session = lash_core::PluginHost::new(
363 std::iter::once(
364 Arc::new(SessionProcessAdminPluginFactory::without_cancel_process())
365 as Arc<dyn PluginFactory>,
366 )
367 .chain(lash_core::testing::test_code_protocol_factories())
368 .collect(),
369 )
370 .build_session("rlm", None)
371 .expect("rlm session");
372 let rlm_names = rlm_session
373 .resolved_tool_catalog("rlm")
374 .expect("rlm tool catalog")
375 .tool_names()
376 .as_ref()
377 .clone();
378
379 assert!(standard_names.contains(&"list_process_handles".to_string()));
380 assert!(standard_names.contains(&"cancel_process".to_string()));
381 assert!(rlm_names.contains(&"list_process_handles".to_string()));
382 assert!(!rlm_names.contains(&"cancel_process".to_string()));
383 }
384
385 #[tokio::test]
386 async fn cancel_process_tool_uses_host_cancel_ability() {
387 let ability = Arc::new(DenyCancelAbility::default());
388 let context = context_with_cancel_ability(ability.clone());
389
390 let result = execute_process_cancel_tool_call(
391 &context,
392 &serde_json::json!({ "process_id": "process-1" }),
393 )
394 .await;
395
396 assert!(!result.is_success());
397 assert_eq!(
398 result.value_for_projection(),
399 serde_json::json!("plugin session error: denied by host")
400 );
401 assert_eq!(
402 ability.calls(),
403 vec![(
404 lash_core::ProcessCancelSource::Tool,
405 "process-1".to_string()
406 )]
407 );
408 }
409}