agnt5_sdk_core/
runtime_adapter.rs

1use crate::client::EngineClient;
2use crate::error::{ErrorCode, Result, SdkError};
3use crate::pb::{
4    ActivationConflictReceipt, ActivationErrorCode, ActivationErrorDetail, ActivationPayload,
5    ActivationStatus, ActivationUnknownOutcomeReceipt, ActivationWaitReceipt,
6    BeginActivationOutcome, BeginActivationRequest, BeginActivationResponse,
7    CompleteActivationRequest, FailActivationRequest, SuspendActivationRequest,
8};
9use opentelemetry::trace::TraceContextExt;
10use opentelemetry::Context;
11use prost::Message;
12use std::collections::HashMap;
13use std::sync::Arc;
14use tokio::sync::{oneshot, Mutex};
15
16/// Authority returned by an accepted EXECUTE decision.
17#[derive(Debug, Clone, PartialEq)]
18pub struct ActivationExecutionReceipt {
19    pub activation_id: String,
20    pub attempt: u32,
21    pub fence_token: Vec<u8>,
22    pub accepted_journal_offset: u64,
23}
24
25/// Canonical output returned by an accepted REPLAY decision.
26#[derive(Debug, Clone, PartialEq)]
27pub struct ActivationReplayReceipt {
28    pub activation_id: String,
29    pub attempt: u32,
30    pub result: ActivationPayload,
31    pub accepted_journal_offset: u64,
32}
33
34/// Complete typed begin surface. Only Execute permits user code to run.
35#[derive(Debug, Clone, PartialEq)]
36pub enum ActivationDecision {
37    Execute(ActivationExecutionReceipt),
38    Replay(ActivationReplayReceipt),
39    Wait {
40        activation_id: String,
41        receipt: ActivationWaitReceipt,
42    },
43    Conflict {
44        activation_id: String,
45        receipt: ActivationConflictReceipt,
46    },
47    Cancelled {
48        activation_id: String,
49        attempt: u32,
50        accepted_journal_offset: u64,
51    },
52    UnknownOutcome {
53        activation_id: String,
54        receipt: ActivationUnknownOutcomeReceipt,
55    },
56}
57
58#[derive(Debug, Clone, PartialEq)]
59pub struct ActivationCompletionReceipt {
60    pub activation_id: String,
61    pub attempt: u32,
62    pub accepted_journal_offset: u64,
63    pub replayed: bool,
64}
65
66#[derive(Debug, Clone, PartialEq)]
67pub struct ActivationFailureReceipt {
68    pub activation_id: String,
69    pub attempt: u32,
70    pub status: ActivationStatus,
71    pub accepted_journal_offset: u64,
72    pub replayed: bool,
73}
74
75#[derive(Debug, Clone, PartialEq)]
76pub struct ActivationSuspensionReceipt {
77    pub activation_id: String,
78    pub attempt: u32,
79    pub accepted_journal_offset: u64,
80    pub timer_id: String,
81    pub replayed: bool,
82}
83
84/// Language-neutral durable activation adapter over the engine client.
85#[derive(Debug, Clone)]
86pub struct ActivationAdapter {
87    client: EngineClient,
88}
89
90impl ActivationAdapter {
91    pub fn new(client: EngineClient) -> Self {
92        Self { client }
93    }
94
95    pub async fn begin(&mut self, request: BeginActivationRequest) -> Result<ActivationDecision> {
96        let response = self.client.begin_activation(request).await?;
97        activation_decision(response)
98    }
99
100    pub async fn complete(
101        &mut self,
102        request: CompleteActivationRequest,
103    ) -> Result<ActivationCompletionReceipt> {
104        let response = self.client.complete_activation(request).await?;
105        if !response.accepted {
106            return Err(activation_error(
107                ErrorCode::UnknownOutcome,
108                "CompleteActivation returned without an accepted durability receipt",
109                &response.activation_id,
110                response.attempt,
111            ));
112        }
113        Ok(ActivationCompletionReceipt {
114            activation_id: response.activation_id,
115            attempt: response.attempt,
116            accepted_journal_offset: response.accepted_journal_offset,
117            replayed: response.replayed,
118        })
119    }
120
121    pub async fn fail(
122        &mut self,
123        request: FailActivationRequest,
124    ) -> Result<ActivationFailureReceipt> {
125        let response = self.client.fail_activation(request).await?;
126        if !response.accepted {
127            return Err(activation_error(
128                ErrorCode::UnknownOutcome,
129                "FailActivation returned without an accepted durability receipt",
130                &response.activation_id,
131                response.attempt,
132            ));
133        }
134        let status = ActivationStatus::try_from(response.status).map_err(|_| {
135            activation_error(
136                ErrorCode::InvalidState,
137                "FailActivation returned an unknown activation status",
138                &response.activation_id,
139                response.attempt,
140            )
141        })?;
142        Ok(ActivationFailureReceipt {
143            activation_id: response.activation_id,
144            attempt: response.attempt,
145            status,
146            accepted_journal_offset: response.accepted_journal_offset,
147            replayed: response.replayed,
148        })
149    }
150
151    pub async fn suspend(
152        &mut self,
153        request: SuspendActivationRequest,
154    ) -> Result<ActivationSuspensionReceipt> {
155        let response = self.client.suspend_activation(request).await?;
156        if !response.accepted || response.timer_id.is_empty() {
157            return Err(activation_error(
158                ErrorCode::UnknownOutcome,
159                "SuspendActivation returned without an accepted timer receipt",
160                &response.activation_id,
161                response.attempt,
162            ));
163        }
164        Ok(ActivationSuspensionReceipt {
165            activation_id: response.activation_id,
166            attempt: response.attempt,
167            accepted_journal_offset: response.accepted_journal_offset,
168            timer_id: response.timer_id,
169            replayed: response.replayed,
170        })
171    }
172}
173
174pub fn activation_decision(response: BeginActivationResponse) -> Result<ActivationDecision> {
175    let outcome = BeginActivationOutcome::try_from(response.outcome).map_err(|_| {
176        activation_error(
177            ErrorCode::InvalidState,
178            "BeginActivation returned an unknown outcome",
179            &response.activation_id,
180            response.attempt,
181        )
182    })?;
183    match outcome {
184        BeginActivationOutcome::Execute => {
185            if response.activation_id.is_empty()
186                || response.attempt == 0
187                || response.fence_token.is_empty()
188            {
189                return Err(activation_error(
190                    ErrorCode::InvalidState,
191                    "EXECUTE receipt is missing activation authority",
192                    &response.activation_id,
193                    response.attempt,
194                ));
195            }
196            Ok(ActivationDecision::Execute(ActivationExecutionReceipt {
197                activation_id: response.activation_id,
198                attempt: response.attempt,
199                fence_token: response.fence_token,
200                accepted_journal_offset: response.accepted_journal_offset,
201            }))
202        }
203        BeginActivationOutcome::Replay => {
204            let result = response.replay_result.ok_or_else(|| {
205                activation_error(
206                    ErrorCode::InvalidState,
207                    "REPLAY receipt is missing its canonical result",
208                    &response.activation_id,
209                    response.attempt,
210                )
211            })?;
212            Ok(ActivationDecision::Replay(ActivationReplayReceipt {
213                activation_id: response.activation_id,
214                attempt: response.attempt,
215                result,
216                accepted_journal_offset: response.accepted_journal_offset,
217            }))
218        }
219        BeginActivationOutcome::Wait => Ok(ActivationDecision::Wait {
220            activation_id: response.activation_id.clone(),
221            receipt: response.wait.ok_or_else(|| {
222                activation_error(
223                    ErrorCode::InvalidState,
224                    "WAIT decision is missing its receipt",
225                    &response.activation_id,
226                    response.attempt,
227                )
228            })?,
229        }),
230        BeginActivationOutcome::Conflict => Ok(ActivationDecision::Conflict {
231            activation_id: response.activation_id.clone(),
232            receipt: response.conflict.ok_or_else(|| {
233                activation_error(
234                    ErrorCode::InvalidState,
235                    "CONFLICT decision is missing its receipt",
236                    &response.activation_id,
237                    response.attempt,
238                )
239            })?,
240        }),
241        BeginActivationOutcome::Cancelled => Ok(ActivationDecision::Cancelled {
242            activation_id: response.activation_id,
243            attempt: response.attempt,
244            accepted_journal_offset: response.accepted_journal_offset,
245        }),
246        BeginActivationOutcome::UnknownOutcome => Ok(ActivationDecision::UnknownOutcome {
247            activation_id: response.activation_id.clone(),
248            receipt: response.unknown_outcome.ok_or_else(|| {
249                activation_error(
250                    ErrorCode::InvalidState,
251                    "UNKNOWN_OUTCOME decision is missing its receipt",
252                    &response.activation_id,
253                    response.attempt,
254                )
255            })?,
256        }),
257        BeginActivationOutcome::Unspecified => Err(activation_error(
258            ErrorCode::InvalidState,
259            "BeginActivation returned an unspecified outcome",
260            &response.activation_id,
261            response.attempt,
262        )),
263    }
264}
265
266fn activation_error(
267    code: ErrorCode,
268    message: impl Into<String>,
269    activation_id: &str,
270    attempt: u32,
271) -> SdkError {
272    SdkError::Activation {
273        message: message.into(),
274        code,
275        activation_id: (!activation_id.is_empty()).then(|| activation_id.to_string()),
276        attempt: (attempt != 0).then_some(attempt),
277    }
278}
279
280#[derive(Clone, PartialEq, Message)]
281struct GoogleRpcStatus {
282    #[prost(int32, tag = "1")]
283    code: i32,
284    #[prost(string, tag = "2")]
285    message: String,
286    #[prost(message, repeated, tag = "3")]
287    details: Vec<prost_types::Any>,
288}
289
290const ACTIVATION_ERROR_DETAIL_TYPE_URL: &str = "type.googleapis.com/api.v1.ActivationErrorDetail";
291
292fn decode_activation_error_detail(bytes: &[u8]) -> Option<ActivationErrorDetail> {
293    if let Ok(envelope) = GoogleRpcStatus::decode(bytes) {
294        if let Some(detail) = envelope
295            .details
296            .into_iter()
297            .find(|detail| detail.type_url == ACTIVATION_ERROR_DETAIL_TYPE_URL)
298        {
299            if let Ok(detail) = ActivationErrorDetail::decode(detail.value.as_slice()) {
300                return Some(detail);
301            }
302        }
303    }
304    // Backward compatibility for runtimes that emitted the detail directly
305    // instead of wrapping it in google.rpc.Status.
306    ActivationErrorDetail::decode(bytes).ok()
307}
308
309pub(crate) fn activation_status_error(operation: &str, status: tonic::Status) -> SdkError {
310    if !status.details().is_empty() {
311        if let Some(detail) = decode_activation_error_detail(status.details()) {
312            let code = match ActivationErrorCode::try_from(detail.code).ok() {
313                Some(ActivationErrorCode::NondeterministicReplay) => {
314                    Some(ErrorCode::NondeterministicReplay)
315                }
316                Some(ActivationErrorCode::StaleAuthority) => Some(ErrorCode::StaleAuthority),
317                Some(ActivationErrorCode::UnknownWriteOutcome) => Some(ErrorCode::UnknownOutcome),
318                Some(ActivationErrorCode::PayloadConflict) => Some(ErrorCode::PayloadConflict),
319                Some(ActivationErrorCode::IllegalTransition) => Some(ErrorCode::IllegalTransition),
320                Some(ActivationErrorCode::StateVersionConflict) => {
321                    Some(ErrorCode::StateVersionConflict)
322                }
323                Some(ActivationErrorCode::RequiredChildUnresolved) => {
324                    Some(ErrorCode::RequiredChildUnresolved)
325                }
326                Some(ActivationErrorCode::InvalidArgument)
327                | Some(ActivationErrorCode::InlinePayloadTooLarge)
328                | Some(ActivationErrorCode::ReferenceRequired) => Some(ErrorCode::InvalidInput),
329                Some(ActivationErrorCode::Unspecified) | None => None,
330            };
331            if let Some(code) = code {
332                return activation_error(
333                    code,
334                    detail.message,
335                    &detail.activation_id,
336                    detail.attempt,
337                );
338            }
339        }
340    }
341
342    let code = match status.code() {
343        tonic::Code::Unimplemented => ErrorCode::DurabilityUnavailable,
344        tonic::Code::Cancelled => ErrorCode::ActivationCancelled,
345        tonic::Code::DeadlineExceeded | tonic::Code::Unavailable | tonic::Code::Unknown => {
346            ErrorCode::UnknownOutcome
347        }
348        _ => ErrorCode::UnknownOutcome,
349    };
350    activation_error(
351        code,
352        format!("{operation} failed without a typed activation detail: {status}"),
353        "",
354        0,
355    )
356}
357
358#[derive(Debug, Clone)]
359pub struct InvocationRequest {
360    pub run_id: String,
361    pub service_name: String,
362    pub handler_name: String,
363    pub input_data: Vec<u8>,
364    pub metadata: HashMap<String, String>,
365}
366
367#[derive(Debug, Clone)]
368pub struct InvocationResponse {
369    pub run_id: String,
370    pub output_data: Vec<u8>,
371    pub success: bool,
372    pub error_message: Option<String>,
373    pub metadata: HashMap<String, String>,
374}
375
376pub struct StreamingInvocationRequest {}
377
378pub struct StreamingInvocationResponse {}
379
380#[async_trait::async_trait]
381pub trait StateManager: Send + Sync {
382    async fn get(&self, key: String) -> Result<Vec<u8>>;
383    async fn set(&self, key: String, value: Vec<u8>) -> Result<()>;
384    async fn delete(&self, key: String) -> Result<()>;
385}
386
387pub struct RuntimeContext {
388    pub run_id: String,
389    pub service_name: String,
390    pub component_name: String,
391    pub tenant_id: String,
392    pub deployment_id: String,
393    pub metadata: HashMap<String, String>,
394    pub state_manager: Arc<dyn StateManager>,
395
396    // OpenTelemetry context for trace propagation
397    pub otel_context: Option<Context>,
398
399    // Pre-extracted correlation IDs for easy access in logging
400    pub trace_id: Option<String>,
401    pub span_id: Option<String>,
402
403    // Whether this is a streaming request (for real-time SSE journal export)
404    pub is_streaming: bool,
405}
406
407impl RuntimeContext {
408    /// Create a new RuntimeContext with OpenTelemetry trace context
409    pub fn with_trace_context(
410        run_id: String,
411        service_name: String,
412        component_name: String,
413        tenant_id: String,
414        deployment_id: String,
415        metadata: HashMap<String, String>,
416        otel_context: Context,
417        state_manager: Arc<dyn StateManager>,
418        is_streaming: bool,
419    ) -> Self {
420        // Extract trace_id and span_id for logging correlation
421        let (trace_id, span_id) = extract_trace_ids(&otel_context);
422
423        Self {
424            run_id,
425            service_name,
426            component_name,
427            tenant_id,
428            deployment_id,
429            metadata,
430            state_manager,
431            otel_context: Some(otel_context),
432            trace_id,
433            span_id,
434            is_streaming,
435        }
436    }
437
438    /// Create a basic RuntimeContext without OpenTelemetry context
439    pub fn new(
440        run_id: String,
441        service_name: String,
442        component_name: String,
443        tenant_id: String,
444        deployment_id: String,
445        metadata: HashMap<String, String>,
446        state_manager: Arc<dyn StateManager>,
447    ) -> Self {
448        Self {
449            run_id,
450            service_name,
451            component_name,
452            tenant_id,
453            deployment_id,
454            metadata,
455            state_manager,
456            otel_context: None,
457            trace_id: None,
458            span_id: None,
459            is_streaming: false,
460        }
461    }
462}
463
464/// Extract trace_id and span_id from OpenTelemetry context for logging
465fn extract_trace_ids(ctx: &Context) -> (Option<String>, Option<String>) {
466    let span = ctx.span();
467    let span_ctx = span.span_context();
468
469    if span_ctx.is_valid() {
470        (
471            Some(span_ctx.trace_id().to_string()),
472            Some(span_ctx.span_id().to_string()),
473        )
474    } else {
475        (None, None)
476    }
477}
478
479pub struct RuntimeCapabilities {
480    pub supports_websockets: bool,
481    pub supports_sse: bool,
482    pub supports_bidirectional: bool,
483    pub max_payload_size: usize,
484    pub timeout_seconds: u32,
485}
486
487#[async_trait::async_trait]
488pub trait RuntimeAdapter: Send + Sync {
489    async fn handle_request(
490        &self,
491        ctx: &RuntimeContext,
492        request: InvocationRequest,
493    ) -> Result<InvocationResponse>;
494
495    async fn handle_stream(
496        &self,
497        ctx: &RuntimeContext,
498        request: StreamingInvocationRequest,
499    ) -> Result<StreamingInvocationResponse>;
500}
501
502/// Dummy state manager for basic testing
503pub struct DummyStateManager;
504
505#[async_trait::async_trait]
506impl StateManager for DummyStateManager {
507    async fn get(&self, _key: String) -> Result<Vec<u8>> {
508        Err(crate::error::SdkError::Other(anyhow::anyhow!(
509            "State management not implemented"
510        )))
511    }
512
513    async fn set(&self, _key: String, _value: Vec<u8>) -> Result<()> {
514        Err(crate::error::SdkError::Other(anyhow::anyhow!(
515            "State management not implemented"
516        )))
517    }
518
519    async fn delete(&self, _key: String) -> Result<()> {
520        Err(crate::error::SdkError::Other(anyhow::anyhow!(
521            "State management not implemented"
522        )))
523    }
524}
525
526/// Entity state load result
527#[derive(Debug, Clone)]
528pub struct EntityStateLoadResult {
529    pub found: bool,
530    pub state_json: Vec<u8>,
531    pub version: i64,
532}
533
534/// Entity state save result
535#[derive(Debug, Clone)]
536pub struct EntityStateSaveResult {
537    pub new_version: i64,
538}
539
540/// Entity state manager that communicates with Worker Coordinator via gRPC
541/// Implements bulk load/save pattern for entity state operations
542pub struct EntityStateManager {
543    stream_sender: flume::Sender<crate::pb::ServiceMessage>,
544    pending_requests:
545        Arc<Mutex<HashMap<String, oneshot::Sender<crate::pb::RuntimeServiceResponse>>>>,
546    _tenant_id: String,
547    session_id: String,
548}
549
550#[cfg(test)]
551mod activation_tests {
552    use super::*;
553    use crate::pb::activation_payload;
554    use bytes::Bytes;
555
556    #[test]
557    fn execute_decision_requires_complete_fenced_authority() {
558        let decision = activation_decision(BeginActivationResponse {
559            outcome: BeginActivationOutcome::Execute as i32,
560            activation_id: "actv1_example".into(),
561            attempt: 1,
562            fence_token: b"fence".to_vec(),
563            accepted_journal_offset: 7,
564            ..Default::default()
565        })
566        .unwrap();
567
568        assert_eq!(
569            decision,
570            ActivationDecision::Execute(ActivationExecutionReceipt {
571                activation_id: "actv1_example".into(),
572                attempt: 1,
573                fence_token: b"fence".to_vec(),
574                accepted_journal_offset: 7,
575            })
576        );
577
578        let error = activation_decision(BeginActivationResponse {
579            outcome: BeginActivationOutcome::Execute as i32,
580            activation_id: "actv1_example".into(),
581            attempt: 1,
582            ..Default::default()
583        })
584        .unwrap_err();
585        assert_eq!(error.code(), ErrorCode::InvalidState);
586    }
587
588    #[test]
589    fn replay_decision_preserves_canonical_payload() {
590        let payload = ActivationPayload {
591            value: Some(activation_payload::Value::InlineData(b"result".to_vec())),
592        };
593        let decision = activation_decision(BeginActivationResponse {
594            outcome: BeginActivationOutcome::Replay as i32,
595            activation_id: "actv1_example".into(),
596            attempt: 2,
597            replay_result: Some(payload.clone()),
598            accepted_journal_offset: 11,
599            ..Default::default()
600        })
601        .unwrap();
602        assert_eq!(
603            decision,
604            ActivationDecision::Replay(ActivationReplayReceipt {
605                activation_id: "actv1_example".into(),
606                attempt: 2,
607                result: payload,
608                accepted_journal_offset: 11,
609            })
610        );
611    }
612
613    #[test]
614    fn typed_runtime_error_detail_maps_without_parsing_message_text() {
615        let detail = ActivationErrorDetail {
616            code: ActivationErrorCode::NondeterministicReplay as i32,
617            activation_id: "actv1_conflict".into(),
618            attempt: 3,
619            message: "definition changed".into(),
620        };
621        let status = tonic::Status::with_details(
622            tonic::Code::FailedPrecondition,
623            "text is not the contract",
624            Bytes::from(detail.encode_to_vec()),
625        );
626
627        let error = activation_status_error("BeginActivation", status);
628
629        assert_eq!(error.code(), ErrorCode::NondeterministicReplay);
630        assert!(matches!(
631            error,
632            SdkError::Activation {
633                activation_id: Some(ref id),
634                attempt: Some(3),
635                ..
636            } if id == "actv1_conflict"
637        ));
638    }
639
640    #[test]
641    fn rich_runtime_error_detail_maps_without_parsing_message_text() {
642        let detail = ActivationErrorDetail {
643            code: ActivationErrorCode::IllegalTransition as i32,
644            activation_id: "actv1_missing".into(),
645            attempt: 1,
646            message: "aggregate is not visible".into(),
647        };
648        let envelope = GoogleRpcStatus {
649            code: tonic::Code::FailedPrecondition as i32,
650            message: "text is not the contract".into(),
651            details: vec![prost_types::Any {
652                type_url: ACTIVATION_ERROR_DETAIL_TYPE_URL.into(),
653                value: detail.encode_to_vec(),
654            }],
655        };
656        let status = tonic::Status::with_details(
657            tonic::Code::FailedPrecondition,
658            "text is not the contract",
659            Bytes::from(envelope.encode_to_vec()),
660        );
661
662        let error = activation_status_error("CompleteActivation", status);
663
664        assert_eq!(error.code(), ErrorCode::IllegalTransition);
665        assert!(matches!(
666            error,
667            SdkError::Activation {
668                activation_id: Some(ref id),
669                attempt: Some(1),
670                ..
671            } if id == "actv1_missing"
672        ));
673    }
674
675    #[test]
676    fn required_child_error_detail_is_preserved() {
677        let detail = ActivationErrorDetail {
678            code: ActivationErrorCode::RequiredChildUnresolved as i32,
679            activation_id: "actv1_parent".into(),
680            attempt: 0,
681            message: "child remains active".into(),
682        };
683        let status = tonic::Status::with_details(
684            tonic::Code::FailedPrecondition,
685            "text is not the contract",
686            Bytes::from(detail.encode_to_vec()),
687        );
688
689        let error = activation_status_error("CompleteActivation", status);
690
691        assert_eq!(error.code(), ErrorCode::RequiredChildUnresolved);
692        assert!(matches!(
693            error,
694            SdkError::Activation {
695                activation_id: Some(ref id),
696                attempt: None,
697                ..
698            } if id == "actv1_parent"
699        ));
700    }
701
702    #[test]
703    fn old_runtime_maps_to_durability_unavailable() {
704        let error = activation_status_error(
705            "BeginActivation",
706            tonic::Status::unimplemented("method is unavailable"),
707        );
708        assert_eq!(error.code(), ErrorCode::DurabilityUnavailable);
709        assert_eq!(error.retry_hint(), crate::error::RetryHint::NotRetryable);
710    }
711}
712
713impl EntityStateManager {
714    /// Create a new EntityStateManager with stream access
715    pub fn new(
716        stream_sender: flume::Sender<crate::pb::ServiceMessage>,
717        tenant_id: String,
718        session_id: String,
719    ) -> Self {
720        Self {
721            stream_sender,
722            pending_requests: Arc::new(Mutex::new(HashMap::new())),
723            _tenant_id: tenant_id,
724            session_id,
725        }
726    }
727
728    /// Handle RuntimeServiceResponse from the worker stream
729    /// This should be called by the worker when it receives a RuntimeServiceResponse
730    pub async fn handle_response(&self, response: crate::pb::RuntimeServiceResponse) {
731        let request_id = response.request_id.clone();
732
733        let mut pending = self.pending_requests.lock().await;
734        if let Some(sender) = pending.remove(&request_id) {
735            let _ = sender.send(response);
736        }
737    }
738
739    /// Load entire entity state from platform
740    pub async fn load_state(
741        &self,
742        entity_type: String,
743        entity_key: String,
744    ) -> Result<EntityStateLoadResult> {
745        use crate::pb::{
746            runtime_service_request, EntityStateLoadRequest, RuntimeServiceRequest, ServiceMessage,
747        };
748
749        // Generate unique request ID
750        let request_id = uuid::Uuid::new_v4().to_string();
751
752        // Create oneshot channel for response
753        let (response_tx, response_rx) = oneshot::channel();
754
755        // Store in pending requests
756        self.pending_requests
757            .lock()
758            .await
759            .insert(request_id.clone(), response_tx);
760
761        // Create RuntimeServiceRequest
762        let request = RuntimeServiceRequest {
763            request_id: request_id.clone(),
764            session_id: self.session_id.clone(),
765            operation: Some(runtime_service_request::Operation::EntityStateLoad(
766                EntityStateLoadRequest {
767                    entity_type,
768                    entity_key,
769                    scope: String::new(),    // Default to global scope
770                    scope_id: String::new(), // Empty for global scope
771                },
772            )),
773        };
774
775        // Send via worker stream
776        let message = ServiceMessage {
777            worker_id: String::new(), // Will be filled by worker
778            metadata: HashMap::new(),
779            message_type: Some(crate::pb::service_message::MessageType::RuntimeService(
780                request,
781            )),
782        };
783
784        self.stream_sender.send_async(message).await.map_err(|e| {
785            crate::error::SdkError::Connection {
786                message: format!("Failed to send load request: {}", e),
787                code: crate::error::ErrorCode::ConnectionFailed,
788                source: None,
789            }
790        })?;
791
792        // Wait for response with timeout
793        let response = tokio::time::timeout(std::time::Duration::from_secs(10), response_rx)
794            .await
795            .map_err(|_| {
796                crate::error::SdkError::Other(anyhow::anyhow!("Entity state load timeout"))
797            })?
798            .map_err(|_| {
799                crate::error::SdkError::Other(anyhow::anyhow!("Response channel closed"))
800            })?;
801
802        // Check if request succeeded
803        if !response.success {
804            return Err(crate::error::SdkError::Other(anyhow::anyhow!(
805                "Entity state load failed: {}",
806                response.error_message
807            )));
808        }
809
810        // Extract result
811        match response.result {
812            Some(crate::pb::runtime_service_response::Result::EntityStateLoad(result)) => {
813                Ok(EntityStateLoadResult {
814                    found: result.found,
815                    state_json: result.state_json,
816                    version: result.version,
817                })
818            }
819            _ => Err(crate::error::SdkError::Other(anyhow::anyhow!(
820                "Unexpected response type for entity state load"
821            ))),
822        }
823    }
824
825    /// Save entire entity state to platform with optimistic locking
826    pub async fn save_state(
827        &self,
828        entity_type: String,
829        entity_key: String,
830        state_json: Vec<u8>,
831        expected_version: i64,
832    ) -> Result<EntityStateSaveResult> {
833        use crate::pb::{
834            runtime_service_request, EntityStateSaveRequest, RuntimeServiceRequest, ServiceMessage,
835        };
836
837        // Generate unique request ID
838        let request_id = uuid::Uuid::new_v4().to_string();
839
840        // Create oneshot channel for response
841        let (response_tx, response_rx) = oneshot::channel();
842
843        // Store in pending requests
844        self.pending_requests
845            .lock()
846            .await
847            .insert(request_id.clone(), response_tx);
848
849        // Create RuntimeServiceRequest
850        let request = RuntimeServiceRequest {
851            request_id: request_id.clone(),
852            session_id: self.session_id.clone(),
853            operation: Some(runtime_service_request::Operation::EntityStateSave(
854                EntityStateSaveRequest {
855                    entity_type,
856                    entity_key,
857                    state_json,
858                    expected_version,
859                    scope: String::new(),    // Default to global scope
860                    scope_id: String::new(), // Empty for global scope
861                },
862            )),
863        };
864
865        // Send via worker stream
866        let message = ServiceMessage {
867            worker_id: String::new(), // Will be filled by worker
868            metadata: HashMap::new(),
869            message_type: Some(crate::pb::service_message::MessageType::RuntimeService(
870                request,
871            )),
872        };
873
874        self.stream_sender.send_async(message).await.map_err(|e| {
875            crate::error::SdkError::Connection {
876                message: format!("Failed to send save request: {}", e),
877                code: crate::error::ErrorCode::ConnectionFailed,
878                source: None,
879            }
880        })?;
881
882        // Wait for response with timeout
883        let response = tokio::time::timeout(std::time::Duration::from_secs(10), response_rx)
884            .await
885            .map_err(|_| {
886                crate::error::SdkError::Other(anyhow::anyhow!("Entity state save timeout"))
887            })?
888            .map_err(|_| {
889                crate::error::SdkError::Other(anyhow::anyhow!("Response channel closed"))
890            })?;
891
892        // Check if request succeeded
893        if !response.success {
894            return Err(crate::error::SdkError::Other(anyhow::anyhow!(
895                "Entity state save failed: {}",
896                response.error_message
897            )));
898        }
899
900        // Extract result
901        match response.result {
902            Some(crate::pb::runtime_service_response::Result::EntityStateSave(result)) => {
903                Ok(EntityStateSaveResult {
904                    new_version: result.new_version,
905                })
906            }
907            _ => Err(crate::error::SdkError::Other(anyhow::anyhow!(
908                "Unexpected response type for entity state save"
909            ))),
910        }
911    }
912
913    // -------------------------------------------------------------------
914    // Session / Message operations
915    // -------------------------------------------------------------------
916
917    /// Send a message to a conversation thread
918    pub async fn send_message(
919        &self,
920        correlation_id: String,
921        message_type: String,
922        payload: Vec<u8>,
923        from_service: String,
924    ) -> Result<String> {
925        use crate::pb::{
926            runtime_service_request, MessageSendRequest, RuntimeServiceRequest, ServiceMessage,
927        };
928
929        let request_id = uuid::Uuid::new_v4().to_string();
930        let (response_tx, response_rx) = oneshot::channel();
931        self.pending_requests
932            .lock()
933            .await
934            .insert(request_id.clone(), response_tx);
935
936        let request = RuntimeServiceRequest {
937            request_id: request_id.clone(),
938            session_id: self.session_id.clone(),
939            operation: Some(runtime_service_request::Operation::MessageSend(
940                MessageSendRequest {
941                    correlation_id,
942                    message_type,
943                    payload,
944                    from_service,
945                },
946            )),
947        };
948
949        let message = ServiceMessage {
950            worker_id: String::new(),
951            metadata: HashMap::new(),
952            message_type: Some(crate::pb::service_message::MessageType::RuntimeService(
953                request,
954            )),
955        };
956
957        self.stream_sender.send_async(message).await.map_err(|e| {
958            crate::error::SdkError::Connection {
959                message: format!("Failed to send message request: {}", e),
960                code: crate::error::ErrorCode::ConnectionFailed,
961                source: None,
962            }
963        })?;
964
965        let response = tokio::time::timeout(std::time::Duration::from_secs(10), response_rx)
966            .await
967            .map_err(|_| crate::error::SdkError::Other(anyhow::anyhow!("Message send timeout")))?
968            .map_err(|_| {
969                crate::error::SdkError::Other(anyhow::anyhow!("Response channel closed"))
970            })?;
971
972        if !response.success {
973            return Err(crate::error::SdkError::Other(anyhow::anyhow!(
974                "Message send failed: {}",
975                response.error_message
976            )));
977        }
978
979        match response.result {
980            Some(crate::pb::runtime_service_response::Result::MessageSend(result)) => {
981                Ok(result.message_id)
982            }
983            _ => Err(crate::error::SdkError::Other(anyhow::anyhow!(
984                "Unexpected response type for message send"
985            ))),
986        }
987    }
988
989    /// List messages in a conversation by correlation ID
990    pub async fn list_messages(&self, correlation_id: String, limit: i32) -> Result<Vec<Vec<u8>>> {
991        use crate::pb::{
992            runtime_service_request, MessageListRequest, RuntimeServiceRequest, ServiceMessage,
993        };
994
995        let request_id = uuid::Uuid::new_v4().to_string();
996        let (response_tx, response_rx) = oneshot::channel();
997        self.pending_requests
998            .lock()
999            .await
1000            .insert(request_id.clone(), response_tx);
1001
1002        let request = RuntimeServiceRequest {
1003            request_id: request_id.clone(),
1004            session_id: self.session_id.clone(),
1005            operation: Some(runtime_service_request::Operation::MessageList(
1006                MessageListRequest {
1007                    correlation_id,
1008                    limit,
1009                    after_message_id: String::new(),
1010                },
1011            )),
1012        };
1013
1014        let message = ServiceMessage {
1015            worker_id: String::new(),
1016            metadata: HashMap::new(),
1017            message_type: Some(crate::pb::service_message::MessageType::RuntimeService(
1018                request,
1019            )),
1020        };
1021
1022        self.stream_sender.send_async(message).await.map_err(|e| {
1023            crate::error::SdkError::Connection {
1024                message: format!("Failed to send message list request: {}", e),
1025                code: crate::error::ErrorCode::ConnectionFailed,
1026                source: None,
1027            }
1028        })?;
1029
1030        let response = tokio::time::timeout(std::time::Duration::from_secs(10), response_rx)
1031            .await
1032            .map_err(|_| crate::error::SdkError::Other(anyhow::anyhow!("Message list timeout")))?
1033            .map_err(|_| {
1034                crate::error::SdkError::Other(anyhow::anyhow!("Response channel closed"))
1035            })?;
1036
1037        if !response.success {
1038            return Err(crate::error::SdkError::Other(anyhow::anyhow!(
1039                "Message list failed: {}",
1040                response.error_message
1041            )));
1042        }
1043
1044        match response.result {
1045            Some(crate::pb::runtime_service_response::Result::MessageList(result)) => {
1046                Ok(result.messages)
1047            }
1048            _ => Err(crate::error::SdkError::Other(anyhow::anyhow!(
1049                "Unexpected response type for message list"
1050            ))),
1051        }
1052    }
1053
1054    /// Create a session
1055    pub async fn create_session(
1056        &self,
1057        session_id: String,
1058        component_name: String,
1059        session_type: String,
1060    ) -> Result<String> {
1061        use crate::pb::{
1062            runtime_service_request, RuntimeServiceRequest, ServiceMessage, SessionCreateRequest,
1063        };
1064
1065        let request_id = uuid::Uuid::new_v4().to_string();
1066        let (response_tx, response_rx) = oneshot::channel();
1067        self.pending_requests
1068            .lock()
1069            .await
1070            .insert(request_id.clone(), response_tx);
1071
1072        let request = RuntimeServiceRequest {
1073            request_id: request_id.clone(),
1074            session_id: self.session_id.clone(),
1075            operation: Some(runtime_service_request::Operation::SessionCreate(
1076                SessionCreateRequest {
1077                    session_id,
1078                    component_name,
1079                    session_type,
1080                    state: vec![],
1081                    metadata: vec![],
1082                    expires_at_ns: 0,
1083                },
1084            )),
1085        };
1086
1087        let message = ServiceMessage {
1088            worker_id: String::new(),
1089            metadata: HashMap::new(),
1090            message_type: Some(crate::pb::service_message::MessageType::RuntimeService(
1091                request,
1092            )),
1093        };
1094
1095        self.stream_sender.send_async(message).await.map_err(|e| {
1096            crate::error::SdkError::Connection {
1097                message: format!("Failed to send session create request: {}", e),
1098                code: crate::error::ErrorCode::ConnectionFailed,
1099                source: None,
1100            }
1101        })?;
1102
1103        let response = tokio::time::timeout(std::time::Duration::from_secs(10), response_rx)
1104            .await
1105            .map_err(|_| crate::error::SdkError::Other(anyhow::anyhow!("Session create timeout")))?
1106            .map_err(|_| {
1107                crate::error::SdkError::Other(anyhow::anyhow!("Response channel closed"))
1108            })?;
1109
1110        if !response.success {
1111            return Err(crate::error::SdkError::Other(anyhow::anyhow!(
1112                "Session create failed: {}",
1113                response.error_message
1114            )));
1115        }
1116
1117        match response.result {
1118            Some(crate::pb::runtime_service_response::Result::SessionCreate(result)) => {
1119                Ok(result.session_id)
1120            }
1121            _ => Err(crate::error::SdkError::Other(anyhow::anyhow!(
1122                "Unexpected response type for session create"
1123            ))),
1124        }
1125    }
1126}
agnt5_sdk_core/runtime_adapter.rs

agnt5_sdk_core/
runtime_adapter.rs
