Skip to main content

lash_core/session/
execution_context.rs

1use std::sync::Arc;
2
3use tokio::sync::mpsc::Sender;
4use tokio_util::sync::CancellationToken;
5
6use crate::tool_dispatch::ToolDispatchContext;
7use crate::{TurnActivity, TurnActivityId, TurnEvent};
8
9#[derive(Clone)]
10pub struct RuntimeExecutionContext<'run> {
11    pub(super) session_id: String,
12    pub(super) dispatch: Arc<ToolDispatchContext<'run>>,
13    process_env_store: Arc<dyn crate::ProcessExecutionEnvStore>,
14    attachment_store: Arc<crate::SessionAttachmentStore>,
15    chronological_projection: Arc<crate::ChronologicalProjection>,
16    protocol_extension: Option<crate::ProtocolTurnExtensionHandle>,
17    turn_context: crate::TurnContext,
18    execution_env_spec: crate::ProcessExecutionEnvSpec,
19    process_originator: Option<crate::ProcessOriginator>,
20    pub(super) runtime_process_id: Option<String>,
21    pub(super) process_event_context: Option<RuntimeExecutionProcessEventContext>,
22    process_env_ref: Option<crate::ProcessExecutionEnvRef>,
23    process_wake_target: Option<crate::SessionScope>,
24    pub(super) parent_invocation: Option<crate::RuntimeInvocation>,
25    turn_phase_probe: Option<Arc<dyn crate::runtime::RuntimeTurnPhaseProbe>>,
26    pub(super) turn_event_tx: Option<Sender<TurnActivity>>,
27    pub(super) cancellation_token: Option<CancellationToken>,
28    observe_turn_cancel: bool,
29    /// Per-tool trace emission handle for this execution. Present only when the
30    /// host installed a trace sink; `None` keeps every trace call a no-op.
31    tracing: Option<RuntimeExecutionTracing>,
32    /// Graph key of the enclosing code block, stamped onto the per-tool
33    /// `TurnEvent`s emitted from this context so consumers can attribute a tool
34    /// call to its code block without ordering heuristics. `None` when the
35    /// context is not executing a code block.
36    code_block_graph_key: Option<String>,
37    /// Call id of the parent `batch` tool call when this context runs the
38    /// children of a batch dispatch, stamped onto child `TurnEvent`s. `None`
39    /// for top-level tool execution.
40    batch_parent_call_id: Option<String>,
41    /// Work-driver handle for this execution's process wiring, when the
42    /// deployment provides one. Threaded through so in-run process
43    /// operations (e.g. signalling another process) that build their own
44    /// `RuntimeEffectLocalExecutor::processes(..)` call can hand it along
45    /// instead of falling back to hub-less backoff polling.
46    process_work_driver: Option<crate::ProcessWorkDriver>,
47    /// Process ids started by THIS execution context. Possession of a handle
48    /// the run itself created is sufficient capability to await/cancel it —
49    /// run-local children are not session handle grants (the ephemeral
50    /// execution scope must never appear in durable grant state).
51    started_process_ids: Arc<std::sync::Mutex<std::collections::HashSet<String>>>,
52}
53
54#[derive(Clone)]
55pub(super) struct RuntimeExecutionProcessEventContext {
56    pub process_id: String,
57    pub registry: Arc<dyn crate::ProcessRegistry>,
58    pub awaiter: crate::ProcessAwaiter,
59    pub store: Option<Arc<dyn crate::RuntimePersistence>>,
60    pub session_store_factory: Option<Arc<dyn crate::SessionStoreFactory>>,
61    pub queued_work_driver: Option<crate::QueuedWorkDriver>,
62}
63
64/// Trace-sink handle threaded into tool execution so per-tool trace events are
65/// emitted from the single shared seam, whichever protocol drives the turn.
66///
67/// `scope_context` carries the turn-scoped identity (session / turn / iteration)
68/// so [`crate::trace::assign_span_identity`] stamps `tool:<call_id>` under the
69/// right turn; `base_context` carries the host's run-level trace context.
70#[derive(Clone)]
71pub(crate) struct RuntimeExecutionTracing {
72    sink: Arc<dyn lash_trace::TraceSink>,
73    base_context: lash_trace::TraceContext,
74    scope_context: lash_trace::TraceContext,
75}
76
77impl RuntimeExecutionTracing {
78    pub(crate) fn new(
79        sink: Arc<dyn lash_trace::TraceSink>,
80        base_context: lash_trace::TraceContext,
81        scope_context: lash_trace::TraceContext,
82    ) -> Self {
83        Self {
84            sink,
85            base_context,
86            scope_context,
87        }
88    }
89
90    fn emit(&self, event: lash_trace::TraceEvent, clock: &dyn crate::Clock) {
91        crate::trace::emit_trace(
92            &Some(Arc::clone(&self.sink)),
93            &self.base_context,
94            self.scope_context.clone(),
95            event,
96            clock,
97        );
98    }
99}
100
101impl<'run> RuntimeExecutionContext<'run> {
102    pub(super) fn process_scope(
103        &self,
104        parent_invocation: Option<crate::RuntimeInvocation>,
105    ) -> crate::ProcessOpScope<'_> {
106        crate::ProcessOpScope::new(self.dispatch.effect_controller.scoped())
107            .with_parent_invocation(parent_invocation)
108            .with_agent_frame_id(Some(self.dispatch.agent_frame_id.clone()))
109    }
110
111    #[allow(
112        clippy::too_many_arguments,
113        reason = "code execution bridge carries explicit per-turn runtime dependencies"
114    )]
115    pub(crate) fn new(
116        session_id: String,
117        dispatch: Arc<ToolDispatchContext<'run>>,
118        process_env_store: Arc<dyn crate::ProcessExecutionEnvStore>,
119        attachment_store: Arc<crate::SessionAttachmentStore>,
120        chronological_projection: Arc<crate::ChronologicalProjection>,
121        protocol_extension: Option<crate::ProtocolTurnExtensionHandle>,
122        turn_context: crate::TurnContext,
123    ) -> Self {
124        Self {
125            session_id,
126            dispatch,
127            process_env_store,
128            attachment_store,
129            chronological_projection,
130            protocol_extension,
131            turn_context,
132            execution_env_spec: crate::ProcessExecutionEnvSpec::new(
133                crate::PluginOptions::default(),
134                crate::SessionPolicy::default(),
135            ),
136            process_originator: None,
137            runtime_process_id: None,
138            process_event_context: None,
139            started_process_ids: Arc::default(),
140            process_env_ref: None,
141            process_wake_target: None,
142            parent_invocation: None,
143            turn_phase_probe: None,
144            turn_event_tx: None,
145            cancellation_token: None,
146            observe_turn_cancel: true,
147            tracing: None,
148            code_block_graph_key: None,
149            batch_parent_call_id: None,
150            process_work_driver: None,
151        }
152    }
153
154    pub fn session_id(&self) -> &str {
155        &self.session_id
156    }
157
158    pub fn execution_scope_id(&self) -> String {
159        self.dispatch
160            .effect_controller
161            .scoped()
162            .scope_id()
163            .to_string()
164    }
165
166    pub fn session_scope(&self) -> crate::SessionScope {
167        if self.dispatch.agent_frame_id.is_empty() {
168            crate::SessionScope::new(self.session_id.clone())
169        } else {
170            crate::SessionScope::for_agent_frame(
171                self.session_id.clone(),
172                self.dispatch.agent_frame_id.clone(),
173            )
174        }
175    }
176
177    pub fn trigger_store(&self) -> Option<Arc<dyn crate::TriggerStore>> {
178        self.dispatch
179            .trigger_router
180            .as_ref()
181            .map(crate::TriggerRouter::store)
182    }
183
184    pub fn trigger_registration_originator(&self) -> crate::ProcessOriginator {
185        self.process_originator
186            .clone()
187            .unwrap_or_else(|| crate::ProcessOriginator::session(self.session_scope()))
188    }
189
190    pub fn trigger_registration_wake_target(&self) -> Option<crate::SessionScope> {
191        self.process_wake_target
192            .clone()
193            .or_else(|| Some(self.session_scope()))
194    }
195
196    pub fn attachment_store(&self) -> Arc<crate::SessionAttachmentStore> {
197        Arc::clone(&self.attachment_store)
198    }
199
200    pub fn process_env_store(&self) -> Arc<dyn crate::ProcessExecutionEnvStore> {
201        Arc::clone(&self.process_env_store)
202    }
203
204    pub fn chronological_projection(&self) -> Arc<crate::ChronologicalProjection> {
205        Arc::clone(&self.chronological_projection)
206    }
207
208    pub fn protocol_extension<T: 'static>(&self) -> Option<&T> {
209        self.protocol_extension
210            .as_ref()
211            .and_then(|extension| extension.as_any().downcast_ref::<T>())
212    }
213
214    pub fn turn_context(&self) -> &crate::TurnContext {
215        &self.turn_context
216    }
217
218    pub fn tool_catalog(&self) -> Arc<crate::ToolCatalog> {
219        Arc::clone(&self.dispatch.tool_catalog)
220    }
221
222    pub(crate) fn session_graph_service(&self) -> &dyn crate::plugin::SessionGraphService {
223        self.dispatch.session_graph.as_ref()
224    }
225
226    pub(super) async fn emit_turn_activity(
227        &self,
228        correlation_id: TurnActivityId,
229        event: TurnEvent,
230    ) {
231        if let Some(tx) = &self.turn_event_tx {
232            let _ = tx.send(TurnActivity::new(correlation_id, event)).await;
233        }
234    }
235
236    pub(crate) fn with_turn_event_sender(mut self, turn_event_tx: Sender<TurnActivity>) -> Self {
237        self.turn_event_tx = Some(turn_event_tx);
238        self
239    }
240
241    pub(crate) fn with_tracing(mut self, tracing: Option<RuntimeExecutionTracing>) -> Self {
242        self.tracing = tracing;
243        self
244    }
245
246    pub(crate) fn with_code_block_graph_key(mut self, graph_key: Option<String>) -> Self {
247        self.code_block_graph_key = graph_key;
248        self
249    }
250
251    pub(crate) fn with_batch_parent_call_id(mut self, parent_call_id: Option<String>) -> Self {
252        self.batch_parent_call_id = parent_call_id;
253        self
254    }
255
256    /// Graph key of the enclosing code block for tool calls run from this
257    /// context, or `None` when no code block is executing.
258    pub(super) fn code_block_graph_key(&self) -> Option<String> {
259        self.code_block_graph_key.clone()
260    }
261
262    /// Parent batch call id for tool calls run from this context, or `None`
263    /// when this context is not executing batch children.
264    pub(super) fn batch_parent_call_id(&self) -> Option<String> {
265        self.batch_parent_call_id.clone()
266    }
267
268    /// Emit a `ToolCallStarted` trace event for a tool run from this context.
269    /// No-op when the host installed no trace sink.
270    pub(super) fn emit_tool_call_started_trace(
271        &self,
272        call_id: &str,
273        name: &str,
274        args: &serde_json::Value,
275    ) {
276        if let Some(tracing) = self.tracing.as_ref() {
277            tracing.emit(
278                lash_trace::TraceEvent::ToolCallStarted {
279                    call_id: Some(call_id.to_string()),
280                    name: name.to_string(),
281                    args: args.clone(),
282                },
283                self.dispatch.clock.as_ref(),
284            );
285        }
286    }
287
288    /// Emit a `ToolCallCompleted` trace event for a tool run from this context.
289    /// No-op when the host installed no trace sink.
290    pub(super) fn emit_tool_call_completed_trace(&self, record: &crate::ToolCallRecord) {
291        if let Some(tracing) = self.tracing.as_ref() {
292            tracing.emit(
293                lash_trace::TraceEvent::ToolCallCompleted {
294                    call_id: record.call_id.clone(),
295                    name: record.tool.clone(),
296                    args: record.args.clone(),
297                    output: crate::trace::trace_tool_call_output(&record.output),
298                    duration_ms: record.duration_ms,
299                },
300                self.dispatch.clock.as_ref(),
301            );
302        }
303    }
304
305    pub(crate) fn with_parent_invocation(mut self, metadata: crate::RuntimeInvocation) -> Self {
306        self.parent_invocation = Some(metadata);
307        self
308    }
309
310    pub(crate) fn with_execution_env_spec(
311        mut self,
312        execution_env_spec: crate::ProcessExecutionEnvSpec,
313    ) -> Self {
314        self.execution_env_spec = execution_env_spec;
315        self
316    }
317
318    pub(crate) fn with_process_registration_context(
319        mut self,
320        registration: &crate::ProcessRegistration,
321    ) -> Self {
322        self.process_originator = Some(registration.provenance.originator.clone());
323        self.runtime_process_id = Some(registration.id.clone());
324        self.process_env_ref = registration.env_ref.clone();
325        self.process_wake_target = registration.wake_target.clone();
326        self
327    }
328
329    pub(crate) fn with_process_event_context(
330        mut self,
331        process_id: impl Into<String>,
332        registry: Arc<dyn crate::ProcessRegistry>,
333        awaiter: crate::ProcessAwaiter,
334        store: Option<Arc<dyn crate::RuntimePersistence>>,
335        session_store_factory: Option<Arc<dyn crate::SessionStoreFactory>>,
336        queued_work_driver: Option<crate::QueuedWorkDriver>,
337    ) -> Self {
338        self.process_event_context = Some(RuntimeExecutionProcessEventContext {
339            process_id: process_id.into(),
340            registry,
341            awaiter,
342            store,
343            session_store_factory,
344            queued_work_driver,
345        });
346        self
347    }
348
349    /// Spawn provenance for children started by this context, present only
350    /// when this context executes a process: children inherit the chain's
351    /// originator and wake target instead of the ephemeral execution scope.
352    pub(super) fn record_started_process(&self, process_id: &str) {
353        self.started_process_ids
354            .lock()
355            .expect("started process ids lock")
356            .insert(process_id.to_string());
357    }
358
359    pub(super) fn is_run_local_process(&self, process_id: &str) -> bool {
360        self.started_process_ids
361            .lock()
362            .expect("started process ids lock")
363            .contains(process_id)
364    }
365
366    pub(crate) fn process_spawn_provenance(&self) -> Option<crate::ProcessSpawnProvenance> {
367        self.process_originator
368            .clone()
369            .map(|originator| crate::ProcessSpawnProvenance {
370                originator,
371                wake_target: self.process_wake_target.clone(),
372            })
373    }
374
375    pub(super) async fn attach_captured_process_execution_env(
376        &self,
377        registration: crate::ProcessRegistration,
378    ) -> Result<crate::ProcessRegistration, crate::PluginError> {
379        if registration.env_ref.is_some() {
380            return Ok(registration);
381        }
382        match registration.input.as_ref() {
383            crate::ProcessInput::ToolCall { .. } | crate::ProcessInput::Engine { .. } => {
384                let env_ref = self.captured_process_execution_env_ref().await?;
385                Ok(registration.with_execution_env_ref(Some(env_ref)))
386            }
387            crate::ProcessInput::External { .. } | crate::ProcessInput::SessionTurn { .. } => {
388                Ok(registration)
389            }
390        }
391    }
392
393    pub async fn captured_process_execution_env_ref(
394        &self,
395    ) -> Result<crate::ProcessExecutionEnvRef, crate::PluginError> {
396        if let Some(env_ref) = self.process_env_ref.clone() {
397            return Ok(env_ref);
398        }
399        crate::persist_process_execution_env(
400            self.process_env_store.as_ref(),
401            &self.execution_env_spec,
402        )
403        .await
404    }
405
406    pub(crate) fn with_turn_phase_probe(
407        mut self,
408        probe: Option<Arc<dyn crate::runtime::RuntimeTurnPhaseProbe>>,
409    ) -> Self {
410        self.turn_phase_probe = probe;
411        self
412    }
413
414    #[doc(hidden)]
415    pub fn named_phase(&self, phase: &'static str) -> crate::runtime::RuntimeNamedPhase {
416        crate::runtime::RuntimeNamedPhase::begin(self.turn_phase_probe.clone(), phase)
417    }
418
419    pub fn parent_invocation(&self) -> Option<&crate::RuntimeInvocation> {
420        self.parent_invocation.as_ref()
421    }
422
423    pub(crate) fn with_cancellation_token(mut self, cancellation_token: CancellationToken) -> Self {
424        self.cancellation_token = Some(cancellation_token);
425        self
426    }
427
428    pub(crate) fn without_turn_cancel_observation(mut self) -> Self {
429        self.observe_turn_cancel = false;
430        self
431    }
432
433    pub(crate) fn with_process_work_driver(
434        mut self,
435        process_work_driver: Option<crate::ProcessWorkDriver>,
436    ) -> Self {
437        self.process_work_driver = process_work_driver;
438        self
439    }
440
441    pub(crate) fn tool_scheduling(&self, name: &str) -> crate::ToolScheduling {
442        crate::tool_dispatch::resolve_tool_scheduling(&self.dispatch, name)
443    }
444
445    pub fn callable_tool_manifest(&self, name: &str) -> Option<crate::ToolManifest> {
446        crate::tool_dispatch::resolve_callable_manifest(&self.dispatch, name)
447    }
448
449    pub fn callable_tool_manifest_by_id(&self, id: &crate::ToolId) -> Option<crate::ToolManifest> {
450        crate::tool_dispatch::resolve_callable_manifest_by_id(&self.dispatch, id)
451    }
452
453    pub fn tool_argument_projection_policy(
454        &self,
455        name: &str,
456    ) -> crate::ToolArgumentProjectionPolicy {
457        crate::tool_dispatch::resolve_tool_argument_projection_policy(&self.dispatch, name)
458    }
459
460    pub async fn start_child_process(
461        &self,
462        registration: crate::ProcessRegistration,
463        kind: impl Into<String>,
464        label: Option<String>,
465    ) -> crate::ToolInvocationReply {
466        let _phase = self.named_phase("process.start_child");
467        let registration = match self
468            .attach_captured_process_execution_env(registration)
469            .await
470        {
471            Ok(registration) => registration,
472            Err(err) => {
473                return crate::ToolInvocationReply::error(serde_json::json!(err.to_string()));
474            }
475        };
476        let process_id = registration.id.clone();
477        let mut options = crate::ProcessStartOptions::new()
478            .with_descriptor(crate::ProcessHandleDescriptor::new(Some(kind), label));
479        if let Some(spawn) = self.process_spawn_provenance() {
480            options = options.with_spawn_provenance(spawn);
481        }
482        match self
483            .dispatch
484            .processes
485            .start(
486                &self.session_id,
487                registration,
488                options,
489                self.process_scope(self.parent_invocation.clone()),
490            )
491            .await
492        {
493            Ok(_) => {
494                self.record_started_process(&process_id);
495                crate::ToolInvocationReply::success(Self::process_handle_json(&process_id))
496            }
497            Err(err) => crate::ToolInvocationReply::error(serde_json::json!(err.to_string())),
498        }
499    }
500
501    pub async fn sleep_process(
502        &self,
503        scope: &str,
504        sequence: u64,
505        duration_ms: u64,
506    ) -> Result<(), crate::RuntimeEffectControllerError> {
507        let cancellation = self.cancellation_token.clone().unwrap_or_default();
508        let invocation = crate::runtime::causal::process_sleep_invocation(
509            &self.session_id,
510            self.parent_invocation.as_ref(),
511            scope,
512            sequence,
513        );
514        let outcome = self
515            .dispatch
516            .effect_controller
517            .controller()
518            .execute_effect(
519                crate::RuntimeEffectEnvelope::new(
520                    invocation,
521                    crate::RuntimeEffectCommand::Sleep { duration_ms },
522                ),
523                crate::RuntimeEffectLocalExecutor::sleep_with_clock(
524                    cancellation,
525                    std::sync::Arc::clone(&self.dispatch.clock),
526                )
527                .with_turn_cancel_observation(self.observe_turn_cancel),
528            )
529            .await?;
530        match outcome {
531            crate::RuntimeEffectOutcome::Sleep => Ok(()),
532            other => Err(crate::RuntimeEffectControllerError::new(
533                "runtime_effect_wrong_outcome",
534                format!("expected sleep outcome, got {}", other.kind().as_str()),
535            )),
536        }
537    }
538
539    pub async fn await_process_signal_event(
540        &self,
541        process_id: &str,
542        signal_name: &str,
543        event_ordinal: u64,
544    ) -> Result<serde_json::Value, crate::RuntimeEffectControllerError> {
545        let cancellation = self.cancellation_token.clone().unwrap_or_default();
546        let key = self
547            .dispatch
548            .effect_controller
549            .controller()
550            .await_event_key(
551                &crate::ExecutionScope::process(process_id),
552                crate::AwaitEventWaitIdentity::process_signal(
553                    process_id,
554                    signal_name,
555                    event_ordinal,
556                ),
557            )
558            .await?;
559        let invocation = crate::runtime::causal::process_await_event_invocation(
560            &self.session_id,
561            self.parent_invocation.as_ref(),
562            process_id,
563            signal_name,
564            event_ordinal,
565        );
566        let outcome = self
567            .dispatch
568            .effect_controller
569            .controller()
570            .execute_effect(
571                crate::RuntimeEffectEnvelope::new(
572                    invocation,
573                    crate::RuntimeEffectCommand::AwaitEvent { key },
574                ),
575                crate::RuntimeEffectLocalExecutor::await_event_with_clock(
576                    cancellation,
577                    None,
578                    std::sync::Arc::clone(&self.dispatch.clock),
579                )
580                .with_turn_cancel_observation(self.observe_turn_cancel),
581            )
582            .await?;
583        match outcome.into_await_event()? {
584            crate::Resolution::Ok(value) => Ok(value),
585            crate::Resolution::Err(err) => Err(crate::RuntimeEffectControllerError::new(
586                err.code,
587                err.message,
588            )),
589            crate::Resolution::Timeout => Err(crate::RuntimeEffectControllerError::new(
590                "process_signal_wait_timeout",
591                "process signal wait timed out",
592            )),
593            crate::Resolution::Cancelled => Err(crate::RuntimeEffectControllerError::new(
594                "process_signal_wait_cancelled",
595                "process signal wait was cancelled",
596            )),
597        }
598    }
599
600    pub async fn signal_process_by_id(
601        &self,
602        registry: Arc<dyn crate::ProcessRegistry>,
603        process_id: &str,
604        signal_name: &str,
605        signal_id: String,
606        payload: serde_json::Value,
607    ) -> Result<crate::ProcessEvent, crate::RuntimeEffectControllerError> {
608        let event_type = crate::process_signal_event_type(signal_name)?;
609        let replay_key = format!("process:{process_id}:signal.{signal_name}:{signal_id}");
610        let signal_payload = payload.clone();
611        let command = crate::ProcessCommand::Signal {
612            process_id: process_id.to_string(),
613            signal_name: signal_name.to_string(),
614            signal_id,
615            request: crate::ProcessEventAppendRequest::new(event_type.clone(), payload)
616                .with_replay_key(replay_key),
617        };
618        let effect_id = command.effect_id();
619        let invocation = crate::runtime::causal::process_effect_invocation(
620            &self.session_id,
621            self.parent_invocation.clone(),
622            &effect_id,
623        );
624        let outcome = self
625            .dispatch
626            .effect_controller
627            .controller()
628            .execute_effect(
629                crate::RuntimeEffectEnvelope::new(
630                    invocation,
631                    crate::RuntimeEffectCommand::process(command),
632                ),
633                crate::RuntimeEffectLocalExecutor::processes(
634                    Arc::clone(&registry),
635                    self.process_work_driver.clone(),
636                ),
637            )
638            .await?;
639        match outcome.into_process()? {
640            crate::ProcessEffectOutcome::Signal { event } => {
641                let waiting_ordinal =
642                    registry
643                        .get_process(process_id)
644                        .await
645                        .and_then(|record| match record.wait {
646                            Some(crate::WaitState {
647                                kind:
648                                    crate::WaitKind::Signal {
649                                        name,
650                                        event_type: wait_event_type,
651                                        ordinal,
652                                        ..
653                                    },
654                                ..
655                            }) if name == signal_name && wait_event_type == event_type => {
656                                Some(ordinal)
657                            }
658                            _ => None,
659                        });
660                let ordinal = match waiting_ordinal {
661                    Some(ordinal) => ordinal,
662                    None => {
663                        registry
664                            .count_events_through(process_id, &event_type, event.sequence)
665                            .await?
666                    }
667                };
668                if ordinal > 0 {
669                    let key = self
670                        .dispatch
671                        .effect_controller
672                        .controller()
673                        .await_event_key(
674                            &crate::ExecutionScope::process(process_id),
675                            crate::AwaitEventWaitIdentity::process_signal(
676                                process_id,
677                                signal_name,
678                                ordinal,
679                            ),
680                        )
681                        .await?;
682                    let _ = self
683                        .dispatch
684                        .effect_controller
685                        .controller()
686                        .resolve_await_event(&key, crate::Resolution::Ok(signal_payload))
687                        .await?;
688                }
689                Ok(*event)
690            }
691            other => Err(crate::RuntimeEffectControllerError::new(
692                "runtime_effect_wrong_outcome",
693                format!("expected signal outcome, got {other:?}"),
694            )),
695        }
696    }
697
698    pub async fn append_process_event(
699        &self,
700        registry: Arc<dyn crate::ProcessRegistry>,
701        process_id: &str,
702        request: crate::ProcessEventAppendRequest,
703    ) -> Result<crate::ProcessEvent, crate::PluginError> {
704        let result = registry.append_event(process_id, request).await?;
705        if let Some(context) = self.process_event_context.as_ref() {
706            crate::tool_provider::process_events::enqueue_wake_delivery(
707                context.store.clone(),
708                context.session_store_factory.as_ref(),
709                result.wake_delivery,
710                Some(self.session_graph_service()),
711                context.queued_work_driver.as_ref(),
712            )
713            .await?;
714        }
715        Ok(result.event)
716    }
717}
718
719#[cfg(test)]
720mod tests {
721    use super::*;
722    use crate::tool_dispatch::ToolDispatchContext;
723    use crate::{ToolCall, ToolProvider, ToolResult};
724
725    struct NoopTools;
726
727    #[async_trait::async_trait]
728    impl ToolProvider for NoopTools {
729        fn tool_manifests(&self) -> Vec<crate::ToolManifest> {
730            Vec::new()
731        }
732
733        fn resolve_contract(&self, _name: &str) -> Option<Arc<crate::ToolContract>> {
734            None
735        }
736
737        async fn execute(&self, _call: ToolCall<'_>) -> ToolResult {
738            ToolResult::err_fmt("not used")
739        }
740    }
741
742    #[test]
743    fn tool_argument_projection_policy_resolves_from_active_catalog_and_defaults_unknown() {
744        let tool = crate::ToolDefinition::raw(
745            "tool:seedy",
746            "seedy",
747            "Seed-aware",
748            crate::ToolDefinition::default_input_schema(),
749            serde_json::json!({ "type": "string" }),
750        )
751        .with_argument_projection(
752            crate::ToolArgumentProjectionPolicy::preserve_projected_refs_in_field("seed"),
753        );
754        let plugins = crate::plugin::PluginHost::empty()
755            .build_session("session", None)
756            .expect("plugin session");
757        let (event_tx, _event_rx) = tokio::sync::mpsc::channel(1);
758        let dispatch = Arc::new(ToolDispatchContext {
759            plugins,
760            tools: Arc::new(NoopTools),
761            tool_catalog: Arc::new(crate::ToolCatalog::from_tools(
762                vec![tool.manifest()],
763                std::collections::BTreeMap::new(),
764            )),
765            sessions: Arc::new(crate::testing::MockSessionManager::default()),
766            session_lifecycle: Arc::new(crate::testing::MockSessionManager::default()),
767            session_graph: Arc::new(crate::testing::MockSessionManager::default()),
768            processes: Arc::new(crate::UnavailableProcessService),
769            process_cancel_ability: Arc::new(crate::DefaultProcessCancelAbility),
770            trigger_router: None,
771            effect_controller: crate::runtime::RuntimeEffectControllerHandle::shared(Arc::new(
772                crate::InlineRuntimeEffectController,
773            )),
774            direct_completions: crate::DirectCompletionClient::unavailable(
775                "direct completions are unavailable in this test context",
776            ),
777            parent_invocation: None,
778            execution_env_spec: crate::ProcessExecutionEnvSpec::new(
779                crate::PluginOptions::default(),
780                crate::SessionPolicy::default(),
781            ),
782            session_id: "session".to_string(),
783            agent_frame_id: String::new(),
784            event_tx,
785            checkpoint_messages: crate::tool_dispatch::CheckpointMessageBuffer::default(),
786            trigger_outcomes: crate::tool_dispatch::ToolTriggerOutcomeBuffer::default(),
787            attachment_store: Arc::new(crate::SessionAttachmentStore::in_memory()),
788            turn_context: crate::TurnContext::default(),
789            clock: std::sync::Arc::new(crate::SystemClock),
790        });
791        let ctx = RuntimeExecutionContext::new(
792            "session".to_string(),
793            dispatch,
794            Arc::new(crate::InMemoryProcessExecutionEnvStore::new()),
795            Arc::new(crate::SessionAttachmentStore::in_memory()),
796            Arc::new(crate::ChronologicalProjection::default()),
797            None,
798            crate::TurnContext::default(),
799        );
800
801        assert_eq!(
802            ctx.tool_argument_projection_policy("seedy"),
803            crate::ToolArgumentProjectionPolicy::preserve_projected_refs_in_field("seed")
804        );
805        assert_eq!(
806            ctx.tool_argument_projection_policy("missing"),
807            crate::ToolArgumentProjectionPolicy::MaterializeProjectedValues
808        );
809    }
810}