conversation_api/execution/wire/
mod.rs

1//! Versioned transport contract for Agent commands, events, and App Facade calls.
2//!
3//! This crate defines envelopes but no WebSocket client, authentication, reconnect policy, or
4//! server. Transport implementations belong to their deployment repository.
5
6/// Golden enqueue envelope used by deployment and Runtime conformance tests.
7pub const ENQUEUE_FIXTURE: &str = include_str!("../../../fixtures/execution/enqueue.v1.json");
8
9use crate::execution::{
10    AgentCommand, AgentObservation, ContextGeneration, ConversationSurface, DurableEvent, EventId,
11    FacadeRequest, FacadeResult, InvocationContext, ModelMode, PreparedAction, Revision, RunId,
12    Scope, ThreadId, UseCase,
13};
14use serde::{Deserialize, Serialize};
15use thiserror::Error;
16
17/// Current protocol major version.
18pub const CURRENT_VERSION: u16 = 1;
19
20/// Ephemeral cloud execution identity assigned by Lion.
21///
22/// Both values belong to the transport boundary and never enter the provider-neutral Runtime.
23#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
24pub struct DispatchBinding {
25    /// Unique identity for one execution attempt.
26    pub dispatch_id: String,
27    /// Authenticated Agent WebSocket session selected for the attempt.
28    pub agent_session_id: String,
29}
30
31/// Immutable object-store reference to one Runtime checkpoint generation.
32#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
33pub struct CheckpointReference {
34    /// Object key inside the configured checkpoint bucket.
35    pub object_key: String,
36    /// Lowercase hexadecimal SHA-256 of the checkpoint bytes.
37    pub sha256: String,
38    /// Runtime checkpoint format understood by `agent-rust`.
39    pub format_version: u32,
40}
41
42/// Metadata shared by every wire message.
43#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
44pub struct Metadata {
45    /// Protocol major version.
46    pub version: u16,
47    /// Globally unique message identifier.
48    pub message_id: String,
49    /// Identifier correlating a request and its response or emitted events.
50    pub correlation_id: String,
51    /// Optional identifier of the message that caused this one.
52    pub causation_id: Option<String>,
53    /// Sender timestamp in Unix milliseconds.
54    pub sent_at_unix_ms: u64,
55    /// Execution-attempt and Agent-session binding.
56    pub dispatch: DispatchBinding,
57}
58
59/// Request to execute an App Facade call over a transport.
60#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
61#[serde(tag = "phase", rename_all = "snake_case")]
62pub enum AppFacadeRequest {
63    /// Executes a direct App Facade request.
64    Invoke {
65        /// Authenticated invocation context.
66        context: InvocationContext,
67        /// Canonical Agent-selected request.
68        request: FacadeRequest,
69    },
70    /// Prepares a mutating operation.
71    Prepare {
72        /// Authenticated invocation context.
73        context: InvocationContext,
74        /// Canonical Agent-selected request.
75        request: FacadeRequest,
76    },
77    /// Commits a previously prepared operation.
78    Commit {
79        /// Authenticated invocation context.
80        context: InvocationContext,
81        /// Prepared operation and idempotency metadata.
82        action: PreparedAction,
83    },
84    /// Rejects a previously prepared operation.
85    Reject {
86        /// Authenticated invocation context.
87        context: InvocationContext,
88        /// Prepared operation and idempotency metadata.
89        action: PreparedAction,
90    },
91}
92
93/// Immediate response to command admission for the selected execution attempt.
94#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
95#[serde(tag = "type", rename_all = "snake_case")]
96pub enum CommandResponse {
97    /// Command was admitted; final state is delivered through events.
98    Accepted {
99        /// Authenticated context of the admitted command.
100        context: InvocationContext,
101        /// Stable admission disposition.
102        disposition: AdmissionDisposition,
103        /// Queue revision when the command mutated the admission queue.
104        queue_revision: Option<Revision>,
105    },
106    /// Command was rejected before admission.
107    Rejected {
108        /// Authenticated context of the rejected command.
109        context: InvocationContext,
110        /// Stable machine-readable error code.
111        code: String,
112        /// Safe diagnostic message.
113        message: String,
114    },
115}
116
117/// Stable wire representation of command admission.
118#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
119#[serde(rename_all = "snake_case")]
120pub enum AdmissionDisposition {
121    /// A Lion-dispatched request entered the Agent process's execution queue.
122    Queued,
123    /// Interaction decisions are durable and ready to resume.
124    InteractionReady,
125    /// Cancellation has been signaled to the live Agent process.
126    CancelRequested,
127}
128
129/// Routed best-effort observation sent outside the durable event outbox.
130#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
131pub struct ObservationMessage {
132    /// Authenticated invocation and routing context.
133    pub context: InvocationContext,
134    /// Request-scoped progress payload.
135    pub observation: AgentObservation,
136}
137
138/// Terminal transport failure for an admitted execution that could not produce a checkpoint event.
139#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
140pub struct ExecutionFailure {
141    pub context: InvocationContext,
142    pub code: String,
143    pub message: String,
144}
145
146/// Acknowledges one durable event after the receiver has applied or deduplicated it.
147#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
148pub struct EventAck {
149    pub event_id: EventId,
150    /// Checkpoint reference Lion actually selected while applying (or deduplicating) the event.
151    pub checkpoint: CheckpointReference,
152}
153
154/// Lion-owned archive GC request for all Runtime checkpoints under one private thread.
155#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
156pub struct CheckpointPurgeRequest {
157    pub purge_id: String,
158    pub conversation_key: String,
159    pub scope: Scope,
160    pub surface_id: ConversationSurface,
161    pub thread_id: ThreadId,
162}
163
164/// Result of one idempotent checkpoint-prefix purge.
165#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
166#[serde(tag = "status", rename_all = "snake_case")]
167pub enum CheckpointPurgeAck {
168    Deleted {
169        purge_id: String,
170        conversation_key: String,
171        thread_id: ThreadId,
172        deleted_objects: u32,
173    },
174    Failed {
175        purge_id: String,
176        conversation_key: String,
177        thread_id: ThreadId,
178        code: String,
179        message: String,
180    },
181}
182
183/// One concrete cloud model route frozen by the deployment before command admission.
184#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
185pub struct LlmRoute {
186    pub backend: String,
187    pub profile: String,
188    pub model: String,
189    pub provider_preferences: Vec<String>,
190    pub temperature: Option<f64>,
191    pub max_output_tokens: Option<u32>,
192    pub cache_control: Option<bool>,
193    pub reasoning_effort: Option<String>,
194    /// Total context window of the exact frozen model route.
195    pub context_window_tokens: u32,
196}
197
198/// One logical selector and its concrete deployment route.
199#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
200pub struct LlmRouteBinding {
201    pub use_case: UseCase,
202    pub model_mode: ModelMode,
203    pub revision: Option<String>,
204    pub route: LlmRoute,
205}
206
207/// Complete deployment-resolved LLM route set bound immutably to one admitted run.
208#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
209pub struct LlmExecutionPlan {
210    pub routes: Vec<LlmRouteBinding>,
211}
212
213impl LlmExecutionPlan {
214    /// Selects the exact frozen route for one provider-neutral Runtime request.
215    #[must_use]
216    pub fn route_for(
217        &self,
218        use_case: &UseCase,
219        model_mode: &ModelMode,
220    ) -> Option<&LlmRouteBinding> {
221        self.routes
222            .iter()
223            .find(|item| item.use_case == *use_case && item.model_mode == *model_mode)
224    }
225}
226
227/// Transport admission request. Environment bindings are consumed by the composition root and do
228/// not enter the provider-neutral Runtime command.
229#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
230pub struct CommandRequest {
231    pub command: AgentCommand,
232    /// Effective prompt-context generation selected by Lion for this dispatch.
233    pub context_generation: Option<ContextGeneration>,
234    pub llm_plan: Option<LlmExecutionPlan>,
235    /// Stable Runtime checkpoint selected by Lion, absent for a new thread.
236    pub base_checkpoint: Option<CheckpointReference>,
237    /// Attempt already terminalized by Lion and settled silently before new work starts.
238    pub abandoned_run_id: Option<RunId>,
239}
240
241/// Checkpoint-coupled Runtime event. Lion commits the generation before projecting the event.
242#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
243pub struct EventMessage {
244    pub event: DurableEvent,
245    pub checkpoint: CheckpointReference,
246}
247
248/// Result of an App Facade request transported back to Runtime.
249#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
250#[serde(tag = "type", rename_all = "snake_case")]
251pub enum AppFacadeResponse {
252    /// Read-only or committed operation result.
253    Result {
254        /// Structured App Facade result.
255        result: FacadeResult,
256    },
257    /// Prepared operation awaiting a commit or rejection.
258    Prepared {
259        /// Prepared action.
260        action: PreparedAction,
261    },
262    /// Rejection completed successfully.
263    Rejected,
264    /// Stable error safe to expose across the transport.
265    Failed {
266        /// Machine-readable category.
267        code: String,
268        /// Safe diagnostic message.
269        message: String,
270        /// Optional retry delay.
271        retry_after_ms: Option<u64>,
272    },
273}
274
275/// Payload families carried by the versioned envelope.
276#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
277#[serde(tag = "family", content = "payload", rename_all = "snake_case")]
278pub enum Payload {
279    /// Command entering Runtime.
280    Command(Box<CommandRequest>),
281    /// Immediate command-admission response.
282    CommandResponse(CommandResponse),
283    /// Event leaving Runtime.
284    Event(EventMessage),
285    /// Durable receiver acknowledgement used to clear the Runtime outbox.
286    EventAck(EventAck),
287    /// Lion asks the cloud composition to remove an archived thread's Runtime checkpoints.
288    CheckpointPurge(CheckpointPurgeRequest),
289    /// Agent-cloud reports the idempotent S3 purge result.
290    CheckpointPurgeAck(CheckpointPurgeAck),
291    /// Non-durable progress leaving Runtime.
292    Observation(ObservationMessage),
293    /// One dispatch failed outside the checkpointed Runtime state machine.
294    ExecutionFailure(ExecutionFailure),
295    /// Runtime request to the App Facade.
296    AppFacadeRequest(AppFacadeRequest),
297    /// App Facade response returned to Runtime.
298    AppFacadeResponse(AppFacadeResponse),
299}
300
301/// Complete transport message.
302#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
303pub struct Envelope {
304    /// Correlation and version metadata.
305    pub metadata: Metadata,
306    /// Typed payload.
307    pub payload: Payload,
308}
309
310impl Envelope {
311    /// Strictly decodes a JSON envelope and rejects fields unknown to the Rust DTO graph.
312    ///
313    /// Transport adapters should use this entry point rather than calling `serde_json` directly.
314    ///
315    /// # Errors
316    ///
317    /// Returns `ProtocolError` for malformed JSON, unknown fields, or failed envelope validation.
318    pub fn decode_json(input: &str) -> Result<Self, ProtocolError> {
319        let input_value: serde_json::Value =
320            serde_json::from_str(input).map_err(|error| ProtocolError::InvalidJson {
321                message: error.to_string(),
322            })?;
323        let mut deserializer = serde_json::Deserializer::from_str(input);
324        let mut unknown = Vec::new();
325        let envelope: Self = serde_ignored::deserialize(&mut deserializer, |path| {
326            unknown.push(path.to_string());
327        })
328        .map_err(|error| ProtocolError::InvalidJson {
329            message: error.to_string(),
330        })?;
331        deserializer
332            .end()
333            .map_err(|error| ProtocolError::InvalidJson {
334                message: error.to_string(),
335            })?;
336        if let Some(path) = unknown.into_iter().next() {
337            return Err(ProtocolError::UnknownField { path });
338        }
339        let canonical =
340            serde_json::to_value(&envelope).map_err(|error| ProtocolError::InvalidJson {
341                message: error.to_string(),
342            })?;
343        if let Some(path) = first_extra_field(&input_value, &canonical, "$") {
344            return Err(ProtocolError::UnknownField { path });
345        }
346        envelope.validate()?;
347        Ok(envelope)
348    }
349
350    /// Validates transport-level invariants before dispatch.
351    ///
352    /// # Errors
353    ///
354    /// Returns `ProtocolError` when the major version is unsupported or a required identifier is
355    /// blank.
356    pub fn validate(&self) -> Result<(), ProtocolError> {
357        if self.metadata.version != CURRENT_VERSION {
358            return Err(ProtocolError::UnsupportedVersion {
359                actual: self.metadata.version,
360            });
361        }
362        if self.metadata.message_id.trim().is_empty() {
363            return Err(ProtocolError::MissingIdentifier("message_id"));
364        }
365        if self.metadata.correlation_id.trim().is_empty() {
366            return Err(ProtocolError::MissingIdentifier("correlation_id"));
367        }
368        validate_dispatch(&self.metadata.dispatch)?;
369        if let Some(context) = self.context() {
370            validate_context(context)?;
371        }
372        if let Payload::Command(request) = &self.payload {
373            validate_command_binding(request)?;
374        }
375        if let Payload::CheckpointPurge(request) = &self.payload {
376            validate_purge_request(request)?;
377        }
378        if let Payload::CheckpointPurgeAck(ack) = &self.payload {
379            validate_purge_ack(ack)?;
380        }
381        if let Payload::EventAck(ack) = &self.payload {
382            validate_checkpoint_reference(&ack.checkpoint)?;
383        }
384        match &self.payload {
385            Payload::Command(request) => {
386                if let Some(reference) = &request.base_checkpoint {
387                    validate_checkpoint_reference(reference)?;
388                }
389                if matches!(&request.command, AgentCommand::Cancel { .. })
390                    && (request.base_checkpoint.is_some() || request.abandoned_run_id.is_some())
391                {
392                    return Err(ProtocolError::InvalidBinding(
393                        "cancel must target live execution without checkpoint recovery metadata"
394                            .to_owned(),
395                    ));
396                }
397                if request.abandoned_run_id.is_some() && request.base_checkpoint.is_none() {
398                    return Err(ProtocolError::InvalidBinding(
399                        "an abandoned run requires a base checkpoint".to_owned(),
400                    ));
401                }
402            }
403            Payload::Event(message) => validate_checkpoint_reference(&message.checkpoint)?,
404            _ => {}
405        }
406        Ok(())
407    }
408
409    fn context(&self) -> Option<&InvocationContext> {
410        match &self.payload {
411            Payload::Command(request) => match &request.command {
412                AgentCommand::Enqueue { context, .. }
413                | AgentCommand::SubmitInteraction { context, .. }
414                | AgentCommand::Cancel { context } => Some(context),
415            },
416            Payload::Event(message) => Some(&message.event.context),
417            Payload::Observation(message) => Some(&message.context),
418            Payload::ExecutionFailure(message) => Some(&message.context),
419            Payload::AppFacadeRequest(request) => match request {
420                AppFacadeRequest::Invoke { context, .. }
421                | AppFacadeRequest::Prepare { context, .. }
422                | AppFacadeRequest::Commit { context, .. }
423                | AppFacadeRequest::Reject { context, .. } => Some(context),
424            },
425            Payload::CommandResponse(response) => match response {
426                CommandResponse::Accepted { context, .. }
427                | CommandResponse::Rejected { context, .. } => Some(context),
428            },
429            Payload::EventAck(_)
430            | Payload::CheckpointPurge(_)
431            | Payload::CheckpointPurgeAck(_)
432            | Payload::AppFacadeResponse(_) => None,
433        }
434    }
435}
436
437fn validate_purge_request(request: &CheckpointPurgeRequest) -> Result<(), ProtocolError> {
438    if request.purge_id.trim().is_empty() {
439        return Err(ProtocolError::MissingIdentifier("purge_id"));
440    }
441    if request.conversation_key.trim().is_empty() {
442        return Err(ProtocolError::MissingIdentifier("conversation_key"));
443    }
444    if request.scope.scope_id.as_str().trim().is_empty() {
445        return Err(ProtocolError::MissingIdentifier("scope_id"));
446    }
447    if request.thread_id.as_str().trim().is_empty() {
448        return Err(ProtocolError::MissingIdentifier("thread_id"));
449    }
450    Ok(())
451}
452
453fn validate_purge_ack(ack: &CheckpointPurgeAck) -> Result<(), ProtocolError> {
454    let (purge_id, conversation_key, thread_id) = match ack {
455        CheckpointPurgeAck::Deleted {
456            purge_id,
457            conversation_key,
458            thread_id,
459            ..
460        }
461        | CheckpointPurgeAck::Failed {
462            purge_id,
463            conversation_key,
464            thread_id,
465            ..
466        } => (purge_id, conversation_key, thread_id),
467    };
468    if purge_id.trim().is_empty() {
469        return Err(ProtocolError::MissingIdentifier("purge_id"));
470    }
471    if conversation_key.trim().is_empty() {
472        return Err(ProtocolError::MissingIdentifier("conversation_key"));
473    }
474    if thread_id.as_str().trim().is_empty() {
475        return Err(ProtocolError::MissingIdentifier("thread_id"));
476    }
477    if let CheckpointPurgeAck::Failed { code, .. } = ack
478        && code.trim().is_empty()
479    {
480        return Err(ProtocolError::MissingIdentifier("code"));
481    }
482    Ok(())
483}
484
485fn validate_dispatch(dispatch: &DispatchBinding) -> Result<(), ProtocolError> {
486    if dispatch.dispatch_id.trim().is_empty() {
487        return Err(ProtocolError::MissingIdentifier("dispatch_id"));
488    }
489    if dispatch.agent_session_id.trim().is_empty() {
490        return Err(ProtocolError::MissingIdentifier("agent_session_id"));
491    }
492    Ok(())
493}
494
495fn validate_checkpoint_reference(reference: &CheckpointReference) -> Result<(), ProtocolError> {
496    if reference.object_key.trim().is_empty() {
497        return Err(ProtocolError::MissingIdentifier("checkpoint_object_key"));
498    }
499    if reference.format_version == 0
500        || reference.sha256.len() != 64
501        || !reference
502            .sha256
503            .bytes()
504            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
505    {
506        return Err(ProtocolError::InvalidBinding(
507            "checkpoint reference is invalid".to_owned(),
508        ));
509    }
510    Ok(())
511}
512
513fn validate_command_binding(request: &CommandRequest) -> Result<(), ProtocolError> {
514    match (&request.command, &request.context_generation) {
515        (AgentCommand::Enqueue { .. } | AgentCommand::SubmitInteraction { .. }, Some(value))
516            if value.is_complete() => {}
517        (AgentCommand::Enqueue { .. } | AgentCommand::SubmitInteraction { .. }, _) => {
518            return Err(ProtocolError::InvalidBinding(
519                "enqueue and submit_interaction require a context generation".to_owned(),
520            ));
521        }
522        (AgentCommand::Cancel { .. }, None) => {}
523        (AgentCommand::Cancel { .. }, Some(_)) => {
524            return Err(ProtocolError::InvalidBinding(
525                "cancel must not carry a context generation".to_owned(),
526            ));
527        }
528    }
529    match (&request.command, &request.llm_plan) {
530        (AgentCommand::Enqueue { options, .. }, Some(plan)) => {
531            validate_llm_plan(plan)?;
532            if plan
533                .route_for(&options.use_case, &options.model_mode)
534                .is_none()
535            {
536                return Err(ProtocolError::InvalidBinding(
537                    "LLM plan does not contain the command's primary selector".to_owned(),
538                ));
539            }
540            Ok(())
541        }
542        (AgentCommand::Enqueue { .. }, None) => Err(ProtocolError::InvalidBinding(
543            "enqueue requires a frozen LLM plan".to_owned(),
544        )),
545        (AgentCommand::SubmitInteraction { .. }, Some(plan)) => validate_llm_plan(plan),
546        (AgentCommand::SubmitInteraction { .. }, None) => Err(ProtocolError::InvalidBinding(
547            "submit_interaction requires the run's frozen LLM plan".to_owned(),
548        )),
549        (_, Some(_)) => Err(ProtocolError::InvalidBinding(
550            "only commands that invoke the model may carry an LLM plan".to_owned(),
551        )),
552        (_, None) => Ok(()),
553    }
554}
555
556fn validate_llm_plan(plan: &LlmExecutionPlan) -> Result<(), ProtocolError> {
557    if plan.routes.is_empty() {
558        return Err(ProtocolError::InvalidBinding(
559            "LLM plan must contain at least one route".to_owned(),
560        ));
561    }
562    let mut selectors = std::collections::HashSet::new();
563    for binding in &plan.routes {
564        let route = &binding.route;
565        if binding.use_case.0.trim().is_empty()
566            || binding.model_mode.0.trim().is_empty()
567            || route.backend.trim().is_empty()
568            || route.profile.trim().is_empty()
569            || route.model.trim().is_empty()
570        {
571            return Err(ProtocolError::InvalidBinding(
572                "LLM use case, model mode, backend, profile, and model are required".to_owned(),
573            ));
574        }
575        if !selectors.insert((binding.use_case.0.as_str(), binding.model_mode.0.as_str())) {
576            return Err(ProtocolError::InvalidBinding(
577                "LLM plan contains a duplicate selector".to_owned(),
578            ));
579        }
580        if binding
581            .revision
582            .as_ref()
583            .is_some_and(|value| value.trim().is_empty())
584            || route
585                .provider_preferences
586                .iter()
587                .any(|value| value.trim().is_empty())
588            || route
589                .reasoning_effort
590                .as_ref()
591                .is_some_and(|value| value.trim().is_empty())
592            || route.temperature.is_some_and(|value| !value.is_finite())
593            || route.max_output_tokens == Some(0)
594            || route.context_window_tokens <= 20_000
595        {
596            return Err(ProtocolError::InvalidBinding(
597                "LLM route contains invalid optional settings".to_owned(),
598            ));
599        }
600    }
601    Ok(())
602}
603
604fn first_extra_field(
605    input: &serde_json::Value,
606    canonical: &serde_json::Value,
607    path: &str,
608) -> Option<String> {
609    match (input, canonical) {
610        (serde_json::Value::Object(input), serde_json::Value::Object(canonical)) => {
611            for (key, value) in input {
612                let child_path = format!("{path}.{key}");
613                let Some(expected) = canonical.get(key) else {
614                    return Some(child_path);
615                };
616                if let Some(extra) = first_extra_field(value, expected, &child_path) {
617                    return Some(extra);
618                }
619            }
620            None
621        }
622        (serde_json::Value::Array(input), serde_json::Value::Array(canonical)) => input
623            .iter()
624            .zip(canonical)
625            .enumerate()
626            .find_map(|(index, (value, expected))| {
627                first_extra_field(value, expected, &format!("{path}[{index}]"))
628            }),
629        _ => None,
630    }
631}
632
633fn validate_context(context: &InvocationContext) -> Result<(), ProtocolError> {
634    validate_scope(&context.scope)?;
635    for (name, value) in [
636        ("user_id", context.actor.user_id.as_str()),
637        ("thread_id", context.thread_id.as_str()),
638        ("run_id", context.run_id.as_str()),
639        ("operation_id", context.operation_id.as_str()),
640    ] {
641        if value.trim().is_empty() {
642            return Err(ProtocolError::MissingIdentifier(name));
643        }
644    }
645    if context
646        .actor
647        .client_id
648        .as_ref()
649        .is_some_and(|client| client.as_str().trim().is_empty())
650    {
651        return Err(ProtocolError::MissingIdentifier("client_id"));
652    }
653    Ok(())
654}
655
656fn validate_scope(scope: &Scope) -> Result<(), ProtocolError> {
657    if scope.scope_id.as_str().trim().is_empty() {
658        return Err(ProtocolError::MissingIdentifier("scope_id"));
659    }
660    if scope
661        .tenant_id
662        .as_ref()
663        .is_some_and(|tenant| tenant.as_str().trim().is_empty())
664    {
665        return Err(ProtocolError::MissingIdentifier("tenant_id"));
666    }
667    Ok(())
668}
669
670/// Invalid wire envelope.
671#[derive(Clone, Debug, Error, Eq, PartialEq)]
672pub enum ProtocolError {
673    /// JSON could not be decoded into an envelope.
674    #[error("invalid protocol JSON: {message}")]
675    InvalidJson {
676        /// Safe parser diagnostic.
677        message: String,
678    },
679    /// The payload contained a field outside the normative schema.
680    #[error("unknown protocol field: {path}")]
681    UnknownField {
682        /// Serde path to the unexpected field.
683        path: String,
684    },
685    /// The sender used a protocol major version this crate does not understand.
686    #[error("unsupported protocol version {actual}")]
687    UnsupportedVersion {
688        /// Received major version.
689        actual: u16,
690    },
691    /// A required identifier was blank.
692    #[error("missing required identifier: {0}")]
693    MissingIdentifier(&'static str),
694    /// Environment binding is absent or disagrees with its Runtime command.
695    #[error("invalid command binding: {0}")]
696    InvalidBinding(String),
697}
698
699#[cfg(test)]
700mod tests {
701    use crate::execution::{
702        AccessMode, Actor, AgentEvent, ClientId, ContentPart, EventId, Message, MessageRole,
703        ModelMode, OperationId, RunId, RunOptions, Scope, ScopeId, TenantId, ThreadId, UseCase,
704        UserId,
705    };
706
707    use super::*;
708
709    #[test]
710    fn rejects_unknown_protocol_version() {
711        let envelope = Envelope {
712            metadata: Metadata {
713                version: CURRENT_VERSION + 1,
714                message_id: "message-1".to_owned(),
715                correlation_id: "correlation-1".to_owned(),
716                causation_id: None,
717                sent_at_unix_ms: 0,
718                dispatch: dispatch(),
719            },
720            payload: Payload::Event(EventMessage {
721                event: DurableEvent {
722                    id: EventId::from("event-1"),
723                    sequence: 1,
724                    context: InvocationContext {
725                        scope: Scope {
726                            tenant_id: None,
727                            scope_id: ScopeId::from("scope-1"),
728                        },
729                        actor: Actor {
730                            user_id: UserId::from("user-1"),
731                            client_id: Some(ClientId::from("client-1")),
732                        },
733                        surface_id: ConversationSurface::client_personal("user-1").unwrap(),
734                        thread_id: ThreadId::from("thread-1"),
735                        run_id: RunId::from("run-1"),
736                        operation_id: OperationId::from("operation-1"),
737                        deadline_unix_ms: None,
738                        traceparent: None,
739                    },
740                    event: AgentEvent::Started,
741                },
742                checkpoint: checkpoint(),
743            }),
744        };
745
746        assert_eq!(
747            envelope.validate(),
748            Err(ProtocolError::UnsupportedVersion {
749                actual: CURRENT_VERSION + 1
750            })
751        );
752    }
753
754    #[test]
755    fn command_fixture_round_trips_without_shape_drift() {
756        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
757        let envelope = Envelope::decode_json(fixture).expect("valid golden fixture");
758
759        let expected = Envelope {
760            metadata: Metadata {
761                version: CURRENT_VERSION,
762                message_id: "message-1".to_owned(),
763                correlation_id: "correlation-1".to_owned(),
764                causation_id: None,
765                sent_at_unix_ms: 1_700_000_000_000,
766                dispatch: dispatch(),
767            },
768            payload: Payload::Command(Box::new(CommandRequest {
769                command: AgentCommand::Enqueue {
770                    context: InvocationContext {
771                        scope: Scope {
772                            tenant_id: Some(TenantId::from("tenant-1")),
773                            scope_id: ScopeId::from("scope-1"),
774                        },
775                        actor: Actor {
776                            user_id: UserId::from("user-1"),
777                            client_id: Some(ClientId::from("client-1")),
778                        },
779                        surface_id: ConversationSurface::client_personal("user-1").unwrap(),
780                        thread_id: ThreadId::from("thread-1"),
781                        run_id: RunId::from("run-1"),
782                        operation_id: OperationId::from("operation-1"),
783                        deadline_unix_ms: None,
784                        traceparent: None,
785                    },
786                    message: Message {
787                        continuation: None,
788                        role: MessageRole::User,
789                        content: vec![ContentPart::Text {
790                            text: "hello".to_owned(),
791                        }],
792                    },
793                    options: RunOptions {
794                        use_case: UseCase("chat".to_owned()),
795                        model_mode: ModelMode("auto".to_owned()),
796                        access_mode: AccessMode::Interactive,
797                        allow_tools: true,
798                        max_steps: 8,
799                        credit_budget: None,
800                        debug: false,
801                    },
802                },
803                context_generation: Some(ContextGeneration {
804                    user_scope: "scope-1".to_owned(),
805                    identity: "identity-1".to_owned(),
806                    memory: "memory-1".to_owned(),
807                    integration_guide: "guide-1".to_owned(),
808                    installed_integrations: "installed-1".to_owned(),
809                    scope_integrations: "scope-integrations-1".to_owned(),
810                }),
811                llm_plan: Some(LlmExecutionPlan {
812                    routes: [
813                        "chat",
814                        "agent_info_merge",
815                        "automation_judge",
816                        "automation_diagnose",
817                        "app_guide",
818                        "workflow.action_operate",
819                        "workflow.automation_generate",
820                        "workflow.description_generate",
821                    ]
822                    .into_iter()
823                    .map(|use_case| LlmRouteBinding {
824                        use_case: UseCase(use_case.to_owned()),
825                        model_mode: ModelMode("auto".to_owned()),
826                        revision: Some("1700000000000".to_owned()),
827                        route: LlmRoute {
828                            backend: "openrouter".to_owned(),
829                            profile: "cloud_openrouter".to_owned(),
830                            model: "openai/gpt-5.4-mini".to_owned(),
831                            provider_preferences: vec!["OpenAI".to_owned()],
832                            temperature: Some(0.5),
833                            max_output_tokens: Some(8192),
834                            cache_control: Some(true),
835                            reasoning_effort: None,
836                            context_window_tokens: 128_000,
837                        },
838                    })
839                    .collect(),
840                }),
841                base_checkpoint: None,
842                abandoned_run_id: None,
843            })),
844        };
845        assert_eq!(envelope, expected);
846        assert_eq!(
847            serde_json::to_value(envelope).expect("serializes"),
848            serde_json::from_str::<serde_json::Value>(fixture).expect("fixture JSON")
849        );
850    }
851
852    fn dispatch() -> DispatchBinding {
853        DispatchBinding {
854            dispatch_id: "dispatch-1".to_owned(),
855            agent_session_id: "agent-session-1".to_owned(),
856        }
857    }
858
859    fn checkpoint() -> CheckpointReference {
860        CheckpointReference {
861            object_key: "checkpoints/abc.json".to_owned(),
862            sha256: "a".repeat(64),
863            format_version: 1,
864        }
865    }
866
867    #[test]
868    fn every_v1_golden_fixture_round_trips() {
869        let schema: serde_json::Value = serde_json::from_str(include_str!(
870            "../../../schema/execution/envelope.v1.schema.json"
871        ))
872        .expect("valid JSON Schema document");
873        let validator = jsonschema::validator_for(&schema).expect("valid JSON Schema semantics");
874        for fixture in [
875            include_str!("../../../fixtures/execution/enqueue.v1.json"),
876            include_str!("../../../fixtures/execution/completed-event.v1.json"),
877            include_str!("../../../fixtures/execution/interaction-required-event.v1.json"),
878            include_str!("../../../fixtures/execution/failed-event.v1.json"),
879            include_str!("../../../fixtures/execution/app-facade-prepare.v1.json"),
880            include_str!("../../../fixtures/execution/observation.v1.json"),
881            include_str!("../../../fixtures/execution/admission.v1.json"),
882            include_str!("../../../fixtures/execution/app-facade-prepared.v1.json"),
883            include_str!("../../../fixtures/execution/app-facade-user-action.v1.json"),
884            include_str!("../../../fixtures/execution/event-ack.v1.json"),
885            include_str!("../../../fixtures/execution/checkpoint-purge.v1.json"),
886            include_str!("../../../fixtures/execution/checkpoint-purge-ack.v1.json"),
887        ] {
888            let json: serde_json::Value =
889                serde_json::from_str(fixture).expect("valid golden fixture JSON");
890            validator
891                .validate(&json)
892                .expect("golden fixture matches the normative schema");
893            let envelope = Envelope::decode_json(fixture).expect("fixture matches Rust DTOs");
894            assert_eq!(serde_json::to_value(envelope).expect("serializes"), json);
895        }
896    }
897
898    #[test]
899    fn rejects_blank_scoped_identifiers() {
900        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
901        let mut envelope: Envelope = serde_json::from_str(fixture).expect("valid golden fixture");
902        let Payload::Command(request) = &mut envelope.payload else {
903            panic!("enqueue fixture changed family");
904        };
905        let CommandRequest {
906            command: AgentCommand::Enqueue { context, .. },
907            ..
908        } = request.as_mut()
909        else {
910            panic!("enqueue fixture changed family");
911        };
912        context.scope.scope_id = ScopeId::from(" ");
913        assert_eq!(
914            envelope.validate(),
915            Err(ProtocolError::MissingIdentifier("scope_id"))
916        );
917    }
918
919    #[test]
920    fn strict_decoder_rejects_unknown_nested_fields() {
921        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
922        let input = fixture.replacen(
923            "\"sent_at_unix_ms\": 1700000000000",
924            "\"sent_at_unix_ms\": 1700000000000, \"unexpected\": true",
925            1,
926        );
927        assert!(matches!(
928            Envelope::decode_json(&input),
929            Err(ProtocolError::UnknownField { .. })
930        ));
931
932        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
933        input["payload"]["payload"]["options"]["unexpected"] = serde_json::Value::Bool(true);
934        assert!(matches!(
935            Envelope::decode_json(&input.to_string()),
936            Err(ProtocolError::UnknownField { .. })
937        ));
938    }
939
940    #[test]
941    fn abandoned_run_requires_a_base_checkpoint() {
942        let mut value: serde_json::Value =
943            serde_json::from_str(include_str!("../../../fixtures/execution/enqueue.v1.json"))
944                .unwrap();
945        value["payload"]["payload"]["abandoned_run_id"] =
946            serde_json::Value::String("run-abandoned".to_owned());
947
948        assert!(matches!(
949            Envelope::decode_json(&value.to_string()),
950            Err(ProtocolError::InvalidBinding(_))
951        ));
952    }
953}
conversation_api/execution/wire/mod.rs

conversation_api/execution/wire/
mod.rs
