lash_core/session/
process_handles.rs

1use serde_json::json;
2
3use super::execution_context::RuntimeExecutionContext;
4use super::tool_execution::ToolInvocationReply;
5use crate::tool_dispatch::ToolPreparationOutcome;
6use crate::{
7    ProcessHandleDescriptor, ProcessInput, ProcessRegistration, ToolCallOutput, ToolCallRecord,
8};
9
10const PROCESS_HANDLE_KIND: &str = "process";
11
12impl RuntimeExecutionContext<'_> {
13    pub(super) fn process_handle_value(id: &str, tool_name: &str) -> serde_json::Value {
14        let _ = tool_name;
15        Self::process_handle_json(id)
16    }
17
18    pub fn process_handle_json(id: &str) -> serde_json::Value {
19        json!({
20            "__handle__": "process",
21            "id": id,
22        })
23    }
24
25    pub(super) fn process_status_value(status: &crate::ProcessRecord) -> serde_json::Value {
26        json!({
27            "process_id": status.id,
28            "status": status.status.label(),
29        })
30    }
31
32    pub(super) fn parse_process_handle(
33        handle: &serde_json::Value,
34    ) -> Result<(String, Option<String>), String> {
35        let kind = handle
36            .get("__handle__")
37            .and_then(|value| value.as_str())
38            .ok_or_else(|| "Invalid process handle: missing `__handle__`".to_string())?;
39        if kind != PROCESS_HANDLE_KIND {
40            return Err(format!("Invalid process handle kind: {kind}"));
41        }
42        let id = handle
43            .get("id")
44            .and_then(|value| value.as_str())
45            .filter(|value| !value.is_empty())
46            .ok_or_else(|| "Invalid process handle: missing `id`".to_string())?;
47        let tool_name = handle
48            .get("tool")
49            .and_then(|value| value.as_str())
50            .map(str::to_string);
51        Ok((id.to_string(), tool_name))
52    }
53
54    pub(super) async fn start_tool_process(
55        &self,
56        call_id: String,
57        tool_name: String,
58        args: serde_json::Value,
59    ) -> ToolInvocationReply {
60        let handle_id = call_id.clone();
61        let pending_call = crate::sansio::PendingToolCall {
62            call_id: call_id.clone(),
63            tool_name: tool_name.clone(),
64            args: args.clone(),
65            replay: None,
66        };
67        let prepared_call = match self.prepare_tool_call(pending_call).await {
68            ToolPreparationOutcome::Prepared(prepared) => prepared,
69            ToolPreparationOutcome::Completed(outcome) => {
70                let mut record = outcome.record;
71                record.call_id = Some(call_id);
72                return ToolInvocationReply::from_output(record.output.clone()).with_record(record);
73            }
74        };
75        let registration = ProcessRegistration::session_start_draft(
76            handle_id.clone(),
77            ProcessInput::ToolCall {
78                call: prepared_call.clone(),
79            },
80            // Tool-call rows are journaled and idempotent by process id, so
81            // recovery may re-execute them (ADR 0019).
82            crate::RecoveryDisposition::Rerunnable,
83        );
84        let registration = match self
85            .attach_captured_process_execution_env(registration)
86            .await
87        {
88            Ok(registration) => registration,
89            Err(err) => return ToolInvocationReply::error(json!(err.to_string())),
90        };
91        if let Err(err) =
92            self.dispatch
93                .processes
94                .start(
95                    &self.session_id,
96                    registration,
97                    crate::ProcessStartOptions::new().with_descriptor(
98                        ProcessHandleDescriptor::new(Some("tool"), Some(tool_name.clone())),
99                    ),
100                    self.process_scope(self.parent_invocation.clone()),
101                )
102                .await
103        {
104            return ToolInvocationReply::error(json!(err.to_string()));
105        }
106
107        let handle_value = Self::process_handle_value(&handle_id, &tool_name);
108        let record = ToolCallRecord {
109            call_id: Some(call_id),
110            tool: prepared_call.tool_name,
111            args: prepared_call.args,
112            output: ToolCallOutput::success(handle_value.clone()),
113            duration_ms: 0,
114        };
115        ToolInvocationReply::success(handle_value).with_record(record)
116    }
117
118    fn elapsed_ms(&self, started: std::time::Instant) -> u64 {
119        self.dispatch
120            .clock
121            .now()
122            .duration_since(started)
123            .as_millis() as u64
124    }
125
126    fn recorded_process_reply(
127        call_id: String,
128        tool: impl Into<String>,
129        args: serde_json::Value,
130        output: ToolCallOutput,
131        duration_ms: u64,
132    ) -> ToolInvocationReply {
133        let record = ToolCallRecord {
134            call_id: Some(call_id),
135            tool: tool.into(),
136            args,
137            output: output.clone(),
138            duration_ms,
139        };
140        ToolInvocationReply::from_output(output).with_record(record)
141    }
142
143    fn recorded_process_error(
144        call_id: String,
145        tool: &'static str,
146        args: serde_json::Value,
147        message: impl Into<String>,
148        duration_ms: u64,
149    ) -> ToolInvocationReply {
150        let output = ToolInvocationReply::error(json!(message.into())).output;
151        Self::recorded_process_reply(call_id, tool, args, output, duration_ms)
152    }
153
154    pub(super) async fn await_process_handle(
155        &self,
156        call_id: String,
157        handle: serde_json::Value,
158    ) -> ToolInvocationReply {
159        let started = self.dispatch.clock.now();
160        let args = json!({ "handle": handle.clone() });
161        let (handle_id, _hinted_tool_name) = match Self::parse_process_handle(&handle) {
162            Ok(parsed) => parsed,
163            Err(err) => {
164                return Self::recorded_process_error(
165                    call_id,
166                    "await_process",
167                    args,
168                    err,
169                    self.elapsed_ms(started),
170                );
171            }
172        };
173        // Possession of a handle this run created is sufficient capability;
174        // session grant visibility only gates handles that arrived from
175        // elsewhere (run-local children carry no grants by design).
176        if !self.is_run_local_process(&handle_id)
177            && let Err(err) = self
178                .dispatch
179                .processes
180                .validate_visible(
181                    &self.session_id,
182                    std::slice::from_ref(&handle_id),
183                    self.process_scope(self.parent_invocation.clone()),
184                )
185                .await
186        {
187            return Self::recorded_process_error(
188                call_id,
189                "await_process",
190                args,
191                err.to_string(),
192                self.elapsed_ms(started),
193            );
194        }
195        let output = self
196            .await_process_with_cancellation(
197                &handle_id,
198                self.parent_invocation.clone(),
199                self.cancellation_token.clone(),
200            )
201            .await;
202        let output = match output {
203            Ok(output) => output.into_tool_output(),
204            Err(err) => ToolInvocationReply::error(json!(err.to_string())).output,
205        };
206        Self::recorded_process_reply(
207            call_id,
208            "await_process",
209            args,
210            output,
211            self.elapsed_ms(started),
212        )
213    }
214
215    pub(super) async fn signal_process_handle(
216        &self,
217        call_id: String,
218        handle: serde_json::Value,
219        signal_name: String,
220        payload: serde_json::Value,
221    ) -> ToolInvocationReply {
222        let started = self.dispatch.clock.now();
223        let args = json!({
224            "handle": handle.clone(),
225            "signal_name": signal_name.clone(),
226            "payload": payload.clone()
227        });
228        let (handle_id, _hinted_tool_name) = match Self::parse_process_handle(&handle) {
229            Ok(parsed) => parsed,
230            Err(err) => {
231                return Self::recorded_process_error(
232                    call_id,
233                    "signal_process",
234                    args,
235                    err,
236                    self.elapsed_ms(started),
237                );
238            }
239        };
240        let signal_id = format!("process-{call_id}");
241        let output = match self
242            .dispatch
243            .processes
244            .signal(
245                &self.session_id,
246                &handle_id,
247                signal_name,
248                signal_id,
249                payload,
250                self.process_scope(self.parent_invocation.clone()),
251            )
252            .await
253        {
254            Ok(event) => ToolCallOutput::success(json!({
255                "process_id": event.process_id,
256                "sequence": event.sequence,
257            })),
258            Err(err) => ToolInvocationReply::error(json!(format!("signal failed: {err}"))).output,
259        };
260        Self::recorded_process_reply(
261            call_id,
262            "signal_process",
263            args,
264            output,
265            self.elapsed_ms(started),
266        )
267    }
268
269    pub(super) async fn cancel_process_handle(
270        &self,
271        call_id: String,
272        handle: serde_json::Value,
273    ) -> ToolInvocationReply {
274        let started = self.dispatch.clock.now();
275        let args = json!({ "handle": handle.clone() });
276        let (handle_id, _hinted_tool_name) = match Self::parse_process_handle(&handle) {
277            Ok(parsed) => parsed,
278            Err(err) => {
279                return Self::recorded_process_error(
280                    call_id,
281                    "cancel_process",
282                    args,
283                    err,
284                    self.elapsed_ms(started),
285                );
286            }
287        };
288        // Run-local children bypass the grant-validating cancel ability:
289        // possession of the handle this run created is the capability, and
290        // these children carry no session grants by design.
291        let result = if self.is_run_local_process(&handle_id) {
292            self.dispatch
293                .processes
294                .cancel(
295                    &self.session_id,
296                    &handle_id,
297                    self.process_scope(self.parent_invocation.clone()),
298                )
299                .await
300        } else {
301            self.dispatch
302                .process_cancel_ability
303                .cancel(
304                    self.dispatch.processes.as_ref(),
305                    crate::ProcessCancelRequest::new(
306                        &self.session_id,
307                        &handle_id,
308                        self.process_scope(self.parent_invocation.clone()),
309                        crate::ProcessCancelSource::Process,
310                    )
311                    .with_handle(handle)
312                    .with_reason("requested by process handle"),
313                )
314                .await
315        };
316        let output = match result {
317            Ok(status) => ToolCallOutput::success(Self::process_status_value(&status)),
318            Err(err) => ToolInvocationReply::error(json!(format!("cancel failed: {err}"))).output,
319        };
320        Self::recorded_process_reply(
321            call_id,
322            "cancel_process",
323            args,
324            output,
325            self.elapsed_ms(started),
326        )
327    }
328}
329
330#[cfg(test)]
331mod tests {
332    use super::*;
333    use crate::plugin::PluginHost;
334    use crate::runtime::RuntimeEffectControllerHandle;
335    use crate::tool_dispatch::ToolDispatchContext;
336    use crate::{
337        PreparedToolCall, ProcessRegistry, ToolCall, ToolDefinition, ToolPrepareCall, ToolProvider,
338        ToolResult,
339    };
340    use std::collections::BTreeMap;
341    use std::sync::Arc;
342    use std::sync::Mutex;
343    use std::sync::atomic::{AtomicUsize, Ordering};
344
345    struct PrepareRecordingTool {
346        prepares: Arc<AtomicUsize>,
347    }
348
349    #[derive(Default)]
350    struct DenyCancelAbility {
351        calls: Mutex<Vec<(crate::ProcessCancelSource, String)>>,
352    }
353
354    impl DenyCancelAbility {
355        fn calls(&self) -> Vec<(crate::ProcessCancelSource, String)> {
356            self.calls.lock().expect("cancel calls").clone()
357        }
358    }
359
360    #[async_trait::async_trait]
361    impl crate::ProcessCancelAbility for DenyCancelAbility {
362        async fn cancel(
363            &self,
364            _processes: &dyn crate::ProcessService,
365            request: crate::ProcessCancelRequest<'_>,
366        ) -> Result<crate::ProcessRecord, crate::PluginError> {
367            self.calls
368                .lock()
369                .expect("cancel calls")
370                .push((request.source, request.process_id.to_string()));
371            Err(crate::PluginError::Session("denied by host".to_string()))
372        }
373    }
374
375    fn process_tool_definition() -> ToolDefinition {
376        ToolDefinition::raw(
377            "tool:process_prepare",
378            "process_prepare",
379            "Records preparation before background registration.",
380            serde_json::json!({
381                "type": "object",
382                "properties": {
383                    "input": { "type": "string" }
384                },
385                "additionalProperties": false
386            }),
387            serde_json::json!({ "type": "object", "additionalProperties": true }),
388        )
389    }
390
391    #[async_trait::async_trait]
392    impl ToolProvider for PrepareRecordingTool {
393        fn tool_manifests(&self) -> Vec<crate::ToolManifest> {
394            vec![process_tool_definition().manifest()]
395        }
396
397        fn resolve_contract(&self, name: &str) -> Option<Arc<crate::ToolContract>> {
398            (name == "process_prepare").then(|| Arc::new(process_tool_definition().contract()))
399        }
400
401        async fn prepare_tool_call(
402            &self,
403            call: ToolPrepareCall<'_>,
404        ) -> Result<PreparedToolCall, ToolResult> {
405            self.prepares.fetch_add(1, Ordering::SeqCst);
406            Ok(PreparedToolCall::from_parts(
407                call.pending.call_id,
408                call.tool_id,
409                call.pending.tool_name,
410                call.pending.args,
411                call.pending.replay,
412                serde_json::json!({ "prepared": true }),
413            ))
414        }
415
416        async fn execute(&self, call: ToolCall<'_>) -> ToolResult {
417            ToolResult::ok(serde_json::json!({
418                "payload": call.context.prepared_payload().clone(),
419            }))
420        }
421    }
422
423    #[tokio::test]
424    async fn process_handle_start_registers_prepared_tool_call() {
425        let prepares = Arc::new(AtomicUsize::new(0));
426        let provider: Arc<dyn ToolProvider> = Arc::new(PrepareRecordingTool {
427            prepares: Arc::clone(&prepares),
428        });
429        let plugins = PluginHost::empty()
430            .build_session("root", None)
431            .expect("plugin session");
432        let tools = Arc::clone(&provider);
433        let tool_catalog = Arc::new(crate::ToolCatalog::from_tools(
434            provider.tool_manifests(),
435            BTreeMap::new(),
436        ));
437        let host = Arc::new(crate::testing::MockSessionManager::default());
438        let (event_tx, _event_rx) = tokio::sync::mpsc::channel(8);
439        let dispatch = Arc::new(ToolDispatchContext {
440            plugins,
441            tools,
442            tool_catalog,
443            sessions: host.clone(),
444            session_lifecycle: host.clone(),
445            session_graph: host.clone(),
446            processes: host.clone(),
447            process_cancel_ability: Arc::new(crate::DefaultProcessCancelAbility),
448            trigger_router: None,
449            effect_controller: RuntimeEffectControllerHandle::shared(Arc::new(
450                crate::InlineRuntimeEffectController::default(),
451            )),
452            direct_completions: crate::DirectCompletionClient::unavailable(
453                "direct completions are unavailable in this test context",
454            ),
455            parent_invocation: None,
456            execution_env_spec: crate::ProcessExecutionEnvSpec::new(
457                crate::PluginOptions::default(),
458                crate::SessionPolicy::default(),
459            ),
460            session_id: "session".to_string(),
461            agent_frame_id: String::new(),
462            event_tx,
463            checkpoint_messages: crate::tool_dispatch::CheckpointMessageBuffer::default(),
464            trigger_outcomes: crate::tool_dispatch::ToolTriggerOutcomeBuffer::default(),
465            attachment_store: Arc::new(crate::SessionAttachmentStore::in_memory()),
466            attachment_source_policy: Arc::new(crate::OpenAttachmentSourcePolicy),
467            turn_context: crate::TurnContext::default(),
468            clock: std::sync::Arc::new(crate::SystemClock),
469        });
470        let context = RuntimeExecutionContext::new(
471            "session".to_string(),
472            dispatch,
473            Arc::new(crate::InMemoryProcessExecutionEnvStore::new()),
474            Arc::new(crate::SessionAttachmentStore::in_memory()),
475            Arc::new(crate::ChronologicalProjection::default()),
476            None,
477            crate::TurnContext::default(),
478        )
479        .with_execution_env_spec(crate::ProcessExecutionEnvSpec::new(
480            crate::PluginOptions::default(),
481            crate::runtime::tests::helpers::standard_test_policy(),
482        ));
483
484        let started = context
485            .start_tool_process(
486                "async-call-1".to_string(),
487                "process_prepare".to_string(),
488                serde_json::json!({ "input": "live" }),
489            )
490            .await;
491        let crate::ToolCallOutcome::Success(handle) = started.output.outcome else {
492            panic!("expected process handle output");
493        };
494        assert_eq!(
495            handle
496                .to_json_value()
497                .get("id")
498                .and_then(|value| value.as_str()),
499            Some("async-call-1")
500        );
501        assert_eq!(prepares.load(Ordering::SeqCst), 1);
502        let record = host
503            .process_registry
504            .get_process("async-call-1")
505            .await
506            .expect("registered process");
507        let ProcessInput::ToolCall { call } = record.input.as_ref() else {
508            panic!("expected prepared tool call process input");
509        };
510        assert_eq!(call.tool_name, "process_prepare");
511        assert_eq!(call.args, serde_json::json!({ "input": "live" }));
512        assert_eq!(
513            call.prepared_payload,
514            serde_json::json!({ "prepared": true })
515        );
516
517        let awaited = context
518            .await_process_handle("await-async-call-1".to_string(), handle.to_json_value())
519            .await;
520
521        assert!(awaited.output.is_success());
522        let record = awaited.record.expect("await record");
523        assert_eq!(record.call_id.as_deref(), Some("await-async-call-1"));
524        assert_eq!(record.tool, "await_process");
525    }
526
527    #[tokio::test]
528    async fn process_handle_signal_appends_event_from_foreground() {
529        let provider: Arc<dyn ToolProvider> = Arc::new(PrepareRecordingTool {
530            prepares: Arc::new(AtomicUsize::new(0)),
531        });
532        let plugins = PluginHost::empty()
533            .build_session("root", None)
534            .expect("plugin session");
535        let tool_catalog = Arc::new(crate::ToolCatalog::from_tools(
536            provider.tool_manifests(),
537            BTreeMap::new(),
538        ));
539        let host = Arc::new(crate::testing::MockSessionManager::default());
540        host.process_registry
541            .register_process(
542                ProcessRegistration::new(
543                    "target-process",
544                    ProcessInput::External {
545                        metadata: serde_json::Value::Null,
546                    },
547                    crate::RecoveryDisposition::ExternallyOwned,
548                    crate::ProcessProvenance::host(),
549                )
550                .with_extra_event_types([crate::ProcessEventType {
551                    name: "signal.ready".to_string(),
552                    payload_schema: crate::LashSchema::any(),
553                    semantics: crate::ProcessEventSemanticsSpec::default(),
554                }]),
555            )
556            .await
557            .expect("register target process");
558        let (event_tx, _event_rx) = tokio::sync::mpsc::channel(8);
559        let dispatch = Arc::new(ToolDispatchContext {
560            plugins,
561            tools: provider,
562            tool_catalog,
563            sessions: host.clone(),
564            session_lifecycle: host.clone(),
565            session_graph: host.clone(),
566            processes: host.clone(),
567            process_cancel_ability: Arc::new(crate::DefaultProcessCancelAbility),
568            trigger_router: None,
569            effect_controller: RuntimeEffectControllerHandle::shared(Arc::new(
570                crate::InlineRuntimeEffectController::default(),
571            )),
572            direct_completions: crate::DirectCompletionClient::unavailable(
573                "direct completions are unavailable in this test context",
574            ),
575            parent_invocation: None,
576            execution_env_spec: crate::ProcessExecutionEnvSpec::new(
577                crate::PluginOptions::default(),
578                crate::SessionPolicy::default(),
579            ),
580            session_id: "session".to_string(),
581            agent_frame_id: String::new(),
582            event_tx,
583            checkpoint_messages: crate::tool_dispatch::CheckpointMessageBuffer::default(),
584            trigger_outcomes: crate::tool_dispatch::ToolTriggerOutcomeBuffer::default(),
585            attachment_store: Arc::new(crate::SessionAttachmentStore::in_memory()),
586            attachment_source_policy: Arc::new(crate::OpenAttachmentSourcePolicy),
587            turn_context: crate::TurnContext::default(),
588            clock: std::sync::Arc::new(crate::SystemClock),
589        });
590        let context = RuntimeExecutionContext::new(
591            "session".to_string(),
592            dispatch,
593            Arc::new(crate::InMemoryProcessExecutionEnvStore::new()),
594            Arc::new(crate::SessionAttachmentStore::in_memory()),
595            Arc::new(crate::ChronologicalProjection::default()),
596            None,
597            crate::TurnContext::default(),
598        );
599
600        let handle = json!({ "__handle__": "process", "id": "target-process" });
601        let signalled = context
602            .signal_process_handle(
603                "signal-1".to_string(),
604                handle,
605                "ready".to_string(),
606                json!({ "kind": "ping" }),
607            )
608            .await;
609
610        assert!(
611            signalled.output.is_success(),
612            "{:?}",
613            signalled.output.value_for_projection()
614        );
615        let record = signalled.record.expect("signal record");
616        assert_eq!(record.call_id.as_deref(), Some("signal-1"));
617        assert_eq!(record.tool, "signal_process");
618        let events = host
619            .process_registry
620            .events_after("target-process", 0)
621            .await
622            .expect("list events");
623        assert!(
624            events.iter().any(|event| event.event_type == "signal.ready"
625                && event.payload.get("kind") == Some(&json!("ping"))),
626            "expected appended signal.ready event, got {events:?}"
627        );
628    }
629
630    #[tokio::test]
631    async fn process_handle_await_and_cancel_require_session_grant() {
632        let provider: Arc<dyn ToolProvider> = Arc::new(PrepareRecordingTool {
633            prepares: Arc::new(AtomicUsize::new(0)),
634        });
635        let plugins = PluginHost::empty()
636            .build_session("root", None)
637            .expect("plugin session");
638        let tool_catalog = Arc::new(crate::ToolCatalog::from_tools(
639            provider.tool_manifests(),
640            BTreeMap::new(),
641        ));
642        let host = Arc::new(crate::testing::MockSessionManager::default());
643        host.process_registry
644            .register_process(ProcessRegistration::new(
645                "hidden-process",
646                ProcessInput::External {
647                    metadata: serde_json::Value::Null,
648                },
649                crate::RecoveryDisposition::ExternallyOwned,
650                crate::ProcessProvenance::host(),
651            ))
652            .await
653            .expect("register hidden process");
654        let (event_tx, _event_rx) = tokio::sync::mpsc::channel(8);
655        let dispatch = Arc::new(ToolDispatchContext {
656            plugins,
657            tools: provider,
658            tool_catalog,
659            sessions: host.clone(),
660            session_lifecycle: host.clone(),
661            session_graph: host.clone(),
662            processes: host.clone(),
663            process_cancel_ability: Arc::new(crate::DefaultProcessCancelAbility),
664            trigger_router: None,
665            effect_controller: RuntimeEffectControllerHandle::shared(Arc::new(
666                crate::InlineRuntimeEffectController::default(),
667            )),
668            direct_completions: crate::DirectCompletionClient::unavailable(
669                "direct completions are unavailable in this test context",
670            ),
671            parent_invocation: None,
672            execution_env_spec: crate::ProcessExecutionEnvSpec::new(
673                crate::PluginOptions::default(),
674                crate::SessionPolicy::default(),
675            ),
676            session_id: "session".to_string(),
677            agent_frame_id: String::new(),
678            event_tx,
679            checkpoint_messages: crate::tool_dispatch::CheckpointMessageBuffer::default(),
680            trigger_outcomes: crate::tool_dispatch::ToolTriggerOutcomeBuffer::default(),
681            attachment_store: Arc::new(crate::SessionAttachmentStore::in_memory()),
682            attachment_source_policy: Arc::new(crate::OpenAttachmentSourcePolicy),
683            turn_context: crate::TurnContext::default(),
684            clock: std::sync::Arc::new(crate::SystemClock),
685        });
686        let context = RuntimeExecutionContext::new(
687            "session".to_string(),
688            dispatch,
689            Arc::new(crate::InMemoryProcessExecutionEnvStore::new()),
690            Arc::new(crate::SessionAttachmentStore::in_memory()),
691            Arc::new(crate::ChronologicalProjection::default()),
692            None,
693            crate::TurnContext::default(),
694        );
695        let handle = json!({
696            "__handle__": "process",
697            "id": "hidden-process"
698        });
699
700        let awaited = context
701            .await_process_handle("await-hidden-process".to_string(), handle.clone())
702            .await;
703        let cancelled = context
704            .cancel_process_handle("cancel-hidden-process".to_string(), handle)
705            .await;
706
707        assert!(!awaited.output.is_success());
708        assert!(!cancelled.output.is_success());
709        assert_eq!(
710            awaited
711                .record
712                .as_ref()
713                .and_then(|record| record.call_id.as_deref()),
714            Some("await-hidden-process")
715        );
716        assert_eq!(
717            cancelled
718                .record
719                .as_ref()
720                .and_then(|record| record.call_id.as_deref()),
721            Some("cancel-hidden-process")
722        );
723    }
724
725    #[tokio::test]
726    async fn process_handle_cancel_uses_host_cancel_ability() {
727        let provider: Arc<dyn ToolProvider> = Arc::new(PrepareRecordingTool {
728            prepares: Arc::new(AtomicUsize::new(0)),
729        });
730        let plugins = PluginHost::empty()
731            .build_session("root", None)
732            .expect("plugin session");
733        let tool_catalog = Arc::new(crate::ToolCatalog::from_tools(
734            provider.tool_manifests(),
735            BTreeMap::new(),
736        ));
737        let host = Arc::new(crate::testing::MockSessionManager::default());
738        let ability = Arc::new(DenyCancelAbility::default());
739        let (event_tx, _event_rx) = tokio::sync::mpsc::channel(8);
740        let dispatch = Arc::new(ToolDispatchContext {
741            plugins,
742            tools: provider,
743            tool_catalog,
744            sessions: host.clone(),
745            session_lifecycle: host.clone(),
746            session_graph: host,
747            processes: Arc::new(crate::UnavailableProcessService),
748            process_cancel_ability: ability.clone(),
749            trigger_router: None,
750            effect_controller: RuntimeEffectControllerHandle::shared(Arc::new(
751                crate::InlineRuntimeEffectController::default(),
752            )),
753            direct_completions: crate::DirectCompletionClient::unavailable(
754                "direct completions are unavailable in this test context",
755            ),
756            parent_invocation: None,
757            execution_env_spec: crate::ProcessExecutionEnvSpec::new(
758                crate::PluginOptions::default(),
759                crate::SessionPolicy::default(),
760            ),
761            session_id: "session".to_string(),
762            agent_frame_id: String::new(),
763            event_tx,
764            checkpoint_messages: crate::tool_dispatch::CheckpointMessageBuffer::default(),
765            trigger_outcomes: crate::tool_dispatch::ToolTriggerOutcomeBuffer::default(),
766            attachment_store: Arc::new(crate::SessionAttachmentStore::in_memory()),
767            attachment_source_policy: Arc::new(crate::OpenAttachmentSourcePolicy),
768            turn_context: crate::TurnContext::default(),
769            clock: std::sync::Arc::new(crate::SystemClock),
770        });
771        let context = RuntimeExecutionContext::new(
772            "session".to_string(),
773            dispatch,
774            Arc::new(crate::InMemoryProcessExecutionEnvStore::new()),
775            Arc::new(crate::SessionAttachmentStore::in_memory()),
776            Arc::new(crate::ChronologicalProjection::default()),
777            None,
778            crate::TurnContext::default(),
779        );
780
781        let cancelled = context
782            .cancel_process_handle(
783                "cancel-process-1".to_string(),
784                json!({
785                    "__handle__": "process",
786                    "id": "process-1"
787                }),
788            )
789            .await;
790
791        assert!(!cancelled.output.is_success());
792        assert_eq!(
793            cancelled.output.value_for_projection()["message"],
794            json!("cancel failed: plugin session error: denied by host")
795        );
796        assert_eq!(
797            ability.calls(),
798            vec![(crate::ProcessCancelSource::Process, "process-1".to_string())]
799        );
800    }
801}
lash_core/session/process_handles.rs

lash_core/session/
process_handles.rs
