okbot_core/
peer.rs

1//! The peer port (ADR 0014): delegating work to other okbots of a team.
2//!
3//! Outbound, the runtime calls [`Peers`] when an agent delegates. Inbound, a peer adapter is also a
4//! [`Channel`](crate::Channel) (`peer:<team>`): each accepted task arrives as an ordinary
5//! [`InboundEvent`](crate::InboundEvent), and [`Peers::caller`] says who sent it.
6
7use std::fmt;
8use std::str::FromStr;
9use std::time::Duration;
10
11use jiff::Timestamp;
12use serde::{Deserialize, Serialize};
13
14use crate::id::{ConversationKey, UserRef};
15use crate::{BoxError, async_trait};
16
17/// The platform name of peer channels (`peer:<team>`).
18pub const PEER_PLATFORM: &str = "peer";
19
20/// Event ids starting with this, on a peer channel, stop the conversation's running turn: the task
21/// was cancelled by its caller. (Another bot's text never runs as a command, so `/stop` cannot.)
22pub const PEER_CANCEL_PREFIX: &str = "peer-cancel:";
23
24/// A peer operation failed.
25#[derive(Debug, thiserror::Error)]
26#[non_exhaustive]
27pub enum PeerError {
28    /// No member matches the target, or no link allows the call.
29    #[error("no member of the team can take this: {0}")]
30    NoRoute(String),
31    /// The other bot refused the request (its reason, safe to show).
32    #[error("refused: {0}")]
33    Refused(String),
34    /// The other bot could not be reached or did not answer in time.
35    #[error("unreachable: {0}")]
36    Unreachable(String),
37    /// No such task.
38    #[error("no such task")]
39    UnknownTask,
40    /// Any other failure.
41    #[error("peer error: {0}")]
42    Other(#[source] BoxError),
43}
44
45/// Who a delegation is for, where it started and how much it may still spend. It travels with
46/// every request of a chain and is signed with it.
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
48pub struct DelegationContext {
49    /// Where the work started.
50    pub origin: Origin,
51    /// Team members the work passed through, the first bot first.
52    pub chain: Vec<String>,
53    /// The work must be done by then; only shrinks along the chain.
54    pub deadline: Timestamp,
55    /// Agent turns the rest of the chain may still use.
56    pub budget: Budget,
57    /// An id shared by every bot's records of this work.
58    pub trace: String,
59}
60
61/// The person and conversation a delegation started from.
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
63pub struct Origin {
64    /// The member name of the first bot.
65    pub bot: String,
66    /// The conversation on the first bot (its storage text form).
67    pub conversation: String,
68    /// The person's id in the team manifest, when the team knows them.
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub person: Option<String>,
71    /// The person's display name.
72    pub name: String,
73}
74
75/// What a chain of delegations may still spend.
76#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
77pub struct Budget {
78    /// Agent turns.
79    pub turns: u32,
80    /// Input plus output tokens of finished tasks; `None`: no token limit.
81    #[serde(default, skip_serializing_if = "Option::is_none")]
82    pub tokens: Option<u64>,
83}
84
85/// Whom to delegate to.
86#[derive(Debug, Clone, PartialEq, Eq, Hash)]
87pub enum PeerTarget {
88    /// One member, by name.
89    Member(String),
90    /// Any member with this team role (`role:<name>`).
91    Role(String),
92    /// Any member with this skill (`skill:<name>`).
93    Skill(String),
94}
95
96impl FromStr for PeerTarget {
97    type Err = PeerError;
98
99    fn from_str(s: &str) -> Result<Self, Self::Err> {
100        let s = s.trim();
101        let target = if let Some(role) = s.strip_prefix("role:") {
102            Self::Role(role.to_string())
103        } else if let Some(skill) = s.strip_prefix("skill:") {
104            Self::Skill(skill.to_string())
105        } else {
106            Self::Member(s.to_string())
107        };
108        match &target {
109            Self::Member(n) | Self::Role(n) | Self::Skill(n) if n.is_empty() => {
110                Err(PeerError::NoRoute("an empty target".into()))
111            }
112            _ => Ok(target),
113        }
114    }
115}
116
117impl fmt::Display for PeerTarget {
118    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
119        match self {
120            Self::Member(n) => f.write_str(n),
121            Self::Role(n) => write!(f, "role:{n}"),
122            Self::Skill(n) => write!(f, "skill:{n}"),
123        }
124    }
125}
126
127/// A request to hand work to another bot.
128#[derive(Debug, Clone, PartialEq, Eq)]
129pub struct Delegation {
130    /// The team; `None`: any team of this bot with a matching member.
131    pub team: Option<String>,
132    /// Whom.
133    pub to: PeerTarget,
134    /// The work, in plain language.
135    pub task: String,
136    /// Who it is for and what it may spend.
137    pub context: DelegationContext,
138    /// How long to wait for the result before returning a running task.
139    pub wait: Duration,
140    /// Identifies this request: sending it again returns the same task.
141    pub message_id: String,
142    /// Who asked, so a result that comes after a restart still reaches them.
143    pub requester: Option<UserRef>,
144    /// The origin conversation is a one-to-one chat.
145    pub direct: bool,
146}
147
148/// Where a task stands (A2A task states).
149#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
150#[serde(rename_all = "kebab-case")]
151pub enum PeerTaskState {
152    /// Accepted, not started.
153    Submitted,
154    /// Running.
155    Working,
156    /// Waiting for a person (an approval).
157    InputRequired,
158    /// Done; the text is the answer.
159    Completed,
160    /// Failed; the text says why.
161    Failed,
162    /// Cancelled.
163    Canceled,
164    /// Refused by the other bot.
165    Rejected,
166}
167
168impl PeerTaskState {
169    /// The task will not change any more.
170    pub fn is_final(self) -> bool {
171        matches!(
172            self,
173            Self::Completed | Self::Failed | Self::Canceled | Self::Rejected
174        )
175    }
176
177    /// The A2A name (`input-required`).
178    pub fn as_str(self) -> &'static str {
179        match self {
180            Self::Submitted => "submitted",
181            Self::Working => "working",
182            Self::InputRequired => "input-required",
183            Self::Completed => "completed",
184            Self::Failed => "failed",
185            Self::Canceled => "canceled",
186            Self::Rejected => "rejected",
187        }
188    }
189}
190
191impl FromStr for PeerTaskState {
192    type Err = PeerError;
193
194    fn from_str(s: &str) -> Result<Self, Self::Err> {
195        Ok(match s {
196            "submitted" => Self::Submitted,
197            "working" => Self::Working,
198            "input-required" => Self::InputRequired,
199            "completed" => Self::Completed,
200            "failed" => Self::Failed,
201            "canceled" => Self::Canceled,
202            "rejected" => Self::Rejected,
203            other => return Err(PeerError::Refused(format!("unknown task state {other}"))),
204        })
205    }
206}
207
208/// A task as the caller sees it.
209#[derive(Debug, Clone, PartialEq, Eq)]
210pub struct PeerTask {
211    /// The team.
212    pub team: String,
213    /// The member doing it.
214    pub peer: String,
215    /// The task id (the other bot's).
216    pub id: String,
217    /// Where it stands.
218    pub state: PeerTaskState,
219    /// The answer, the reason for a failure, or the question while input is required.
220    pub text: Option<String>,
221    /// While input is required: the question's buttons, to show the origin person.
222    pub prompt: Option<PeerPrompt>,
223    /// Tokens the task used (its turn and its own delegations), once it ended.
224    pub tokens: Option<u64>,
225}
226
227/// A question a task asks the origin person (an approval), answered with one of its buttons.
228#[derive(Debug, Clone, PartialEq, Eq)]
229pub struct PeerPrompt {
230    /// The prompt id on the other bot.
231    pub id: String,
232    /// `(button id, label)`, in order.
233    pub buttons: Vec<(String, String)>,
234}
235
236/// A member another bot may delegate to, as the agent sees it.
237#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
238pub struct PeerInfo {
239    /// The team.
240    pub team: String,
241    /// The member's name.
242    pub name: String,
243    /// Team roles.
244    pub roles: Vec<String>,
245    /// Skills.
246    pub skills: Vec<String>,
247    /// What it does.
248    pub description: String,
249}
250
251/// The authenticated sender of an inbound task, for the runtime's permission checks.
252#[derive(Debug, Clone, PartialEq, Eq)]
253pub struct PeerCaller {
254    /// The team.
255    pub team: String,
256    /// The calling member.
257    pub member: String,
258    /// Its team roles (from the verified manifest).
259    pub roles: Vec<String>,
260    /// The delegation context it sent.
261    pub context: DelegationContext,
262}
263
264/// The work done, as a direction-tagged record kept by the store (`peer_tasks`).
265#[derive(Debug, Clone, PartialEq, Eq)]
266pub struct PeerTaskRecord {
267    /// This bot's id for the task: the A2A task id for inbound tasks, the callee's for outbound.
268    pub id: String,
269    /// Received from another bot (`true`) or sent to one.
270    pub inbound: bool,
271    /// The team.
272    pub team: String,
273    /// The other member.
274    pub peer: String,
275    /// The A2A context id.
276    pub context_id: String,
277    /// The A2A message id that created it (idempotency key, unique per team, peer and direction).
278    pub message_id: String,
279    /// Where it stands.
280    pub state: PeerTaskState,
281    /// The conversation that delegated it (outbound) or that runs it (inbound).
282    pub conversation: Option<ConversationKey>,
283    /// The trace id of the delegation.
284    pub trace: String,
285    /// The answer or the reason for a failure.
286    pub text: Option<String>,
287    /// Outbound: who asked, to bring a late result to them.
288    pub requester: Option<UserRef>,
289    /// Outbound: the conversation is a one-to-one chat.
290    pub direct: bool,
291    /// Tokens the task used, once it ended.
292    pub tokens: Option<u64>,
293    /// When it was created.
294    pub created_at: Timestamp,
295    /// When it last changed.
296    pub updated_at: Timestamp,
297}
298
299/// Delegates work to other bots of this bot's teams. Implemented by the peer adapter.
300#[async_trait]
301pub trait Peers: Send + Sync {
302    /// Members this bot may delegate to (the manifest's links from this bot).
303    async fn directory(&self) -> Result<Vec<PeerInfo>, PeerError>;
304
305    /// Hands work to a member and waits up to [`Delegation::wait`] for it to end.
306    async fn delegate(&self, delegation: Delegation) -> Result<PeerTask, PeerError>;
307
308    /// The current state of a task this bot delegated.
309    async fn status(&self, team: &str, peer: &str, task: &str) -> Result<PeerTask, PeerError>;
310
311    /// Asks the other bot to stop a task this bot delegated.
312    async fn cancel(&self, team: &str, peer: &str, task: &str) -> Result<PeerTask, PeerError>;
313
314    /// The tokens the turn answering the task of `conversation` (`task`, or the oldest) used, its
315    /// own delegations included; reported with the task's end.
316    async fn report_usage(&self, conversation: &ConversationKey, task: Option<&str>, tokens: u64);
317
318    /// What the turn answering the task of `conversation` (`task`, or the oldest) is doing now,
319    /// shown to the caller while the task works. Called from the agent's output reader: quick.
320    fn progress(&self, conversation: &ConversationKey, task: Option<&str>, text: &str);
321
322    /// The runtime could not answer the task of `conversation` (`task`, or the oldest): it fails
323    /// with `reason` instead of completing with a fixed reply.
324    async fn fail(&self, conversation: &ConversationKey, task: Option<&str>, reason: &str);
325
326    /// Answers the question of a task waiting for input (the origin person pressed `button`).
327    async fn answer(
328        &self,
329        task: &PeerTask,
330        prompt: &str,
331        button: &str,
332    ) -> Result<PeerTask, PeerError>;
333
334    /// The authenticated sender of an inbound event of a peer channel; `None` for other events.
335    async fn caller(&self, conversation: &ConversationKey) -> Option<PeerCaller>;
336
337    /// The team person whose accounts include `account` (`platform:instance/user-id`).
338    fn team_person(&self, account: &str) -> Option<String>;
339
340    /// This bot's member name in `team`.
341    fn member_name(&self, team: &str) -> Option<String>;
342}
343
344#[cfg(test)]
345mod tests {
346    use super::*;
347
348    #[test]
349    fn targets_parse() {
350        assert_eq!(
351            "ops".parse::<PeerTarget>().unwrap(),
352            PeerTarget::Member("ops".into())
353        );
354        assert_eq!(
355            "role:leader".parse::<PeerTarget>().unwrap(),
356            PeerTarget::Role("leader".into())
357        );
358        assert_eq!(
359            "skill:web".parse::<PeerTarget>().unwrap().to_string(),
360            "skill:web"
361        );
362        assert!("role:".parse::<PeerTarget>().is_err());
363        assert!("".parse::<PeerTarget>().is_err());
364    }
365
366    #[test]
367    fn states_round_trip() {
368        for state in [
369            PeerTaskState::Submitted,
370            PeerTaskState::Working,
371            PeerTaskState::InputRequired,
372            PeerTaskState::Completed,
373            PeerTaskState::Failed,
374            PeerTaskState::Canceled,
375            PeerTaskState::Rejected,
376        ] {
377            assert_eq!(state.as_str().parse::<PeerTaskState>().unwrap(), state);
378        }
379        assert!(PeerTaskState::Completed.is_final());
380        assert!(!PeerTaskState::InputRequired.is_final());
381    }
382}
okbot_core/peer.rs

okbot_core/
peer.rs
