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

conversation_api/execution/wire/
mod.rs
