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    /// Strictly decodes a JSON envelope and rejects fields unknown to the Rust DTO graph.
332    ///
333    /// Transport adapters should use this entry point rather than calling `serde_json` directly.
334    /// Explicit `null` on a skip-optional field (canonical serialization omits `None`) is treated
335    /// as absent. Explicit `null` on an unknown key is still rejected.
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 input_value: serde_json::Value =
342            serde_json::from_str(input).map_err(|error| ProtocolError::InvalidJson {
343                message: error.to_string(),
344            })?;
345        let mut deserializer = serde_json::Deserializer::from_str(input);
346        let mut unknown = Vec::new();
347        let envelope: Self = serde_ignored::deserialize(&mut deserializer, |path| {
348            unknown.push(path.to_string());
349        })
350        .map_err(|error| ProtocolError::InvalidJson {
351            message: error.to_string(),
352        })?;
353        deserializer
354            .end()
355            .map_err(|error| ProtocolError::InvalidJson {
356                message: error.to_string(),
357            })?;
358        if let Some(path) = unknown.into_iter().next() {
359            return Err(ProtocolError::UnknownField { path });
360        }
361        let canonical =
362            serde_json::to_value(&envelope).map_err(|error| ProtocolError::InvalidJson {
363                message: error.to_string(),
364            })?;
365        if let Some(path) = first_extra_field(&input_value, &canonical, "$") {
366            return Err(ProtocolError::UnknownField { path });
367        }
368        envelope.validate()?;
369        Ok(envelope)
370    }
371
372    /// Validates transport-level invariants before dispatch.
373    ///
374    /// # Errors
375    ///
376    /// Returns `ProtocolError` when the major version is unsupported or a required identifier is
377    /// blank.
378    pub fn validate(&self) -> Result<(), ProtocolError> {
379        if self.metadata.version != CURRENT_VERSION {
380            return Err(ProtocolError::UnsupportedVersion {
381                actual: self.metadata.version,
382            });
383        }
384        if self.metadata.message_id.trim().is_empty() {
385            return Err(ProtocolError::MissingIdentifier("message_id"));
386        }
387        if self.metadata.correlation_id.trim().is_empty() {
388            return Err(ProtocolError::MissingIdentifier("correlation_id"));
389        }
390        validate_dispatch(&self.metadata.dispatch)?;
391        if let Some(context) = self.context() {
392            validate_context(context)?;
393        }
394        if let Payload::Command(request) = &self.payload {
395            validate_command_binding(request)?;
396        }
397        if let Payload::CheckpointPurge(request) = &self.payload {
398            validate_purge_request(request)?;
399        }
400        if let Payload::CheckpointPurgeAck(ack) = &self.payload {
401            validate_purge_ack(ack)?;
402        }
403        if let Payload::EventAck(ack) = &self.payload {
404            validate_checkpoint_reference(&ack.checkpoint)?;
405        }
406        match &self.payload {
407            Payload::Command(request) => {
408                if let Some(reference) = &request.base_checkpoint {
409                    validate_checkpoint_reference(reference)?;
410                }
411                if matches!(&request.command, AgentCommand::Cancel { .. })
412                    && (request.base_checkpoint.is_some() || request.abandoned_run_id.is_some())
413                {
414                    return Err(ProtocolError::InvalidBinding(
415                        "cancel must target live execution without checkpoint recovery metadata"
416                            .to_owned(),
417                    ));
418                }
419                if request.abandoned_run_id.is_some() && request.base_checkpoint.is_none() {
420                    return Err(ProtocolError::InvalidBinding(
421                        "an abandoned run requires a base checkpoint".to_owned(),
422                    ));
423                }
424            }
425            Payload::Event(message) => validate_checkpoint_reference(&message.checkpoint)?,
426            _ => {}
427        }
428        Ok(())
429    }
430
431    fn context(&self) -> Option<&InvocationContext> {
432        match &self.payload {
433            Payload::Command(request) => match &request.command {
434                AgentCommand::Enqueue { context, .. }
435                | AgentCommand::SubmitInteraction { context, .. }
436                | AgentCommand::Cancel { context } => Some(context),
437            },
438            Payload::Event(message) => Some(&message.event.context),
439            Payload::Observation(message) => Some(&message.context),
440            Payload::ExecutionFailure(message) => Some(&message.context),
441            Payload::AppFacadeRequest(request) => match request {
442                AppFacadeRequest::Invoke { context, .. }
443                | AppFacadeRequest::Prepare { context, .. }
444                | AppFacadeRequest::Commit { context, .. }
445                | AppFacadeRequest::Reject { context, .. } => Some(context),
446            },
447            Payload::CommandResponse(response) => match response {
448                CommandResponse::Accepted { context, .. }
449                | CommandResponse::Rejected { context, .. } => Some(context),
450            },
451            Payload::EventAck(_)
452            | Payload::CheckpointPurge(_)
453            | Payload::CheckpointPurgeAck(_)
454            | Payload::AppFacadeResponse(_) => None,
455        }
456    }
457}
458
459fn validate_purge_request(request: &CheckpointPurgeRequest) -> Result<(), ProtocolError> {
460    if request.purge_id.trim().is_empty() {
461        return Err(ProtocolError::MissingIdentifier("purge_id"));
462    }
463    if request.conversation_key.trim().is_empty() {
464        return Err(ProtocolError::MissingIdentifier("conversation_key"));
465    }
466    if request.scope.scope_id.as_str().trim().is_empty() {
467        return Err(ProtocolError::MissingIdentifier("scope_id"));
468    }
469    if request.thread_id.as_str().trim().is_empty() {
470        return Err(ProtocolError::MissingIdentifier("thread_id"));
471    }
472    Ok(())
473}
474
475fn validate_purge_ack(ack: &CheckpointPurgeAck) -> Result<(), ProtocolError> {
476    let (purge_id, conversation_key, thread_id) = match ack {
477        CheckpointPurgeAck::Deleted {
478            purge_id,
479            conversation_key,
480            thread_id,
481            ..
482        }
483        | CheckpointPurgeAck::Failed {
484            purge_id,
485            conversation_key,
486            thread_id,
487            ..
488        } => (purge_id, conversation_key, thread_id),
489    };
490    if purge_id.trim().is_empty() {
491        return Err(ProtocolError::MissingIdentifier("purge_id"));
492    }
493    if conversation_key.trim().is_empty() {
494        return Err(ProtocolError::MissingIdentifier("conversation_key"));
495    }
496    if thread_id.as_str().trim().is_empty() {
497        return Err(ProtocolError::MissingIdentifier("thread_id"));
498    }
499    if let CheckpointPurgeAck::Failed { code, .. } = ack
500        && code.trim().is_empty()
501    {
502        return Err(ProtocolError::MissingIdentifier("code"));
503    }
504    Ok(())
505}
506
507fn validate_dispatch(dispatch: &DispatchBinding) -> Result<(), ProtocolError> {
508    if dispatch.dispatch_id.trim().is_empty() {
509        return Err(ProtocolError::MissingIdentifier("dispatch_id"));
510    }
511    if dispatch.agent_session_id.trim().is_empty() {
512        return Err(ProtocolError::MissingIdentifier("agent_session_id"));
513    }
514    Ok(())
515}
516
517fn validate_checkpoint_reference(reference: &CheckpointReference) -> Result<(), ProtocolError> {
518    if reference.object_key.trim().is_empty() {
519        return Err(ProtocolError::MissingIdentifier("checkpoint_object_key"));
520    }
521    if reference.format_version == 0
522        || reference.sha256.len() != 64
523        || !reference
524            .sha256
525            .bytes()
526            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
527    {
528        return Err(ProtocolError::InvalidBinding(
529            "checkpoint reference is invalid".to_owned(),
530        ));
531    }
532    Ok(())
533}
534
535fn validate_command_binding(request: &CommandRequest) -> Result<(), ProtocolError> {
536    match (&request.command, &request.context_generation) {
537        (AgentCommand::Enqueue { .. } | AgentCommand::SubmitInteraction { .. }, Some(value))
538            if value.is_complete() => {}
539        (AgentCommand::Enqueue { .. } | AgentCommand::SubmitInteraction { .. }, _) => {
540            return Err(ProtocolError::InvalidBinding(
541                "enqueue and submit_interaction require a context generation".to_owned(),
542            ));
543        }
544        (AgentCommand::Cancel { .. }, None) => {}
545        (AgentCommand::Cancel { .. }, Some(_)) => {
546            return Err(ProtocolError::InvalidBinding(
547                "cancel must not carry a context generation".to_owned(),
548            ));
549        }
550    }
551    match (&request.command, &request.llm_plan) {
552        (AgentCommand::Enqueue { options, .. }, Some(plan)) => {
553            validate_llm_plan(plan)?;
554            if plan
555                .route_for(&options.use_case, &options.model_mode)
556                .is_none()
557            {
558                return Err(ProtocolError::InvalidBinding(
559                    "LLM plan does not contain the command's primary selector".to_owned(),
560                ));
561            }
562            Ok(())
563        }
564        (AgentCommand::Enqueue { .. }, None) => Err(ProtocolError::InvalidBinding(
565            "enqueue requires a frozen LLM plan".to_owned(),
566        )),
567        (AgentCommand::SubmitInteraction { .. }, Some(plan)) => validate_llm_plan(plan),
568        (AgentCommand::SubmitInteraction { .. }, None) => Err(ProtocolError::InvalidBinding(
569            "submit_interaction requires the run's frozen LLM plan".to_owned(),
570        )),
571        (_, Some(_)) => Err(ProtocolError::InvalidBinding(
572            "only commands that invoke the model may carry an LLM plan".to_owned(),
573        )),
574        (_, None) => Ok(()),
575    }
576}
577
578fn validate_llm_plan(plan: &LlmExecutionPlan) -> Result<(), ProtocolError> {
579    if plan.routes.is_empty() {
580        return Err(ProtocolError::InvalidBinding(
581            "LLM plan must contain at least one route".to_owned(),
582        ));
583    }
584    let mut selectors = std::collections::HashSet::new();
585    for binding in &plan.routes {
586        let route = &binding.route;
587        route
588            .model_snapshot
589            .generation_support
590            .validate()
591            .map_err(|error| ProtocolError::InvalidBinding(error.into()))?;
592        llm_api::GenerationParameters {
593            temperature: route.temperature,
594            reasoning_effort: route.reasoning_effort.clone(),
595            thinking: route.thinking,
596            fast_mode: route.fast_mode,
597        }
598        .validate()
599        .map_err(|error| ProtocolError::InvalidBinding(error.into()))?;
600        if binding.use_case.0.trim().is_empty()
601            || binding.model_mode.0.trim().is_empty()
602            || route.backend.trim().is_empty()
603            || route.profile.trim().is_empty()
604            || route.model.trim().is_empty()
605        {
606            return Err(ProtocolError::InvalidBinding(
607                "LLM use case, model mode, backend, profile, and model are required".to_owned(),
608            ));
609        }
610        if !selectors.insert((binding.use_case.0.as_str(), binding.model_mode.0.as_str())) {
611            return Err(ProtocolError::InvalidBinding(
612                "LLM plan contains a duplicate selector".to_owned(),
613            ));
614        }
615        if binding
616            .revision
617            .as_ref()
618            .is_some_and(|value| value.trim().is_empty())
619            || route
620                .provider_preferences
621                .iter()
622                .any(|value| value.trim().is_empty())
623            || route
624                .reasoning_effort
625                .as_ref()
626                .is_some_and(|value| value.trim().is_empty())
627            || route.temperature.is_some_and(|value| !value.is_finite())
628            || route.context_window_tokens <= 20_000
629        {
630            return Err(ProtocolError::InvalidBinding(
631                "LLM route contains invalid optional settings".to_owned(),
632            ));
633        }
634    }
635    Ok(())
636}
637
638fn first_extra_field(
639    input: &serde_json::Value,
640    canonical: &serde_json::Value,
641    path: &str,
642) -> Option<String> {
643    match (input, canonical) {
644        (serde_json::Value::Object(input), serde_json::Value::Object(canonical)) => {
645            for (key, value) in input {
646                let child_path = format!("{path}.{key}");
647                let Some(expected) = canonical.get(key) else {
648                    // `skip_serializing_if = "Option::is_none"` omits the key. Serde still
649                    // accepts an explicit null as `None`; treat that as canonical absence.
650                    // Unknown keys are rejected earlier by `serde_ignored`, including `"k": null`.
651                    if value.is_null() {
652                        continue;
653                    }
654                    return Some(child_path);
655                };
656                if let Some(extra) = first_extra_field(value, expected, &child_path) {
657                    return Some(extra);
658                }
659            }
660            None
661        }
662        (serde_json::Value::Array(input), serde_json::Value::Array(canonical)) => input
663            .iter()
664            .zip(canonical)
665            .enumerate()
666            .find_map(|(index, (value, expected))| {
667                first_extra_field(value, expected, &format!("{path}[{index}]"))
668            }),
669        _ => None,
670    }
671}
672
673fn validate_context(context: &InvocationContext) -> Result<(), ProtocolError> {
674    validate_scope(&context.scope)?;
675    for (name, value) in [
676        ("user_id", context.actor.user_id.as_str()),
677        ("thread_id", context.thread_id.as_str()),
678        ("run_id", context.run_id.as_str()),
679        ("operation_id", context.operation_id.as_str()),
680    ] {
681        if value.trim().is_empty() {
682            return Err(ProtocolError::MissingIdentifier(name));
683        }
684    }
685    if context
686        .actor
687        .client_id
688        .as_ref()
689        .is_some_and(|client| client.as_str().trim().is_empty())
690    {
691        return Err(ProtocolError::MissingIdentifier("client_id"));
692    }
693    Ok(())
694}
695
696fn validate_scope(scope: &Scope) -> Result<(), ProtocolError> {
697    if scope.scope_id.as_str().trim().is_empty() {
698        return Err(ProtocolError::MissingIdentifier("scope_id"));
699    }
700    if scope
701        .tenant_id
702        .as_ref()
703        .is_some_and(|tenant| tenant.as_str().trim().is_empty())
704    {
705        return Err(ProtocolError::MissingIdentifier("tenant_id"));
706    }
707    Ok(())
708}
709
710/// Invalid wire envelope.
711#[derive(Clone, Debug, Error, Eq, PartialEq)]
712pub enum ProtocolError {
713    /// JSON could not be decoded into an envelope.
714    #[error("invalid protocol JSON: {message}")]
715    InvalidJson {
716        /// Safe parser diagnostic.
717        message: String,
718    },
719    /// The payload contained a field outside the normative schema.
720    #[error("unknown protocol field: {path}")]
721    UnknownField {
722        /// Serde path to the unexpected field.
723        path: String,
724    },
725    /// The sender used a protocol major version this crate does not understand.
726    #[error("unsupported protocol version {actual}")]
727    UnsupportedVersion {
728        /// Received major version.
729        actual: u16,
730    },
731    /// A required identifier was blank.
732    #[error("missing required identifier: {0}")]
733    MissingIdentifier(&'static str),
734    /// Environment binding is absent or disagrees with its Runtime command.
735    #[error("invalid command binding: {0}")]
736    InvalidBinding(String),
737}
738
739#[cfg(test)]
740mod tests {
741    use crate::execution::{
742        AccessMode, Actor, AgentEvent, ClientId, ContentPart, EventId, Message, MessageRole,
743        ModelMode, OperationId, RunId, RunOptions, Scope, ScopeId, TenantId, ThreadId, UseCase,
744        UserId,
745    };
746
747    use super::*;
748
749    #[test]
750    fn rejects_unknown_protocol_version() {
751        let envelope = Envelope {
752            metadata: Metadata {
753                version: CURRENT_VERSION + 1,
754                message_id: "message-1".to_owned(),
755                correlation_id: "correlation-1".to_owned(),
756                causation_id: None,
757                sent_at_unix_ms: 0,
758                dispatch: dispatch(),
759            },
760            payload: Payload::Event(EventMessage {
761                event: DurableEvent {
762                    id: EventId::from("event-1"),
763                    sequence: 1,
764                    context: InvocationContext {
765                        scope: Scope {
766                            tenant_id: None,
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                    event: AgentEvent::Started,
781                },
782                checkpoint: checkpoint(),
783            }),
784        };
785
786        assert_eq!(
787            envelope.validate(),
788            Err(ProtocolError::UnsupportedVersion {
789                actual: CURRENT_VERSION + 1
790            })
791        );
792    }
793
794    #[test]
795    fn command_fixture_round_trips_without_shape_drift() {
796        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
797        let envelope = Envelope::decode_json(fixture).expect("valid golden fixture");
798
799        let expected = Envelope {
800            metadata: Metadata {
801                version: CURRENT_VERSION,
802                message_id: "message-1".to_owned(),
803                correlation_id: "correlation-1".to_owned(),
804                causation_id: None,
805                sent_at_unix_ms: 1_700_000_000_000,
806                dispatch: dispatch(),
807            },
808            payload: Payload::Command(Box::new(CommandRequest {
809                command: AgentCommand::Enqueue {
810                    context: InvocationContext {
811                        scope: Scope {
812                            tenant_id: Some(TenantId::from("tenant-1")),
813                            scope_id: ScopeId::from("scope-1"),
814                        },
815                        actor: Actor {
816                            user_id: UserId::from("user-1"),
817                            client_id: Some(ClientId::from("client-1")),
818                        },
819                        surface_id: ConversationSurface::client_personal("user-1").unwrap(),
820                        thread_id: ThreadId::from("thread-1"),
821                        run_id: RunId::from("run-1"),
822                        operation_id: OperationId::from("operation-1"),
823                        deadline_unix_ms: None,
824                        traceparent: None,
825                    },
826                    message: Message {
827                        continuation: None,
828                        role: MessageRole::User,
829                        content: vec![ContentPart::Text {
830                            text: "hello".to_owned(),
831                        }],
832                    },
833                    options: RunOptions {
834                        use_case: UseCase("chat".to_owned()),
835                        model_mode: ModelMode("auto".to_owned()),
836                        access_mode: AccessMode::Interactive,
837                        allow_tools: true,
838                        max_steps: 8,
839                        credit_budget: None,
840                        debug: false,
841                    },
842                },
843                context_generation: Some(ContextGeneration {
844                    user_scope: "scope-1".to_owned(),
845                    identity: "identity-1".to_owned(),
846                    memory: "memory-1".to_owned(),
847                    integration_guide: "guide-1".to_owned(),
848                    installed_integrations: "installed-1".to_owned(),
849                    scope_integrations: "scope-integrations-1".to_owned(),
850                }),
851                llm_plan: Some(LlmExecutionPlan {
852                    routes: [
853                        "chat",
854                        "agent_info_merge",
855                        "automation_judge",
856                        "automation_diagnose",
857                        "app_guide",
858                        "workflow.action_operate",
859                        "workflow.automation_generate",
860                        "workflow.description_generate",
861                    ]
862                    .into_iter()
863                    .map(|use_case| LlmRouteBinding {
864                        use_case: UseCase(use_case.to_owned()),
865                        model_mode: ModelMode("auto".to_owned()),
866                        revision: Some("1700000000000".to_owned()),
867                        route: LlmRoute {
868                            backend: "openrouter".to_owned(),
869                            profile: "cloud_openrouter".to_owned(),
870                            model: "openai/gpt-5.4-mini".to_owned(),
871                            provider_preferences: vec!["OpenAI".to_owned()],
872                            temperature: Some(0.5),
873                            cache_control: Some(true),
874                            reasoning_effort: None,
875                            thinking: None,
876                            fast_mode: None,
877                            context_window_tokens: 128_000,
878                            model_snapshot: LlmModelSnapshot {
879                                capabilities: llm_api::ModelCapabilities {
880                                    input: vec!["image".to_owned()],
881                                    tool_calling: true,
882                                    structured_output: true,
883                                },
884                                generation_support: LlmGenerationSupport {
885                                    temperature: Some(true),
886                                    max_tokens: Some(true),
887                                    thinking: Some(true),
888                                    reasoning_efforts: Some(vec!["high".into()]),
889                                    temperature_with_reasoning: Some(true),
890                                    ..LlmGenerationSupport::default()
891                                },
892                            },
893                        },
894                    })
895                    .collect(),
896                }),
897                base_checkpoint: None,
898                abandoned_run_id: None,
899            })),
900        };
901        assert_eq!(envelope, expected);
902        assert_eq!(
903            serde_json::to_value(envelope).expect("serializes"),
904            serde_json::from_str::<serde_json::Value>(fixture).expect("fixture JSON")
905        );
906    }
907
908    #[test]
909    fn snapshot_is_required_and_support_metadata_is_strict() {
910        let mut fixture: serde_json::Value = serde_json::from_str(ENQUEUE_FIXTURE).unwrap();
911        fixture["payload"]["payload"]["llm_plan"]["routes"][0]["route"]
912            .as_object_mut()
913            .unwrap()
914            .remove("model_snapshot");
915        assert!(Envelope::decode_json(&fixture.to_string()).is_err());
916        let mut envelope = Envelope::decode_json(ENQUEUE_FIXTURE).unwrap();
917        let Payload::Command(command) = &mut envelope.payload else {
918            panic!("command");
919        };
920        command.llm_plan.as_mut().unwrap().routes[0]
921            .route
922            .model_snapshot
923            .generation_support
924            .max_output_tokens = Some(0);
925        assert!(envelope.validate().is_err());
926    }
927
928    #[test]
929    fn generation_support_travels_in_catalog_spelling() {
930        let support = LlmGenerationSupport {
931            temperature: Some(true),
932            max_tokens: Some(true),
933            thinking: Some(true),
934            reasoning_efforts: Some(vec!["low".into(), "high".into()]),
935            reasoning_effort_default: Some("high".into()),
936            temperature_with_reasoning: Some(false),
937            temperature_max: Some(2.0),
938            max_output_tokens: Some(8_192),
939            fast_mode: Some(true),
940        };
941        let encoded = serde_json::to_value(&support).unwrap();
942        assert_eq!(
943            encoded["reasoningEfforts"],
944            serde_json::json!(["low", "high"])
945        );
946        assert_eq!(
947            encoded["temperatureWithReasoning"],
948            serde_json::json!(false)
949        );
950        assert_eq!(
951            serde_json::from_value::<LlmGenerationSupport>(encoded).unwrap(),
952            support
953        );
954        let ignored = serde_json::from_value::<LlmGenerationSupport>(serde_json::json!({
955            "temperature": true, "maxTokens": true, "reasoningEfforts": ["low"],
956            "temperatureWithReasoning": null, "temperatureMax": null, "maxOutputTokens": null,
957            "reasoningIntensity": "high"
958        }))
959        .unwrap();
960        assert_eq!(ignored.temperature, Some(true));
961        assert_eq!(
962            ignored.reasoning_efforts.as_deref(),
963            Some(["low".to_string()].as_slice())
964        );
965    }
966
967    fn dispatch() -> DispatchBinding {
968        DispatchBinding {
969            dispatch_id: "dispatch-1".to_owned(),
970            agent_session_id: "agent-session-1".to_owned(),
971        }
972    }
973
974    fn checkpoint() -> CheckpointReference {
975        CheckpointReference {
976            object_key: "checkpoints/abc.json".to_owned(),
977            sha256: "a".repeat(64),
978            format_version: 1,
979        }
980    }
981
982    #[test]
983    fn every_v1_golden_fixture_round_trips() {
984        let schema: serde_json::Value = serde_json::from_str(include_str!(
985            "../../../schema/execution/envelope.v1.schema.json"
986        ))
987        .expect("valid JSON Schema document");
988        let validator = jsonschema::validator_for(&schema).expect("valid JSON Schema semantics");
989        for fixture in [
990            include_str!("../../../fixtures/execution/enqueue.v1.json"),
991            include_str!("../../../fixtures/execution/completed-event.v1.json"),
992            include_str!("../../../fixtures/execution/interaction-required-event.v1.json"),
993            include_str!("../../../fixtures/execution/failed-event.v1.json"),
994            include_str!("../../../fixtures/execution/app-facade-prepare.v1.json"),
995            include_str!("../../../fixtures/execution/observation.v1.json"),
996            include_str!("../../../fixtures/execution/admission.v1.json"),
997            include_str!("../../../fixtures/execution/app-facade-prepared.v1.json"),
998            include_str!("../../../fixtures/execution/app-facade-user-action.v1.json"),
999            include_str!("../../../fixtures/execution/event-ack.v1.json"),
1000            include_str!("../../../fixtures/execution/checkpoint-purge.v1.json"),
1001            include_str!("../../../fixtures/execution/checkpoint-purge-ack.v1.json"),
1002        ] {
1003            let json: serde_json::Value =
1004                serde_json::from_str(fixture).expect("valid golden fixture JSON");
1005            validator
1006                .validate(&json)
1007                .expect("golden fixture matches the normative schema");
1008            let envelope = Envelope::decode_json(fixture).expect("fixture matches Rust DTOs");
1009            assert_eq!(serde_json::to_value(envelope).expect("serializes"), json);
1010        }
1011    }
1012
1013    #[test]
1014    fn rejects_blank_scoped_identifiers() {
1015        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1016        let mut envelope: Envelope = serde_json::from_str(fixture).expect("valid golden fixture");
1017        let Payload::Command(request) = &mut envelope.payload else {
1018            panic!("enqueue fixture changed family");
1019        };
1020        let CommandRequest {
1021            command: AgentCommand::Enqueue { context, .. },
1022            ..
1023        } = request.as_mut()
1024        else {
1025            panic!("enqueue fixture changed family");
1026        };
1027        context.scope.scope_id = ScopeId::from(" ");
1028        assert_eq!(
1029            envelope.validate(),
1030            Err(ProtocolError::MissingIdentifier("scope_id"))
1031        );
1032    }
1033
1034    #[test]
1035    fn strict_decoder_rejects_unknown_nested_fields() {
1036        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1037        let input = fixture.replacen(
1038            "\"sent_at_unix_ms\": 1700000000000",
1039            "\"sent_at_unix_ms\": 1700000000000, \"unexpected\": true",
1040            1,
1041        );
1042        assert!(matches!(
1043            Envelope::decode_json(&input),
1044            Err(ProtocolError::UnknownField { .. })
1045        ));
1046
1047        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1048        input["payload"]["payload"]["options"]["unexpected"] = serde_json::Value::Bool(true);
1049        assert!(matches!(
1050            Envelope::decode_json(&input.to_string()),
1051            Err(ProtocolError::UnknownField { .. })
1052        ));
1053    }
1054
1055    #[test]
1056    fn strict_decoder_treats_skip_optional_nulls_as_absent() {
1057        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1058        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1059        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["fast_mode"] =
1060            serde_json::Value::Null;
1061        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["thinking"] =
1062            serde_json::Value::Null;
1063        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["model_snapshot"]["generation_support"]
1064            ["fastMode"] = serde_json::Value::Null;
1065        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["model_snapshot"]["generation_support"]
1066            ["reasoningEffortDefault"] = serde_json::Value::Null;
1067        input["payload"]["payload"]["command"]["message"]["continuation"] = serde_json::Value::Null;
1068
1069        let envelope = Envelope::decode_json(&input.to_string())
1070            .expect("null on skip-optional fields matches canonical omit");
1071        let canonical = serde_json::to_value(&envelope).expect("serializes");
1072        let canonical_route = &canonical["payload"]["payload"]["llm_plan"]["routes"][0]["route"];
1073        assert!(canonical_route.get("fast_mode").is_none());
1074        assert!(canonical_route.get("thinking").is_none());
1075        let canonical_support = &canonical_route["model_snapshot"]["generation_support"];
1076        assert!(canonical_support.get("fastMode").is_none());
1077        assert!(canonical_support.get("reasoningEffortDefault").is_none());
1078        assert!(
1079            canonical["payload"]["payload"]["command"]["message"]
1080                .get("continuation")
1081                .is_none()
1082        );
1083    }
1084
1085    #[test]
1086    fn strict_decoder_still_rejects_unknown_null_fields() {
1087        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1088        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1089        input["payload"]["payload"]["llm_plan"]["routes"][0]["route"]["unexpected"] =
1090            serde_json::Value::Null;
1091        assert!(matches!(
1092            Envelope::decode_json(&input.to_string()),
1093            Err(ProtocolError::UnknownField { .. })
1094        ));
1095    }
1096
1097    #[test]
1098    fn abandoned_run_requires_a_base_checkpoint() {
1099        let mut value: serde_json::Value =
1100            serde_json::from_str(include_str!("../../../fixtures/execution/enqueue.v1.json"))
1101                .unwrap();
1102        value["payload"]["payload"]["abandoned_run_id"] =
1103            serde_json::Value::String("run-abandoned".to_owned());
1104
1105        assert!(matches!(
1106            Envelope::decode_json(&value.to_string()),
1107            Err(ProtocolError::InvalidBinding(_))
1108        ));
1109    }
1110}
conversation_api/execution/wire/mod.rs

conversation_api/execution/wire/
mod.rs
