laser_sdk/agent/
contract.rs

1use crate::agent::clock::{Clock, SystemClock};
2use crate::agent::consumer::AgentMessage;
3use crate::agent::laser::ContractTerminal;
4use crate::agent::router::{CapabilitySelector, InboxRoute, Router};
5use crate::error::LaserError;
6use crate::laser::Laser;
7use crate::provenance::AgentTopic;
8use crate::types::{AgentId, MintUlid};
9use laser_wire::agent::{ConversationId, CorrelationId};
10use std::time::Duration;
11use tokio::time::{Instant, sleep};
12
13impl Laser {
14    /// Scatter a directed contract to every agent advertising `selector`'s skill,
15    /// concurrently, and collect the reply body of each that completes (excluding
16    /// unavailable and quarantined agents, which capability resolution drops). The
17    /// Laser-level gather the workflow's all-capable step and a verifier panel
18    /// build on. Empty when no capable agent completes, [`LaserError::NoCapableAgent`]
19    /// when none advertise the skill at all. A convenience over
20    /// [`scatter_report`](Self::scatter_report), which keeps per-agent outcomes.
21    pub async fn scatter(
22        &self,
23        source: AgentId,
24        selector: &CapabilitySelector,
25        payload: &[u8],
26        inbox_route: &InboxRoute,
27        deadline: Duration,
28    ) -> Result<Vec<Vec<u8>>, LaserError> {
29        let report = self
30            .scatter_report(source, selector, payload, inbox_route, deadline)
31            .await?;
32        Ok(report
33            .completed()
34            .map(|(_, reply)| reply.body().to_vec())
35            .collect())
36    }
37
38    /// Like [`scatter`](Self::scatter) but returns the outcome of every branch
39    /// attributed to its agent, so a caller sees which agent completed, failed, or
40    /// timed out rather than a bag of bodies. An all-failed scatter is a report of
41    /// failures, distinguishable from an empty one (no completions).
42    pub async fn scatter_report(
43        &self,
44        source: AgentId,
45        selector: &CapabilitySelector,
46        payload: &[u8],
47        inbox_route: &InboxRoute,
48        deadline: Duration,
49    ) -> Result<ScatterReport, LaserError> {
50        let now = SystemClock.now_micros();
51        let agents: Vec<AgentId> = {
52            let mut registry = self.agent_registry()?;
53            registry.refresh(now).await?;
54            #[cfg(feature = "query")]
55            if matches!(inbox_route, InboxRoute::Advertised) || selector.principal.is_some() {
56                registry.refresh_presence().await?;
57            }
58            Router::AllCapable(selector.clone()).resolve_targets(&registry, now)?
59        };
60        if agents.is_empty() {
61            return Err(LaserError::NoCapableAgent {
62                skill: selector.skill.clone(),
63            });
64        }
65
66        let mut branches = tokio::task::JoinSet::new();
67        for agent in agents {
68            let laser = self.clone();
69            let source = source.clone();
70            let route = inbox_route.clone();
71            let body = payload.to_vec();
72            branches.spawn(async move {
73                let result = laser
74                    .contract(Router::to(agent.clone()))
75                    .from(source)
76                    .payload(body)
77                    .inbox_route(route)
78                    .deadline(deadline)
79                    .send()
80                    .await;
81                (agent, result)
82            });
83        }
84
85        let mut outcomes = Vec::new();
86        while let Some(joined) = branches.join_next().await {
87            match joined {
88                Ok((agent, result)) => outcomes.push(ScatterOutcome { agent, result }),
89                Err(join_error) => {
90                    tracing::warn!(%join_error, "scatter branch panicked");
91                }
92            }
93        }
94        Ok(ScatterReport { outcomes })
95    }
96}
97
98/// One agent's outcome in a [`ScatterReport`].
99#[derive(Debug)]
100pub struct ScatterOutcome {
101    /// The agent this branch was contracted to.
102    pub agent: AgentId,
103    /// Its terminal contract state, or the error that failed the branch.
104    pub result: Result<Contract, LaserError>,
105}
106
107/// The per-agent outcome of a [`scatter_report`](Laser::scatter_report): every
108/// contracted agent's terminal state, so an all-failed scatter is a report of
109/// failures rather than an empty success.
110#[derive(Debug)]
111pub struct ScatterReport {
112    /// One entry per contracted agent, in completion order.
113    pub outcomes: Vec<ScatterOutcome>,
114}
115
116impl ScatterReport {
117    /// The agents that completed, each with its reply.
118    pub fn completed(&self) -> impl Iterator<Item = (&AgentId, &AgentMessage)> {
119        self.outcomes
120            .iter()
121            .filter_map(|outcome| match &outcome.result {
122                Ok(Contract::Completed(reply)) => Some((&outcome.agent, reply)),
123                _ => None,
124            })
125    }
126
127    /// The agents whose branch errored (an infrastructure or send failure), each
128    /// with its cause. A non-completing terminal (`Failed`/`TimedOut`/`NotConsumed`)
129    /// is an `Ok` outcome, read it from [`outcomes`](Self::outcomes).
130    pub fn failures(&self) -> impl Iterator<Item = (&AgentId, &LaserError)> {
131        self.outcomes
132            .iter()
133            .filter_map(|outcome| match &outcome.result {
134                Err(error) => Some((&outcome.agent, error)),
135                _ => None,
136            })
137    }
138}
139
140/// The outcome of a [`contract`](Laser::contract): a directed request to one
141/// agent, with a deadline and an optional consumption expiry, resolved to one
142/// terminal state.
143#[derive(Debug)]
144pub enum Contract {
145    /// The target replied within the deadline (a non-error reply).
146    Completed(AgentMessage),
147    /// The target replied with a terminal `error`.
148    Failed(AgentMessage),
149    /// The command was not consumed within the expiry: no pickup acknowledgment
150    /// landed, and the `deadline_micros` dropped it before any handler ran. Only
151    /// reported when [`expire_if_not_consumed`](ContractBuilder::expire_if_not_consumed)
152    /// is set and the target acknowledges pickup
153    /// ([`Agent::ack_on_pickup`](crate::agent::Agent)).
154    NotConsumed,
155    /// Consumed (or no expiry was set) but no terminal reply landed within the
156    /// completion deadline.
157    TimedOut,
158}
159
160/// A directed request to one agent, bundling resolution, a deadline, and an
161/// optional consumption expiry into one call. The "narrow and concrete"
162/// orchestration request: hand a task to one agent (or one capability) and learn
163/// the reply, whether it was picked up, or that it did not finish in time,
164/// without hand-rolling correlation ids and timers.
165///
166/// Built by [`Laser::contract`], then `.send().await`. It is directed messaging
167/// with confirmation over the log, never a bespoke consensus protocol: a thin
168/// client-side state machine over the reply topic and two durations. The command
169/// is an AGDX command, so a target built with
170/// [`ack_on_pickup`](crate::agent::Agent) emits a `Working` status the contract
171/// reads as the consumption signal.
172pub struct ContractBuilder<'a> {
173    laser: &'a Laser,
174    router: Router,
175    from: Option<AgentId>,
176    payload: Vec<u8>,
177    inbox_route: InboxRoute,
178    reply_topic: AgentTopic<'static>,
179    expiry: Option<Duration>,
180    deadline: Duration,
181    fence: Option<u64>,
182    conversation: Option<ConversationId>,
183    parent: Option<(ConversationId, ConversationId)>,
184}
185
186impl Laser {
187    /// Open a [`ContractBuilder`] addressed by `router` (one agent via
188    /// [`Router::to`], or one capability via [`Router::to_capable`]). A broadcast
189    /// or all-capable route is rejected at send time, since a contract is directed
190    /// to exactly one target (use [`AgentCtx::fan_out`](crate::agent::AgentCtx::fan_out)
191    /// to scatter).
192    pub fn contract(&self, router: Router) -> ContractBuilder<'_> {
193        ContractBuilder {
194            laser: self,
195            router,
196            from: None,
197            payload: Vec::new(),
198            inbox_route: InboxRoute::default(),
199            reply_topic: AgentTopic::Sessions,
200            expiry: None,
201            deadline: Duration::from_secs(30),
202            fence: None,
203            conversation: None,
204            parent: None,
205        }
206    }
207}
208
209impl ContractBuilder<'_> {
210    /// The agent id the contract sends as (the orchestrator). Required: the
211    /// command is an AGDX command, which carries a source.
212    pub fn from(mut self, source: AgentId) -> Self {
213        self.from = Some(source);
214        self
215    }
216
217    /// The task payload sent to the target.
218    pub fn payload(mut self, payload: impl Into<Vec<u8>>) -> Self {
219        self.payload = payload.into();
220        self
221    }
222
223    /// How the target is resolved to the topic its work is sent on (default
224    /// [`InboxRoute::Advertised`]).
225    pub fn inbox_route(mut self, inbox_route: InboxRoute) -> Self {
226        self.inbox_route = inbox_route;
227        self
228    }
229
230    /// The topic the contract awaits the reply on (default
231    /// [`AgentTopic::Sessions`]). Must be where the target replies and acks.
232    pub fn reply_on(mut self, reply_topic: AgentTopic<'static>) -> Self {
233        self.reply_topic = reply_topic;
234        self
235    }
236
237    /// Drop the command if it is not consumed within `expiry`: rides the command
238    /// as its `deadline_micros`, so the reliable consumer drops an expired command
239    /// before the handler runs. This is the consumption expiry, distinct from the
240    /// completion [`deadline`](Self::deadline). Reported as
241    /// [`Contract::NotConsumed`] only when the target acknowledges pickup.
242    pub fn expire_if_not_consumed(mut self, expiry: Duration) -> Self {
243        self.expiry = Some(expiry);
244        self
245    }
246
247    /// How long to await a terminal reply before returning [`Contract::TimedOut`]
248    /// (default 30s). Client-side, distinct from the consumption expiry.
249    pub fn deadline(mut self, deadline: Duration) -> Self {
250        self.deadline = deadline;
251        self
252    }
253
254    /// Pin the conversation the command rides. The fence gate buckets its
255    /// monotonic high-water mark by conversation, so a fenced task that may be
256    /// re-dispatched (a reassignment, a workflow resume) must use the SAME
257    /// conversation on every attempt, or each lands in its own bucket and the
258    /// gate never compares the stale holder's token against the new one. Defaults
259    /// to a fresh conversation per send, correct for a one-shot, non-fenced
260    /// contract. [`fence`](Self::fence) is only meaningful with a pinned, stable
261    /// conversation.
262    pub fn conversation(mut self, conversation: crate::types::ConversationId) -> Self {
263        self.conversation = Some(conversation.into());
264        self
265    }
266
267    /// Stamp a fence token (`agdx.fence`) on the command. The token comes from a
268    /// lease grant ([`Lease::token`](crate::kv::Lease)). It buys two distinct
269    /// guarantees, and only together do they give at-most-once:
270    ///
271    /// - **Same-holder replay gate (local).** A consumer drops a command whose
272    ///   fence is below the highest it has accepted for the task, so a stale
273    ///   holder cannot act after a reassignment. This gate is per-consumer-process
274    ///   and keyed by conversation, so it only bites when the task rides a stable
275    ///   [`conversation`](Self::conversation) across re-dispatches. Without that
276    ///   pin (or across separate agent replicas, which do not share the gate) it
277    ///   does nothing.
278    /// - **Cross-holder effect gate (durable).** For an at-most-once EXTERNAL
279    ///   effect across different holders or replicas, the handler must commit the
280    ///   effect through a fenced compare-and-swap
281    ///   ([`Kv::cas_fenced`](crate::kv::Kv::cas_fenced)) carrying this token, so
282    ///   the managed backend rejects a lower-token writer at the sink. The engine provides
283    ///   the token. The durable guarantee is the handler's to claim.
284    pub fn fence(mut self, fence: u64) -> Self {
285        self.fence = Some(fence);
286        self
287    }
288
289    /// Run the contract as a child session of `parent`, in the tree rooted at
290    /// `root` (the parent itself when it has no parent). [`send`](Self::send)
291    /// writes the child's submitted start on `agent.sessions` before the
292    /// command, stamps the ancestry on the command, and ends the child by the
293    /// outcome: completed on a reply, failed otherwise. A lifecycle record
294    /// that fails to publish surfaces as the error unless the contract itself
295    /// already erred.
296    pub fn parent(
297        mut self,
298        parent: crate::types::ConversationId,
299        root: crate::types::ConversationId,
300    ) -> Self {
301        self.parent = Some((parent.into(), root.into()));
302        self
303    }
304
305    /// Resolve the target and its inbox, send the command, and run the contract
306    /// state machine until a terminal state.
307    #[tracing::instrument(target = "laser", level = "info", skip_all, fields(conversation = self.conversation.as_ref().map(tracing::field::display), operation = "contract"))]
308    pub async fn send(self) -> Result<Contract, LaserError> {
309        let source = self.from.clone().ok_or_else(|| {
310            LaserError::Invalid("a contract requires `.from(source agent id)`".to_owned())
311        })?;
312        #[cfg(feature = "sign")]
313        let expected_principal = self.router.required_principal();
314        let (target, inbox) = self.resolve().await?;
315
316        let laser = self.laser;
317        let expiry = self.expiry;
318        let deadline = self.deadline;
319        let conversation = self.conversation.unwrap_or_else(ConversationId::mint);
320        let correlation = CorrelationId::mint();
321
322        // A child session's start lands before its command, so a reader never
323        // sees work for a session that does not exist yet.
324        let child = match self.parent {
325            Some((parent, root)) => Some(
326                start_child(
327                    laser,
328                    &source.wire_id(),
329                    &target.wire_id(),
330                    conversation,
331                    parent,
332                    root,
333                )
334                .await?,
335            ),
336            None => None,
337        };
338
339        // Seed the reply reader at the topic tail BEFORE sending, so it reads only
340        // the ack and the reply, never the topic's history. Under a verifier the
341        // reply signer must be the route's authenticated principal when one was
342        // required, otherwise the resolved target's enrolled identity.
343        let mut reader = laser
344            .agdx_reply_reader(self.reply_topic.clone(), Some(&source.wire_id()))
345            .await?;
346        #[cfg(feature = "sign")]
347        {
348            reader.expected_signer = Some(
349                expected_principal
350                    .map(|principal| principal.to_string())
351                    .unwrap_or_else(|| target.to_string()),
352            );
353        }
354
355        // The command carries the AGDX envelope correlation. A worker handling it
356        // sees that correlation as its message's `correlation_id` (synthesized from
357        // the envelope), and `ctx.respond` echoes `correlation_id` onto its plain
358        // reply, so the contract matches the terminal by correlation on both the
359        // AGDX (`envelope.correlation`) and the plain (`agdx.corr`) shapes. The
360        // `Working` ack matches on the envelope correlation directly.
361        let producer = laser.agdx(AgentTopic::Custom(&inbox), source.wire_id(), conversation);
362        let mut command = producer
363            .command(correlation, self.payload)
364            .with_target(target.wire_id());
365        if let Some(fence) = self.fence {
366            command = command.with_metadata(
367                laser_wire::headers::FENCE,
368                laser_wire::query::Value::Uint(fence),
369            );
370        }
371        if let Some((parent, root)) = self.parent {
372            command = command
373                .with_ancestry(Some(parent), Some(root))
374                .with_metadata(
375                    laser_wire::agent::METADATA_SUBMITTED,
376                    laser_wire::query::Value::Bool(true),
377                );
378        }
379        if let Some(expiry) = expiry {
380            let at = SystemClock
381                .now_micros()
382                .saturating_add(expiry.as_micros().min(u128::from(u64::MAX)) as u64);
383            command = command.with_deadline_micros(at);
384        }
385        command.send().await?;
386
387        let outcome = watch_terminal(laser, &mut reader, correlation, expiry, deadline).await;
388
389        // End the child by the outcome before returning. A lifecycle record that
390        // fails to publish surfaces as the error, except when the contract
391        // itself already erred.
392        if let Some(child) = child {
393            let ended = match &outcome {
394                Ok(Contract::Completed(_)) => child.end().await,
395                Ok(other) => child.fail(contract_failure(failure_reason(other))).await,
396                Err(error) => child.fail(contract_failure(error.to_string())).await,
397            };
398            if let (Ok(_), Err(error)) = (&outcome, ended) {
399                return Err(error);
400            }
401        }
402        outcome
403    }
404
405    /// Resolve the route to exactly one target and its inbox topic in one registry
406    /// read. Errors on a non-directed route or no match.
407    async fn resolve(&self) -> Result<(AgentId, iggy::prelude::Identifier), LaserError> {
408        if matches!(self.router, Router::Broadcast | Router::AllCapable(_)) {
409            return Err(LaserError::Invalid(
410                "a contract is directed to one agent, not a broadcast or all-capable route"
411                    .to_owned(),
412            ));
413        }
414        let mut registry = self.laser.agent_registry()?;
415        let now = SystemClock.now_micros();
416        registry.refresh(now).await?;
417        #[cfg(feature = "query")]
418        if matches!(self.inbox_route, InboxRoute::Advertised) || self.router.requires_presence() {
419            registry.refresh_presence().await?;
420        }
421        let target = self
422            .router
423            .resolve_targets(&registry, now)?
424            .into_iter()
425            .next()
426            .ok_or_else(|| {
427                LaserError::Invalid("the contract route resolved no target".to_owned())
428            })?;
429        let inbox = self
430            .inbox_route
431            .resolve(&target, registry.inbox_for(&target))?;
432        Ok((target, inbox))
433    }
434}
435
436// Write a child session's submitted start as `source` and return its lens.
437pub(crate) async fn start_child(
438    laser: &Laser,
439    source: &laser_wire::agent::AgentId,
440    target: &laser_wire::agent::AgentId,
441    conversation: ConversationId,
442    parent: ConversationId,
443    root: ConversationId,
444) -> Result<crate::agent::Session, LaserError> {
445    let sessions = laser.sessions();
446    let start = laser_wire::agent::SessionStart {
447        label: None,
448        namespace: None,
449        agent: target.clone(),
450        sdk: sessions.config().sdk_info().clone(),
451        parent: Some(parent),
452        root: Some(root),
453        idle_timeout_micros: Some(
454            u64::try_from(sessions.config().idle_timeout_value().as_micros()).unwrap_or(u64::MAX),
455        ),
456        budget: None,
457        tags: Vec::new(),
458    };
459    let session = sessions
460        .open(conversation.into())
461        .as_agent(source.clone())
462        .with_ancestry(Some(parent.into()), Some(root.into()));
463    session
464        .lane()?
465        .status(laser_wire::agent::OPERATION_SESSION)
466        .with_task_state(laser_wire::agent::TaskState::Submitted)
467        .body(laser_wire::framing::encode_named(&start)?)
468        .content_type(laser_wire::content::ContentType::Cbor)
469        .with_ancestry(Some(parent), Some(root))
470        .send()
471        .await?;
472    Ok(session)
473}
474
475fn failure_reason(outcome: &Contract) -> String {
476    match outcome {
477        Contract::Completed(_) => "completed".to_owned(),
478        Contract::Failed(_) => "the target replied with a terminal error".to_owned(),
479        Contract::NotConsumed => "the command was not consumed within the expiry".to_owned(),
480        Contract::TimedOut => "no terminal reply landed within the deadline".to_owned(),
481    }
482}
483
484fn contract_failure(message: String) -> laser_wire::agent::AgentErrorBody {
485    laser_wire::agent::AgentErrorBody {
486        code: laser_wire::agent::AgentErrorCode::Internal,
487        message: Some(message),
488        retryable: false,
489        detail: None,
490    }
491}
492
493/// Watch the reply topic for the pickup ack and the terminal, bounded by the
494/// consumption expiry and the completion deadline.
495async fn watch_terminal(
496    laser: &Laser,
497    reader: &mut crate::agent::laser::AgentReplyReader,
498    correlation: CorrelationId,
499    expiry: Option<Duration>,
500    deadline: Duration,
501) -> Result<Contract, LaserError> {
502    let start = Instant::now();
503    let expiry_at = expiry.map(|expiry| start + expiry);
504    let deadline_at = start + deadline;
505    let mut consumed = false;
506    loop {
507        let pass = reader.poll_contract(&laser.client(), correlation).await?;
508        match pass.terminal {
509            Some(ContractTerminal::Completed(reply)) => return Ok(Contract::Completed(reply)),
510            Some(ContractTerminal::Failed(reply)) => return Ok(Contract::Failed(reply)),
511            None => {}
512        }
513        consumed |= pass.consumed;
514
515        let now = Instant::now();
516        if !consumed
517            && let Some(expiry_at) = expiry_at
518            && now >= expiry_at
519        {
520            return Ok(Contract::NotConsumed);
521        }
522        if now >= deadline_at {
523            return Ok(Contract::TimedOut);
524        }
525        if !pass.read_any {
526            // Never sleep past the deadline a tight contract set.
527            let remaining = deadline_at.saturating_duration_since(Instant::now());
528            sleep(Duration::from_millis(100).min(remaining)).await;
529        }
530    }
531}
laser_sdk/agent/contract.rs

laser_sdk/agent/
contract.rs
