atman_runtime/
rendezvous.rs

1use std::sync::Arc;
2
3use tokio::sync::oneshot;
4use uuid::Uuid;
5
6use crate::error::RuntimeError;
7
8// PromptId lives in atman-proto but runtime can't depend on proto (proto has no runtime deps).
9// Represent as Uuid at this trait boundary; adapter in daemon maps to proto::PromptId.
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
11pub struct PromptId(pub Uuid);
12
13impl PromptId {
14    pub fn now() -> Self {
15        Self(Uuid::now_v7())
16    }
17}
18
19impl std::fmt::Display for PromptId {
20    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
21        self.0.fmt(f)
22    }
23}
24
25pub trait PromptResolver: Send + Sync {
26    fn register(&self, id: PromptId) -> oneshot::Receiver<serde_json::Value>;
27    fn drop_pending(&self, id: &PromptId);
28
29    fn register_with_payload(
30        &self,
31        id: PromptId,
32        _kind: &str,
33        _payload: serde_json::Value,
34    ) -> oneshot::Receiver<serde_json::Value> {
35        self.register(id)
36    }
37}
38
39// In-proc fallback: prompts are auto-answered by a caller-provided default. Used when
40// no daemon is present; the flow author owns the auto-answer contract via tool args.
41pub struct AutoResolveResolver {
42    pub default: serde_json::Value,
43}
44
45impl PromptResolver for AutoResolveResolver {
46    fn register(&self, _id: PromptId) -> oneshot::Receiver<serde_json::Value> {
47        let (tx, rx) = oneshot::channel();
48        let _ = tx.send(self.default.clone());
49        rx
50    }
51    fn drop_pending(&self, _id: &PromptId) {}
52}
53
54pub async fn await_prompt(
55    resolver: &Arc<dyn PromptResolver>,
56    id: PromptId,
57    timeout: std::time::Duration,
58) -> Result<serde_json::Value, RuntimeError> {
59    await_prompt_inner(resolver.register(id), resolver, id, timeout).await
60}
61
62pub async fn await_prompt_with_payload(
63    resolver: &Arc<dyn PromptResolver>,
64    id: PromptId,
65    kind: &str,
66    payload: serde_json::Value,
67    timeout: std::time::Duration,
68) -> Result<serde_json::Value, RuntimeError> {
69    await_prompt_inner(
70        resolver.register_with_payload(id, kind, payload),
71        resolver,
72        id,
73        timeout,
74    )
75    .await
76}
77
78async fn await_prompt_inner(
79    rx: oneshot::Receiver<serde_json::Value>,
80    resolver: &Arc<dyn PromptResolver>,
81    id: PromptId,
82    timeout: std::time::Duration,
83) -> Result<serde_json::Value, RuntimeError> {
84    match tokio::time::timeout(timeout, rx).await {
85        Ok(Ok(v)) => Ok(v),
86        Ok(Err(_)) => {
87            resolver.drop_pending(&id);
88            Err(RuntimeError::ToolFailed(format!(
89                "prompt {id} channel closed before answer"
90            )))
91        }
92        Err(_) => {
93            resolver.drop_pending(&id);
94            Err(RuntimeError::ToolFailed(format!(
95                "prompt {id} timed out after {}s",
96                timeout.as_secs()
97            )))
98        }
99    }
100}
atman_runtime/rendezvous.rs

atman_runtime/
rendezvous.rs
