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/// Model facts captured by the deployment with the desired route parameters.
184/// This snapshot belongs to admission, not to the provider-neutral LLM request.
185#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
186#[serde(deny_unknown_fields)]
187pub struct LlmModelSnapshot {
188    pub capabilities: llm_api::ModelCapabilities,
189    pub generation_support: LlmGenerationSupport,
190}
191
192/// Catalog parameter support travels with the plan so a run never resolves against a
193/// catalog that changed after admission. Resolution itself belongs to `llm_api::resolve`.
194pub use llm_api::GenerationSupport as LlmGenerationSupport;
195
196/// One concrete cloud model route frozen by the deployment before command admission.
197/// The generation fields are the deployment's desired preferences, not resolved values:
198/// the runtime resolves them once against `model_snapshot` and the backend protocol.
199#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
200pub struct LlmRoute {
201    pub backend: String,
202    pub profile: String,
203    pub model: String,
204    pub provider_preferences: Vec<String>,
205    pub temperature: Option<f64>,
206    pub cache_control: Option<bool>,
207    pub reasoning_effort: Option<String>,
208    #[serde(default, skip_serializing_if = "Option::is_none")]
209    pub thinking: Option<bool>,
210    #[serde(default, skip_serializing_if = "Option::is_none")]
211    pub fast_mode: Option<bool>,
212    /// Total context window of the exact frozen model route.
213    pub context_window_tokens: u32,
214    /// Required immutable facts; old plans must not silently resolve against a new catalog.
215    pub model_snapshot: LlmModelSnapshot,
216}
217
218/// One logical selector and its concrete deployment route.
219#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
220pub struct LlmRouteBinding {
221    pub use_case: UseCase,
222    pub model_mode: ModelMode,
223    pub revision: Option<String>,
224    pub route: LlmRoute,
225}
226
227/// Complete deployment-resolved LLM route set bound immutably to one admitted run.
228#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
229pub struct LlmExecutionPlan {
230    pub routes: Vec<LlmRouteBinding>,
231}
232
233impl LlmExecutionPlan {
234    /// Selects the exact frozen route for one provider-neutral Runtime request.
235    #[must_use]
236    pub fn route_for(
237        &self,
238        use_case: &UseCase,
239        model_mode: &ModelMode,
240    ) -> Option<&LlmRouteBinding> {
241        self.routes
242            .iter()
243            .find(|item| item.use_case == *use_case && item.model_mode == *model_mode)
244    }
245}
246
247/// Transport admission request. Environment bindings are consumed by the composition root and do
248/// not enter the provider-neutral Runtime command.
249#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
250pub struct CommandRequest {
251    pub command: AgentCommand,
252    /// Effective prompt-context generation selected by Lion for this dispatch.
253    pub context_generation: Option<ContextGeneration>,
254    pub llm_plan: Option<LlmExecutionPlan>,
255    /// Stable Runtime checkpoint selected by Lion, absent for a new thread.
256    pub base_checkpoint: Option<CheckpointReference>,
257    /// Attempt already terminalized by Lion and settled silently before new work starts.
258    pub abandoned_run_id: Option<RunId>,
259}
260
261/// Checkpoint-coupled Runtime event. Lion commits the generation before projecting the event.
262#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
263pub struct EventMessage {
264    pub event: DurableEvent,
265    pub checkpoint: CheckpointReference,
266}
267
268/// Result of an App Facade request transported back to Runtime.
269#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
270#[serde(tag = "type", rename_all = "snake_case")]
271pub enum AppFacadeResponse {
272    /// Read-only or committed operation result.
273    Result {
274        /// Structured App Facade result.
275        result: FacadeResult,
276    },
277    /// Prepared operation awaiting a commit or rejection.
278    Prepared {
279        /// Prepared action.
280        action: PreparedAction,
281    },
282    /// Rejection completed successfully.
283    Rejected,
284    /// Stable error safe to expose across the transport.
285    Failed {
286        /// Machine-readable category.
287        code: String,
288        /// Safe diagnostic message.
289        message: String,
290        /// Optional retry delay.
291        retry_after_ms: Option<u64>,
292    },
293}
294
295/// Payload families carried by the versioned envelope.
296#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
297#[serde(tag = "family", content = "payload", rename_all = "snake_case")]
298pub enum Payload {
299    /// Command entering Runtime.
300    Command(Box<CommandRequest>),
301    /// Immediate command-admission response.
302    CommandResponse(CommandResponse),
303    /// Event leaving Runtime.
304    Event(EventMessage),
305    /// Durable receiver acknowledgement used to clear the Runtime outbox.
306    EventAck(EventAck),
307    /// Lion asks the cloud composition to remove an archived thread's Runtime checkpoints.
308    CheckpointPurge(CheckpointPurgeRequest),
309    /// Agent-cloud reports the idempotent S3 purge result.
310    CheckpointPurgeAck(CheckpointPurgeAck),
311    /// Non-durable progress leaving Runtime.
312    Observation(ObservationMessage),
313    /// One dispatch failed outside the checkpointed Runtime state machine.
314    ExecutionFailure(ExecutionFailure),
315    /// Runtime request to the App Facade.
316    AppFacadeRequest(AppFacadeRequest),
317    /// App Facade response returned to Runtime.
318    AppFacadeResponse(AppFacadeResponse),
319}
320
321/// Complete transport message.
322#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
323pub struct Envelope {
324    /// Correlation and version metadata.
325    pub metadata: Metadata,
326    /// Typed payload.
327    pub payload: Payload,
328}
329
330impl Envelope {
331    /// Decodes a JSON envelope against the protocol types.
332    ///
333    /// Transport adapters should use this entry point rather than calling `serde_json` directly.
334    /// Unknown keys are rejected. Canonical emit omits skip-optional `None`; an explicit `null` on
335    /// those keys is accepted as absent so accept follows the types instead of a serialize roundtrip.
336    ///
337    /// # Errors
338    ///
339    /// Returns `ProtocolError` for malformed JSON, unknown fields, or failed envelope validation.
340    pub fn decode_json(input: &str) -> Result<Self, ProtocolError> {
341        let mut deserializer = serde_json::Deserializer::from_str(input);
342        let mut unknown = Vec::new();
343        let envelope: Self = serde_ignored::deserialize(&mut deserializer, |path| {
344            unknown.push(path.to_string());
345        })
346        .map_err(|error| ProtocolError::InvalidJson {
347            message: error.to_string(),
348        })?;
349        deserializer
350            .end()
351            .map_err(|error| ProtocolError::InvalidJson {
352                message: error.to_string(),
353            })?;
354        if let Some(path) = unknown.into_iter().next() {
355            return Err(ProtocolError::UnknownField { path });
356        }
357        envelope.validate()?;
358        Ok(envelope)
359    }
360
361    /// Validates transport-level invariants before dispatch.
362    ///
363    /// # Errors
364    ///
365    /// Returns `ProtocolError` when the major version is unsupported or a required identifier is
366    /// blank.
367    pub fn validate(&self) -> Result<(), ProtocolError> {
368        if self.metadata.version != CURRENT_VERSION {
369            return Err(ProtocolError::UnsupportedVersion {
370                actual: self.metadata.version,
371            });
372        }
373        if self.metadata.message_id.trim().is_empty() {
374            return Err(ProtocolError::MissingIdentifier("message_id"));
375        }
376        if self.metadata.correlation_id.trim().is_empty() {
377            return Err(ProtocolError::MissingIdentifier("correlation_id"));
378        }
379        validate_dispatch(&self.metadata.dispatch)?;
380        if let Some(context) = self.context() {
381            validate_context(context)?;
382        }
383        if let Payload::Command(request) = &self.payload {
384            validate_command_binding(request)?;
385        }
386        if let Payload::CheckpointPurge(request) = &self.payload {
387            validate_purge_request(request)?;
388        }
389        if let Payload::CheckpointPurgeAck(ack) = &self.payload {
390            validate_purge_ack(ack)?;
391        }
392        if let Payload::EventAck(ack) = &self.payload {
393            validate_checkpoint_reference(&ack.checkpoint)?;
394        }
395        match &self.payload {
396            Payload::Command(request) => {
397                if let Some(reference) = &request.base_checkpoint {
398                    validate_checkpoint_reference(reference)?;
399                }
400                if matches!(&request.command, AgentCommand::Cancel { .. })
401                    && (request.base_checkpoint.is_some() || request.abandoned_run_id.is_some())
402                {
403                    return Err(ProtocolError::InvalidBinding(
404                        "cancel must target live execution without checkpoint recovery metadata"
405                            .to_owned(),
406                    ));
407                }
408                if request.abandoned_run_id.is_some() && request.base_checkpoint.is_none() {
409                    return Err(ProtocolError::InvalidBinding(
410                        "an abandoned run requires a base checkpoint".to_owned(),
411                    ));
412                }
413            }
414            Payload::Event(message) => validate_checkpoint_reference(&message.checkpoint)?,
415            _ => {}
416        }
417        Ok(())
418    }
419
420    fn context(&self) -> Option<&InvocationContext> {
421        match &self.payload {
422            Payload::Command(request) => match &request.command {
423                AgentCommand::Enqueue { context, .. }
424                | AgentCommand::SubmitInteraction { context, .. }
425                | AgentCommand::Cancel { context } => Some(context),
426            },
427            Payload::Event(message) => Some(&message.event.context),
428            Payload::Observation(message) => Some(&message.context),
429            Payload::ExecutionFailure(message) => Some(&message.context),
430            Payload::AppFacadeRequest(request) => match request {
431                AppFacadeRequest::Invoke { context, .. }
432                | AppFacadeRequest::Prepare { context, .. }
433                | AppFacadeRequest::Commit { context, .. }
434                | AppFacadeRequest::Reject { context, .. } => Some(context),
435            },
436            Payload::CommandResponse(response) => match response {
437                CommandResponse::Accepted { context, .. }
438                | CommandResponse::Rejected { context, .. } => Some(context),
439            },
440            Payload::EventAck(_)
441            | Payload::CheckpointPurge(_)
442            | Payload::CheckpointPurgeAck(_)
443            | Payload::AppFacadeResponse(_) => None,
444        }
445    }
446}
447
448fn validate_purge_request(request: &CheckpointPurgeRequest) -> Result<(), ProtocolError> {
449    if request.purge_id.trim().is_empty() {
450        return Err(ProtocolError::MissingIdentifier("purge_id"));
451    }
452    if request.conversation_key.trim().is_empty() {
453        return Err(ProtocolError::MissingIdentifier("conversation_key"));
454    }
455    if request.scope.scope_id.as_str().trim().is_empty() {
456        return Err(ProtocolError::MissingIdentifier("scope_id"));
457    }
458    if request.thread_id.as_str().trim().is_empty() {
459        return Err(ProtocolError::MissingIdentifier("thread_id"));
460    }
461    Ok(())
462}
463
464fn validate_purge_ack(ack: &CheckpointPurgeAck) -> Result<(), ProtocolError> {
465    let (purge_id, conversation_key, thread_id) = match ack {
466        CheckpointPurgeAck::Deleted {
467            purge_id,
468            conversation_key,
469            thread_id,
470            ..
471        }
472        | CheckpointPurgeAck::Failed {
473            purge_id,
474            conversation_key,
475            thread_id,
476            ..
477        } => (purge_id, conversation_key, thread_id),
478    };
479    if purge_id.trim().is_empty() {
480        return Err(ProtocolError::MissingIdentifier("purge_id"));
481    }
482    if conversation_key.trim().is_empty() {
483        return Err(ProtocolError::MissingIdentifier("conversation_key"));
484    }
485    if thread_id.as_str().trim().is_empty() {
486        return Err(ProtocolError::MissingIdentifier("thread_id"));
487    }
488    if let CheckpointPurgeAck::Failed { code, .. } = ack
489        && code.trim().is_empty()
490    {
491        return Err(ProtocolError::MissingIdentifier("code"));
492    }
493    Ok(())
494}
495
496fn validate_dispatch(dispatch: &DispatchBinding) -> Result<(), ProtocolError> {
497    if dispatch.dispatch_id.trim().is_empty() {
498        return Err(ProtocolError::MissingIdentifier("dispatch_id"));
499    }
500    if dispatch.agent_session_id.trim().is_empty() {
501        return Err(ProtocolError::MissingIdentifier("agent_session_id"));
502    }
503    Ok(())
504}
505
506fn validate_checkpoint_reference(reference: &CheckpointReference) -> Result<(), ProtocolError> {
507    if reference.object_key.trim().is_empty() {
508        return Err(ProtocolError::MissingIdentifier("checkpoint_object_key"));
509    }
510    if reference.format_version == 0
511        || reference.sha256.len() != 64
512        || !reference
513            .sha256
514            .bytes()
515            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
516    {
517        return Err(ProtocolError::InvalidBinding(
518            "checkpoint reference is invalid".to_owned(),
519        ));
520    }
521    Ok(())
522}
523
524fn validate_command_binding(request: &CommandRequest) -> Result<(), ProtocolError> {
525    match (&request.command, &request.context_generation) {
526        (AgentCommand::Enqueue { .. } | AgentCommand::SubmitInteraction { .. }, Some(value))
527            if value.is_complete() => {}
528        (AgentCommand::Enqueue { .. } | AgentCommand::SubmitInteraction { .. }, _) => {
529            return Err(ProtocolError::InvalidBinding(
530                "enqueue and submit_interaction require a context generation".to_owned(),
531            ));
532        }
533        (AgentCommand::Cancel { .. }, None) => {}
534        (AgentCommand::Cancel { .. }, Some(_)) => {
535            return Err(ProtocolError::InvalidBinding(
536                "cancel must not carry a context generation".to_owned(),
537            ));
538        }
539    }
540    match (&request.command, &request.llm_plan) {
541        (AgentCommand::Enqueue { options, .. }, Some(plan)) => {
542            validate_llm_plan(plan)?;
543            if plan
544                .route_for(&options.use_case, &options.model_mode)
545                .is_none()
546            {
547                return Err(ProtocolError::InvalidBinding(
548                    "LLM plan does not contain the command's primary selector".to_owned(),
549                ));
550            }
551            Ok(())
552        }
553        (AgentCommand::Enqueue { .. }, None) => Err(ProtocolError::InvalidBinding(
554            "enqueue requires a frozen LLM plan".to_owned(),
555        )),
556        (AgentCommand::SubmitInteraction { .. }, Some(plan)) => validate_llm_plan(plan),
557        (AgentCommand::SubmitInteraction { .. }, None) => Err(ProtocolError::InvalidBinding(
558            "submit_interaction requires the run's frozen LLM plan".to_owned(),
559        )),
560        (_, Some(_)) => Err(ProtocolError::InvalidBinding(
561            "only commands that invoke the model may carry an LLM plan".to_owned(),
562        )),
563        (_, None) => Ok(()),
564    }
565}
566
567fn validate_llm_plan(plan: &LlmExecutionPlan) -> Result<(), ProtocolError> {
568    if plan.routes.is_empty() {
569        return Err(ProtocolError::InvalidBinding(
570            "LLM plan must contain at least one route".to_owned(),
571        ));
572    }
573    let mut selectors = std::collections::HashSet::new();
574    for binding in &plan.routes {
575        let route = &binding.route;
576        route
577            .model_snapshot
578            .generation_support
579            .validate()
580            .map_err(|error| ProtocolError::InvalidBinding(error.into()))?;
581        llm_api::GenerationParameters {
582            temperature: route.temperature,
583            reasoning_effort: route.reasoning_effort.clone(),
584            thinking: route.thinking,
585            fast_mode: route.fast_mode,
586        }
587        .validate()
588        .map_err(|error| ProtocolError::InvalidBinding(error.into()))?;
589        if binding.use_case.0.trim().is_empty()
590            || binding.model_mode.0.trim().is_empty()
591            || route.backend.trim().is_empty()
592            || route.profile.trim().is_empty()
593            || route.model.trim().is_empty()
594        {
595            return Err(ProtocolError::InvalidBinding(
596                "LLM use case, model mode, backend, profile, and model are required".to_owned(),
597            ));
598        }
599        if !selectors.insert((binding.use_case.0.as_str(), binding.model_mode.0.as_str())) {
600            return Err(ProtocolError::InvalidBinding(
601                "LLM plan contains a duplicate selector".to_owned(),
602            ));
603        }
604        if binding
605            .revision
606            .as_ref()
607            .is_some_and(|value| value.trim().is_empty())
608            || route
609                .provider_preferences
610                .iter()
611                .any(|value| value.trim().is_empty())
612            || route
613                .reasoning_effort
614                .as_ref()
615                .is_some_and(|value| value.trim().is_empty())
616            || route.temperature.is_some_and(|value| !value.is_finite())
617            || route.context_window_tokens <= 20_000
618        {
619            return Err(ProtocolError::InvalidBinding(
620                "LLM route contains invalid optional settings".to_owned(),
621            ));
622        }
623    }
624    Ok(())
625}
626
627fn validate_context(context: &InvocationContext) -> Result<(), ProtocolError> {
628    validate_scope(&context.scope)?;
629    for (name, value) in [
630        ("user_id", context.actor.user_id.as_str()),
631        ("thread_id", context.thread_id.as_str()),
632        ("run_id", context.run_id.as_str()),
633        ("operation_id", context.operation_id.as_str()),
634    ] {
635        if value.trim().is_empty() {
636            return Err(ProtocolError::MissingIdentifier(name));
637        }
638    }
639    if context
640        .actor
641        .client_id
642        .as_ref()
643        .is_some_and(|client| client.as_str().trim().is_empty())
644    {
645        return Err(ProtocolError::MissingIdentifier("client_id"));
646    }
647    Ok(())
648}
649
650fn validate_scope(scope: &Scope) -> Result<(), ProtocolError> {
651    if scope.scope_id.as_str().trim().is_empty() {
652        return Err(ProtocolError::MissingIdentifier("scope_id"));
653    }
654    if scope
655        .tenant_id
656        .as_ref()
657        .is_some_and(|tenant| tenant.as_str().trim().is_empty())
658    {
659        return Err(ProtocolError::MissingIdentifier("tenant_id"));
660    }
661    Ok(())
662}
663
664/// Invalid wire envelope.
665#[derive(Clone, Debug, Error, Eq, PartialEq)]
666pub enum ProtocolError {
667    /// JSON could not be decoded into an envelope.
668    #[error("invalid protocol JSON: {message}")]
669    InvalidJson {
670        /// Safe parser diagnostic.
671        message: String,
672    },
673    /// The payload contained a field outside the normative schema.
674    #[error("unknown protocol field: {path}")]
675    UnknownField {
676        /// Serde path to the unexpected field.
677        path: String,
678    },
679    /// The sender used a protocol major version this crate does not understand.
680    #[error("unsupported protocol version {actual}")]
681    UnsupportedVersion {
682        /// Received major version.
683        actual: u16,
684    },
685    /// A required identifier was blank.
686    #[error("missing required identifier: {0}")]
687    MissingIdentifier(&'static str),
688    /// Environment binding is absent or disagrees with its Runtime command.
689    #[error("invalid command binding: {0}")]
690    InvalidBinding(String),
691}
692
693#[cfg(test)]
694mod tests {
695    use crate::execution::{
696        AccessMode, Actor, AgentEvent, ClientId, ContentPart, EventId, Message, MessageRole,
697        ModelMode, OperationId, RunId, RunOptions, Scope, ScopeId, TenantId, ThreadId, UseCase,
698        UserId,
699    };
700
701    use super::*;
702
703    #[test]
704    fn rejects_unknown_protocol_version() {
705        let envelope = Envelope {
706            metadata: Metadata {
707                version: CURRENT_VERSION + 1,
708                message_id: "message-1".to_owned(),
709                correlation_id: "correlation-1".to_owned(),
710                causation_id: None,
711                sent_at_unix_ms: 0,
712                dispatch: dispatch(),
713            },
714            payload: Payload::Event(EventMessage {
715                event: DurableEvent {
716                    id: EventId::from("event-1"),
717                    sequence: 1,
718                    context: InvocationContext {
719                        scope: Scope {
720                            tenant_id: None,
721                            scope_id: ScopeId::from("scope-1"),
722                        },
723                        actor: Actor {
724                            user_id: UserId::from("user-1"),
725                            client_id: Some(ClientId::from("client-1")),
726                        },
727                        surface_id: ConversationSurface::client_personal("user-1").unwrap(),
728                        thread_id: ThreadId::from("thread-1"),
729                        run_id: RunId::from("run-1"),
730                        operation_id: OperationId::from("operation-1"),
731                        deadline_unix_ms: None,
732                        traceparent: None,
733                    },
734                    event: AgentEvent::Started,
735                },
736                checkpoint: checkpoint(),
737            }),
738        };
739
740        assert_eq!(
741            envelope.validate(),
742            Err(ProtocolError::UnsupportedVersion {
743                actual: CURRENT_VERSION + 1
744            })
745        );
746    }
747
748    #[test]
749    fn command_fixture_round_trips_without_shape_drift() {
750        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
751        let envelope = Envelope::decode_json(fixture).expect("valid golden fixture");
752
753        let expected = Envelope {
754            metadata: Metadata {
755                version: CURRENT_VERSION,
756                message_id: "message-1".to_owned(),
757                correlation_id: "correlation-1".to_owned(),
758                causation_id: None,
759                sent_at_unix_ms: 1_700_000_000_000,
760                dispatch: dispatch(),
761            },
762            payload: Payload::Command(Box::new(CommandRequest {
763                command: AgentCommand::Enqueue {
764                    context: InvocationContext {
765                        scope: Scope {
766                            tenant_id: Some(TenantId::from("tenant-1")),
767                            scope_id: ScopeId::from("scope-1"),
768                        },
769                        actor: Actor {
770                            user_id: UserId::from("user-1"),
771                            client_id: Some(ClientId::from("client-1")),
772                        },
773                        surface_id: ConversationSurface::client_personal("user-1").unwrap(),
774                        thread_id: ThreadId::from("thread-1"),
775                        run_id: RunId::from("run-1"),
776                        operation_id: OperationId::from("operation-1"),
777                        deadline_unix_ms: None,
778                        traceparent: None,
779                    },
780                    message: Message {
781                        continuation: None,
782                        role: MessageRole::User,
783                        content: vec![ContentPart::Text {
784                            text: "hello".to_owned(),
785                        }],
786                    },
787                    options: RunOptions {
788                        use_case: UseCase("chat".to_owned()),
789                        model_mode: ModelMode("auto".to_owned()),
790                        access_mode: AccessMode::Interactive,
791                        allow_tools: true,
792                        max_steps: 8,
793                        credit_budget: None,
794                        debug: false,
795                    },
796                },
797                context_generation: Some(ContextGeneration {
798                    user_scope: "scope-1".to_owned(),
799                    identity: "identity-1".to_owned(),
800                    memory: "memory-1".to_owned(),
801                    integration_guide: "guide-1".to_owned(),
802                    installed_integrations: "installed-1".to_owned(),
803                    scope_integrations: "scope-integrations-1".to_owned(),
804                }),
805                llm_plan: Some(LlmExecutionPlan {
806                    routes: [
807                        "chat",
808                        "agent_info_merge",
809                        "automation_judge",
810                        "automation_diagnose",
811                        "app_guide",
812                        "workflow.action_operate",
813                        "workflow.automation_generate",
814                        "workflow.description_generate",
815                    ]
816                    .into_iter()
817                    .map(|use_case| LlmRouteBinding {
818                        use_case: UseCase(use_case.to_owned()),
819                        model_mode: ModelMode("auto".to_owned()),
820                        revision: Some("1700000000000".to_owned()),
821                        route: LlmRoute {
822                            backend: "openrouter".to_owned(),
823                            profile: "cloud_openrouter".to_owned(),
824                            model: "openai/gpt-5.4-mini".to_owned(),
825                            provider_preferences: vec!["OpenAI".to_owned()],
826                            temperature: Some(0.5),
827                            cache_control: Some(true),
828                            reasoning_effort: None,
829                            thinking: None,
830                            fast_mode: None,
831                            context_window_tokens: 128_000,
832                            model_snapshot: LlmModelSnapshot {
833                                capabilities: llm_api::ModelCapabilities {
834                                    input: vec!["image".to_owned()],
835                                    tool_calling: true,
836                                    structured_output: true,
837                                },
838                                generation_support: LlmGenerationSupport {
839                                    temperature: Some(true),
840                                    max_tokens: Some(true),
841                                    thinking: Some(true),
842                                    reasoning_efforts: Some(vec!["high".into()]),
843                                    temperature_with_reasoning: Some(true),
844                                    ..LlmGenerationSupport::default()
845                                },
846                            },
847                        },
848                    })
849                    .collect(),
850                }),
851                base_checkpoint: None,
852                abandoned_run_id: None,
853            })),
854        };
855        assert_eq!(envelope, expected);
856        assert_eq!(
857            serde_json::to_value(envelope).expect("serializes"),
858            serde_json::from_str::<serde_json::Value>(fixture).expect("fixture JSON")
859        );
860    }
861
862    #[test]
863    fn snapshot_is_required_and_support_metadata_is_strict() {
864        let mut fixture: serde_json::Value = serde_json::from_str(ENQUEUE_FIXTURE).unwrap();
865        fixture["payload"]["payload"]["llm_plan"]["routes"][0]["route"]
866            .as_object_mut()
867            .unwrap()
868            .remove("model_snapshot");
869        assert!(Envelope::decode_json(&fixture.to_string()).is_err());
870        let mut envelope = Envelope::decode_json(ENQUEUE_FIXTURE).unwrap();
871        let Payload::Command(command) = &mut envelope.payload else {
872            panic!("command");
873        };
874        command.llm_plan.as_mut().unwrap().routes[0]
875            .route
876            .model_snapshot
877            .generation_support
878            .max_output_tokens = Some(0);
879        assert!(envelope.validate().is_err());
880    }
881
882    #[test]
883    fn generation_support_travels_in_catalog_spelling() {
884        let support = LlmGenerationSupport {
885            temperature: Some(true),
886            max_tokens: Some(true),
887            thinking: Some(true),
888            reasoning_efforts: Some(vec!["low".into(), "high".into()]),
889            reasoning_effort_default: Some("high".into()),
890            temperature_with_reasoning: Some(false),
891            temperature_max: Some(2.0),
892            max_output_tokens: Some(8_192),
893            fast_mode: Some(true),
894        };
895        let encoded = serde_json::to_value(&support).unwrap();
896        assert_eq!(
897            encoded["reasoningEfforts"],
898            serde_json::json!(["low", "high"])
899        );
900        assert_eq!(
901            encoded["temperatureWithReasoning"],
902            serde_json::json!(false)
903        );
904        assert_eq!(
905            serde_json::from_value::<LlmGenerationSupport>(encoded).unwrap(),
906            support
907        );
908        let ignored = serde_json::from_value::<LlmGenerationSupport>(serde_json::json!({
909            "temperature": true, "maxTokens": true, "reasoningEfforts": ["low"],
910            "temperatureWithReasoning": null, "temperatureMax": null, "maxOutputTokens": null,
911            "reasoningIntensity": "high"
912        }))
913        .unwrap();
914        assert_eq!(ignored.temperature, Some(true));
915        assert_eq!(
916            ignored.reasoning_efforts.as_deref(),
917            Some(["low".to_string()].as_slice())
918        );
919    }
920
921    fn dispatch() -> DispatchBinding {
922        DispatchBinding {
923            dispatch_id: "dispatch-1".to_owned(),
924            agent_session_id: "agent-session-1".to_owned(),
925        }
926    }
927
928    fn checkpoint() -> CheckpointReference {
929        CheckpointReference {
930            object_key: "checkpoints/abc.json".to_owned(),
931            sha256: "a".repeat(64),
932            format_version: 1,
933        }
934    }
935
936    #[test]
937    fn every_v1_golden_fixture_round_trips() {
938        let schema: serde_json::Value = serde_json::from_str(include_str!(
939            "../../../schema/execution/envelope.v1.schema.json"
940        ))
941        .expect("valid JSON Schema document");
942        let validator = jsonschema::validator_for(&schema).expect("valid JSON Schema semantics");
943        for fixture in [
944            include_str!("../../../fixtures/execution/enqueue.v1.json"),
945            include_str!("../../../fixtures/execution/completed-event.v1.json"),
946            include_str!("../../../fixtures/execution/interaction-required-event.v1.json"),
947            include_str!("../../../fixtures/execution/failed-event.v1.json"),
948            include_str!("../../../fixtures/execution/app-facade-prepare.v1.json"),
949            include_str!("../../../fixtures/execution/observation.v1.json"),
950            include_str!("../../../fixtures/execution/admission.v1.json"),
951            include_str!("../../../fixtures/execution/app-facade-prepared.v1.json"),
952            include_str!("../../../fixtures/execution/app-facade-user-action.v1.json"),
953            include_str!("../../../fixtures/execution/event-ack.v1.json"),
954            include_str!("../../../fixtures/execution/checkpoint-purge.v1.json"),
955            include_str!("../../../fixtures/execution/checkpoint-purge-ack.v1.json"),
956        ] {
957            let json: serde_json::Value =
958                serde_json::from_str(fixture).expect("valid golden fixture JSON");
959            validator
960                .validate(&json)
961                .expect("golden fixture matches the normative schema");
962            let envelope = Envelope::decode_json(fixture).expect("fixture matches Rust DTOs");
963            assert_eq!(serde_json::to_value(envelope).expect("serializes"), json);
964        }
965    }
966
967    #[test]
968    fn rejects_blank_scoped_identifiers() {
969        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
970        let mut envelope: Envelope = serde_json::from_str(fixture).expect("valid golden fixture");
971        let Payload::Command(request) = &mut envelope.payload else {
972            panic!("enqueue fixture changed family");
973        };
974        let CommandRequest {
975            command: AgentCommand::Enqueue { context, .. },
976            ..
977        } = request.as_mut()
978        else {
979            panic!("enqueue fixture changed family");
980        };
981        context.scope.scope_id = ScopeId::from(" ");
982        assert_eq!(
983            envelope.validate(),
984            Err(ProtocolError::MissingIdentifier("scope_id"))
985        );
986    }
987
988    #[test]
989    fn strict_decoder_rejects_unknown_nested_fields() {
990        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
991        let input = fixture.replacen(
992            "\"sent_at_unix_ms\": 1700000000000",
993            "\"sent_at_unix_ms\": 1700000000000, \"unexpected\": true",
994            1,
995        );
996        assert!(matches!(
997            Envelope::decode_json(&input),
998            Err(ProtocolError::UnknownField { .. })
999        ));
1000
1001        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1002        input["payload"]["payload"]["options"]["unexpected"] = serde_json::Value::Bool(true);
1003        assert!(matches!(
1004            Envelope::decode_json(&input.to_string()),
1005            Err(ProtocolError::UnknownField { .. })
1006        ));
1007    }
1008
1009    #[test]
1010    fn strict_decoder_treats_skip_optional_nulls_as_absent() {
1011        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1012        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1013        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["fast_mode"] =
1014            serde_json::Value::Null;
1015        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["thinking"] =
1016            serde_json::Value::Null;
1017        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["model_snapshot"]["generation_support"]
1018            ["fastMode"] = serde_json::Value::Null;
1019        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["model_snapshot"]["generation_support"]
1020            ["reasoningEffortDefault"] = serde_json::Value::Null;
1021        input["payload"]["payload"]["command"]["message"]["continuation"] = serde_json::Value::Null;
1022
1023        let envelope = Envelope::decode_json(&input.to_string())
1024            .expect("null on skip-optional fields deserializes as absent");
1025        let canonical = serde_json::to_value(&envelope).expect("serializes");
1026        let canonical_route = &canonical["payload"]["payload"]["llm_plan"]["routes"][0]["route"];
1027        assert!(canonical_route.get("fast_mode").is_none());
1028        assert!(canonical_route.get("thinking").is_none());
1029        let canonical_support = &canonical_route["model_snapshot"]["generation_support"];
1030        assert!(canonical_support.get("fastMode").is_none());
1031        assert!(canonical_support.get("reasoningEffortDefault").is_none());
1032        assert!(
1033            canonical["payload"]["payload"]["command"]["message"]
1034                .get("continuation")
1035                .is_none()
1036        );
1037
1038        let schema: serde_json::Value = serde_json::from_str(include_str!(
1039            "../../../schema/execution/envelope.v1.schema.json"
1040        ))
1041        .expect("schema");
1042        let validator = jsonschema::validator_for(&schema).expect("schema");
1043        assert!(
1044            validator.validate(&input).is_err(),
1045            "canonical emit omits skip-optional keys; schema must not advertise null"
1046        );
1047    }
1048
1049    #[test]
1050    fn schema_rejects_generation_support_and_continuation_nulls() {
1051        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1052        let schema: serde_json::Value = serde_json::from_str(include_str!(
1053            "../../../schema/execution/envelope.v1.schema.json"
1054        ))
1055        .expect("schema");
1056        let validator = jsonschema::validator_for(&schema).expect("schema");
1057        let golden: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1058        assert!(
1059            validator.validate(&golden).is_ok(),
1060            "golden enqueue fixture must remain emit-valid"
1061        );
1062
1063        let mut support_null = golden.clone();
1064        support_null["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["model_snapshot"]["generation_support"]
1065            ["fastMode"] = serde_json::Value::Null;
1066        assert!(
1067            validator.validate(&support_null).is_err(),
1068            "skip-optional generation_support.fastMode must omit, not null"
1069        );
1070
1071        let mut continuation_null = golden;
1072        continuation_null["payload"]["payload"]["command"]["message"]["continuation"] =
1073            serde_json::Value::Null;
1074        assert!(
1075            validator.validate(&continuation_null).is_err(),
1076            "skip-optional message.continuation must omit, not null"
1077        );
1078    }
1079
1080    #[test]
1081    fn strict_decoder_still_rejects_unknown_null_fields() {
1082        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1083        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1084        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["unexpected"] =
1085            serde_json::Value::Null;
1086        assert!(matches!(
1087            Envelope::decode_json(&input.to_string()),
1088            Err(ProtocolError::UnknownField { .. })
1089        ));
1090    }
1091
1092    #[test]
1093    fn abandoned_run_requires_a_base_checkpoint() {
1094        let mut value: serde_json::Value =
1095            serde_json::from_str(include_str!("../../../fixtures/execution/enqueue.v1.json"))
1096                .unwrap();
1097        value["payload"]["payload"]["abandoned_run_id"] =
1098            serde_json::Value::String("run-abandoned".to_owned());
1099
1100        assert!(matches!(
1101            Envelope::decode_json(&value.to_string()),
1102            Err(ProtocolError::InvalidBinding(_))
1103        ));
1104    }
1105}
conversation_api/execution/wire/mod.rs

conversation_api/execution/wire/
mod.rs
