1use serde::Serialize;
30use serde::de::DeserializeOwned;
31use snafu::ResultExt;
32
33use crate::cassettes::discovery::Discovery;
34use crate::core::contract::{self, core, ops};
35use crate::core::models::params::ContractParams;
36use crate::core::models::{
37 CreateSkillRequest, ExportSessionParams, ExportSessionsParams, GenerateSkillRequest,
38 PublishSkillRequest, RawTurnListResponse, SearchSpansParams, SeedDemoRequest, SeedResult,
39 SessionDetailResponse, SessionItem, SessionListParams, SessionListResponse,
40 SessionSkillsResponse, SessionTracesParams, SessionTracesResponse, SessionUpdateRequest,
41 SkillResponse, SkillVersionResponse, SkillVersionsResponse, SkillsListParams,
42 SkillsListResponse, SpanItem, SpanSearchOutput, StatsParams, StatsResponse, TraceDetail,
43 TraceListParams, TraceListResponse, TraceParams, UpdateSkillRequest,
44};
45use crate::decode;
46use crate::error::{Result, error};
47use crate::page;
48use crate::transport::{StreamingTransport, TapesTransport, WireRequest};
49
50#[derive(Debug, Clone, Copy)]
52pub struct CoreClient<T> {
53 transport: T,
54}
55
56impl<T> CoreClient<T> {
57 #[must_use]
59 pub fn new(transport: T) -> Self {
60 Self { transport }
61 }
62
63 #[must_use]
65 pub fn transport(&self) -> &T {
66 &self.transport
67 }
68
69 #[must_use]
71 pub fn into_transport(self) -> T {
72 self.transport
73 }
74}
75
76impl<T: TapesTransport> CoreClient<T> {
77 pub async fn call<R: DeserializeOwned>(
89 &self,
90 operation_id: &str,
91 values: Vec<(&str, String)>,
92 ) -> Result<R> {
93 self.call_with_body(operation_id, values, None).await
94 }
95
96 pub async fn call_with_body<R: DeserializeOwned>(
109 &self,
110 operation_id: &str,
111 values: Vec<(&str, String)>,
112 body: Option<String>,
113 ) -> Result<R> {
114 let method = core()?.method(operation_id)?;
115 let request = contract::call_for_with_body(method, values, body)?;
116 let response = self
117 .transport
118 .send(&request)
119 .await
120 .context(error::TransportSnafu)?;
121 decode::json_typed(&response)
122 }
123
124 pub fn request_for(
134 &self,
135 operation_id: &str,
136 values: Vec<(&str, String)>,
137 ) -> Result<WireRequest<'static>> {
138 contract::call_for(core()?.method(operation_id)?, values)
139 }
140
141 async fn with_params<P: ContractParams, R: DeserializeOwned>(&self, params: &P) -> Result<R> {
143 self.call(P::OPERATION, params.values()).await
144 }
145
146 async fn with_params_at<P: ContractParams, R: DeserializeOwned>(
148 &self,
149 params: &P,
150 path: Vec<(&str, String)>,
151 ) -> Result<R> {
152 let mut values: Vec<(&str, String)> = params.values();
153 values.extend(path);
154 self.call(P::OPERATION, values).await
155 }
156
157 async fn with_body<B: Serialize, R: DeserializeOwned>(
159 &self,
160 operation_id: &str,
161 values: Vec<(&str, String)>,
162 body: &B,
163 ) -> Result<R> {
164 let rendered = serde_json::to_string(body).context(error::RenderBodySnafu)?;
165 self.call_with_body(operation_id, values, Some(rendered))
166 .await
167 }
168
169 pub async fn list_sessions(&self, params: &SessionListParams) -> Result<SessionListResponse> {
175 self.with_params(params).await
176 }
177
178 pub async fn list_all_sessions(&self, params: &SessionListParams) -> Result<Vec<SessionItem>> {
188 page::walk(|cursor| {
189 let mut params = params.clone();
190 params.cursor = cursor;
191 async move { Ok(self.list_sessions(¶ms).await?.into_page()) }
192 })
193 .await
194 }
195
196 pub async fn get_session(&self, id: &str) -> Result<SessionDetailResponse> {
202 self.call(ops::GET_SESSION, vec![("id", id.to_owned())])
203 .await
204 }
205
206 pub async fn update_session(
212 &self,
213 id: &str,
214 body: &SessionUpdateRequest,
215 ) -> Result<SessionDetailResponse> {
216 self.with_body(ops::UPDATE_SESSION, vec![("id", id.to_owned())], body)
217 .await
218 }
219
220 pub async fn delete_session(&self, id: &str) -> Result<()> {
226 self.call(ops::DELETE_SESSION, vec![("id", id.to_owned())])
227 .await
228 }
229
230 pub async fn get_session_traces(
236 &self,
237 id: &str,
238 params: &SessionTracesParams,
239 ) -> Result<SessionTracesResponse> {
240 self.with_params_at(params, vec![("id", id.to_owned())])
241 .await
242 }
243
244 pub async fn list_raw_turns(&self, id: &str) -> Result<RawTurnListResponse> {
250 self.call(ops::LIST_RAW_TURNS, vec![("id", id.to_owned())])
251 .await
252 }
253
254 pub async fn list_session_skills(&self, id: &str) -> Result<SessionSkillsResponse> {
260 self.call(ops::LIST_SESSION_SKILLS, vec![("id", id.to_owned())])
261 .await
262 }
263
264 pub async fn list_traces(&self, params: &TraceListParams) -> Result<TraceListResponse> {
270 self.with_params(params).await
271 }
272
273 pub async fn get_trace(&self, trace_id: &str, params: &TraceParams) -> Result<TraceDetail> {
279 self.with_params_at(params, vec![("trace_id", trace_id.to_owned())])
280 .await
281 }
282
283 pub async fn get_span(&self, trace_id: &str, span_id: &str) -> Result<SpanItem> {
289 self.call(
290 ops::GET_SPAN,
291 vec![
292 ("trace_id", trace_id.to_owned()),
293 ("span_id", span_id.to_owned()),
294 ],
295 )
296 .await
297 }
298
299 pub async fn search_spans(&self, params: &SearchSpansParams) -> Result<SpanSearchOutput> {
305 self.with_params(params).await
306 }
307
308 pub async fn get_stats(&self, params: &StatsParams) -> Result<StatsResponse> {
314 self.with_params(params).await
315 }
316
317 pub async fn list_skills(&self, params: &SkillsListParams) -> Result<SkillsListResponse> {
323 self.with_params(params).await
324 }
325
326 pub async fn list_all_skills(&self, params: &SkillsListParams) -> Result<Vec<SkillResponse>> {
332 page::walk(|cursor| {
333 let mut params = params.clone();
334 params.cursor = cursor;
335 async move { Ok(self.list_skills(¶ms).await?.into_page()) }
336 })
337 .await
338 }
339
340 pub async fn get_skill(&self, id: &str) -> Result<SkillResponse> {
346 self.call(ops::GET_SKILL, vec![("id", id.to_owned())]).await
347 }
348
349 pub async fn create_skill(&self, body: &CreateSkillRequest) -> Result<SkillResponse> {
355 self.with_body(ops::CREATE_SKILL, Vec::new(), body).await
356 }
357
358 pub async fn update_skill(&self, id: &str, body: &UpdateSkillRequest) -> Result<SkillResponse> {
364 self.with_body(ops::UPDATE_SKILL, vec![("id", id.to_owned())], body)
365 .await
366 }
367
368 pub async fn delete_skill(&self, id: &str) -> Result<()> {
374 self.call(ops::DELETE_SKILL, vec![("id", id.to_owned())])
375 .await
376 }
377
378 pub async fn duplicate_skill(&self, id: &str) -> Result<SkillResponse> {
384 self.call(ops::DUPLICATE_SKILL, vec![("id", id.to_owned())])
385 .await
386 }
387
388 pub async fn list_skill_versions(&self, id: &str) -> Result<SkillVersionsResponse> {
394 self.call(ops::LIST_SKILL_VERSIONS, vec![("id", id.to_owned())])
395 .await
396 }
397
398 pub async fn publish_skill(
404 &self,
405 id: &str,
406 body: &PublishSkillRequest,
407 ) -> Result<SkillVersionResponse> {
408 self.with_body(ops::PUBLISH_SKILL, vec![("id", id.to_owned())], body)
409 .await
410 }
411
412 pub async fn generate_skill(&self, body: &GenerateSkillRequest) -> Result<SkillResponse> {
418 self.with_body(ops::GENERATE_SKILL, Vec::new(), body).await
419 }
420
421 pub async fn list_cassettes(&self) -> Result<Discovery> {
432 self.call(ops::LIST_CASSETTES, Vec::new()).await
433 }
434
435 pub async fn seed_demo(&self, body: &SeedDemoRequest) -> Result<SeedResult> {
441 self.with_body(ops::SEED_DEMO, Vec::new(), body).await
442 }
443}
444
445impl<T: StreamingTransport> CoreClient<T> {
446 pub async fn stream(&self, operation_id: &str, values: Vec<(&str, String)>) -> Result<T::Body> {
459 let method = core()?.method(operation_id)?;
460 let request = contract::call_for(method, values)?;
461 self.transport.send_stream(&request).await
462 }
463
464 pub async fn export_session(&self, id: &str, params: &ExportSessionParams) -> Result<T::Body> {
475 let mut values = params.values();
476 values.push(("id", id.to_owned()));
477 self.stream(ops::EXPORT_SESSION, values).await
478 }
479
480 pub async fn export_sessions(&self, params: &ExportSessionsParams) -> Result<T::Body> {
486 self.stream(ops::EXPORT_SESSIONS, params.values()).await
487 }
488}
489
490#[cfg(test)]
491#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
492mod tests {
493 use super::*;
494 use crate::core::models::params::PayloadDetail;
495 use crate::path::{PathMode, call_url};
496 use crate::transport::{TransportError, WireResponse};
497 use serde::Deserialize;
498 use serde_json::Value;
499 use std::cell::RefCell;
500 use url::Url;
501
502 struct Recorder {
510 base: Url,
511 responses: RefCell<Vec<Value>>,
512 seen: RefCell<Vec<String>>,
513 bodies: RefCell<Vec<Option<String>>>,
514 }
515
516 impl Recorder {
517 fn new(base: &str, responses: Vec<Value>) -> Self {
518 Self {
519 base: Url::parse(base).unwrap(),
520 responses: RefCell::new(responses),
521 seen: RefCell::new(Vec::new()),
522 bodies: RefCell::new(Vec::new()),
523 }
524 }
525 }
526
527 impl TapesTransport for Recorder {
528 async fn send(
529 &self,
530 request: &WireRequest<'_>,
531 ) -> std::result::Result<WireResponse, TransportError> {
532 let url = call_url(&self.base, request, PathMode::UnderBase)
533 .map_err(|error| TransportError::new(error.to_string()))?;
534 self.seen.borrow_mut().push(url.to_string());
535 self.bodies.borrow_mut().push(request.body.clone());
536 let mut responses = self.responses.borrow_mut();
537 let body = if responses.len() > 1 {
538 responses.remove(0)
539 } else {
540 responses.first().cloned().unwrap_or(Value::Null)
541 };
542 Ok(WireResponse::new(
543 200,
544 url.to_string(),
545 Vec::new(),
546 body.to_string().into_bytes(),
547 ))
548 }
549 }
550
551 fn client(base: &str, response: Value) -> CoreClient<Recorder> {
552 CoreClient::new(Recorder::new(base, vec![response]))
553 }
554
555 #[tokio::test]
556 async fn an_operation_is_routed_through_the_contract_and_the_transport() {
557 let client = client(
558 "https://acme.example/primary/tapes/",
559 serde_json::json!({"traces": []}),
560 );
561 let _ = client
562 .get_session_traces("s-1", &SessionTracesParams::default())
563 .await
564 .unwrap();
565
566 assert_eq!(
567 client.transport().seen.borrow()[0],
568 "https://acme.example/primary/tapes/v1/sessions/s-1/traces",
569 );
570 }
571
572 #[tokio::test]
573 async fn a_typed_method_decodes_the_contracts_own_shape() {
574 let client = client(
577 "http://127.0.0.1:8081",
578 serde_json::json!({
579 "items": [{"id": "s1", "rollup": {"turn_count": 3}}],
580 "next_cursor": "abc",
581 }),
582 );
583 let listing = client
584 .list_sessions(&SessionListParams::default())
585 .await
586 .unwrap();
587
588 assert_eq!(listing.items[0].id, "s1");
589 assert_eq!(listing.items[0].rollup.turn_count, 3);
590 assert_eq!(listing.next_cursor, "abc");
591 }
592
593 #[tokio::test]
594 async fn a_typed_method_survives_a_field_it_has_never_heard_of() {
595 let client = client(
598 "http://127.0.0.1:8081",
599 serde_json::json!({"items": [{"id": "s1", "a_field_from_the_future": 7}]}),
600 );
601 let listing = client
602 .list_sessions(&SessionListParams::default())
603 .await
604 .unwrap();
605 assert_eq!(listing.items[0].id, "s1");
606 }
607
608 #[tokio::test]
609 async fn the_generic_seam_still_decodes_into_a_callers_own_type() {
610 #[derive(Debug, Deserialize)]
613 struct Listing {
614 next_cursor: String,
615 }
616
617 let client = client(
618 "http://127.0.0.1:8081",
619 serde_json::json!({"items": [], "next_cursor": "abc"}),
620 );
621 let got: Listing = client.call(ops::LIST_SESSIONS, Vec::new()).await.unwrap();
622 assert_eq!(got.next_cursor, "abc");
623
624 let raw: Value = client.call(ops::LIST_SESSIONS, Vec::new()).await.unwrap();
625 assert_eq!(raw["next_cursor"], "abc");
626 }
627
628 #[tokio::test]
629 async fn a_typed_parameter_travels_under_the_contracts_own_name() {
630 let client = client("http://127.0.0.1:8081", serde_json::json!({"traces": []}));
631 let _ = client
632 .get_session_traces(
633 "s-1",
634 &SessionTracesParams {
635 payload: Some(PayloadDetail::Preview),
636 },
637 )
638 .await
639 .unwrap();
640 assert!(
641 client.transport().seen.borrow()[0].ends_with("/traces?payload=preview"),
642 "got: {:?}",
643 client.transport().seen.borrow(),
644 );
645 }
646
647 #[tokio::test]
648 async fn a_listing_walk_follows_the_cursor_to_the_end() {
649 let client = CoreClient::new(Recorder::new(
652 "http://127.0.0.1:8081",
653 vec![
654 serde_json::json!({"items": [{"id": "s1"}], "next_cursor": "c1"}),
655 serde_json::json!({"items": [{"id": "s2"}], "next_cursor": ""}),
656 ],
657 ));
658 let sessions = client
659 .list_all_sessions(&SessionListParams::default())
660 .await
661 .unwrap();
662
663 assert_eq!(
664 sessions.iter().map(|s| s.id.as_str()).collect::<Vec<_>>(),
665 vec!["s1", "s2"],
666 );
667 assert!(
668 client.transport().seen.borrow()[1].contains("cursor=c1"),
669 "got: {:?}",
670 client.transport().seen.borrow(),
671 );
672 }
673
674 #[tokio::test]
675 async fn an_undeclared_parameter_is_refused_before_the_transport_is_reached() {
676 let client = client("http://127.0.0.1:8081", Value::Null);
677 let err = client
678 .call::<Value>(ops::GET_SESSION, vec![("payolad", "full".to_owned())])
679 .await
680 .unwrap_err();
681 assert!(err.to_string().contains("payolad"), "got: {err}");
682 assert!(
683 client.transport().seen.borrow().is_empty(),
684 "nothing may be sent for a call the contract refused",
685 );
686 }
687
688 #[tokio::test]
689 async fn a_typed_body_reaches_the_transport_as_the_contracts_own_json() {
690 let client = client("http://127.0.0.1:8081", serde_json::json!({"id": "sk-1"}));
694 let skill = client
695 .create_skill(&CreateSkillRequest {
696 name: "gum".to_owned(),
697 ..Default::default()
698 })
699 .await
700 .unwrap();
701
702 assert_eq!(skill.id, "sk-1");
703 let bodies = client.transport().bodies.borrow();
704 let sent: Value = serde_json::from_str(bodies[0].as_deref().unwrap()).unwrap();
705 assert_eq!(sent["name"], "gum");
706 }
707
708 #[tokio::test]
709 async fn the_bodyless_facade_refuses_an_operation_that_requires_a_body() {
710 let client = client("http://127.0.0.1:8081", Value::Null);
714 let err = client
715 .call::<Value>(ops::CREATE_SKILL, Vec::new())
716 .await
717 .unwrap_err();
718
719 assert!(
720 err.to_string().contains("requires a request body"),
721 "got: {err}",
722 );
723 assert!(client.transport().seen.borrow().is_empty());
724 }
725
726 #[tokio::test]
727 async fn the_facade_refuses_a_body_on_an_operation_that_declares_none() {
728 let client = client("http://127.0.0.1:8081", Value::Null);
729 let err = client
730 .call_with_body::<Value>(
731 ops::GET_SESSION,
732 vec![("id", "s-1".to_owned())],
733 Some("{}".to_owned()),
734 )
735 .await
736 .unwrap_err();
737
738 assert!(
739 err.to_string().contains("declares no request body"),
740 "got: {err}",
741 );
742 assert!(client.transport().seen.borrow().is_empty());
743 }
744
745 #[tokio::test]
746 async fn the_bodyless_facade_still_sends_no_body_for_an_ordinary_read() {
747 let client = client("http://127.0.0.1:8081", serde_json::json!({"items": []}));
750 let _ = client
751 .list_sessions(&SessionListParams::default())
752 .await
753 .unwrap();
754 assert_eq!(client.transport().bodies.borrow().as_slice(), [None]);
755 }
756
757 #[tokio::test]
758 async fn a_named_method_and_its_operation_id_build_the_same_request() {
759 let named = client("http://127.0.0.1:8081", serde_json::json!({}));
763 let _ = named.get_span("t-1", "sp-1").await.unwrap();
764
765 let raw = client("http://127.0.0.1:8081", serde_json::json!({}));
766 let _: Value = raw
767 .call(
768 ops::GET_SPAN,
769 vec![
770 ("trace_id", "t-1".to_owned()),
771 ("span_id", "sp-1".to_owned()),
772 ],
773 )
774 .await
775 .unwrap();
776
777 assert_eq!(
778 *named.transport().seen.borrow(),
779 *raw.transport().seen.borrow()
780 );
781 }
782}