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,
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")]
62#[serde(deny_unknown_fields)]
63pub enum AppFacadeRequest {
64    /// Executes a direct App Facade request.
65    Invoke {
66        /// Authenticated invocation context.
67        context: InvocationContext,
68        /// Canonical Agent-selected request.
69        request: FacadeRequest,
70    },
71    /// Prepares a mutating operation.
72    Prepare {
73        /// Authenticated invocation context.
74        context: InvocationContext,
75        /// Canonical Agent-selected request.
76        request: FacadeRequest,
77    },
78    /// Commits a previously prepared operation.
79    Commit {
80        /// Authenticated invocation context.
81        context: InvocationContext,
82        /// Prepared operation and idempotency metadata.
83        action: PreparedAction,
84    },
85    /// Rejects a previously prepared operation.
86    Reject {
87        /// Authenticated invocation context.
88        context: InvocationContext,
89        /// Prepared operation and idempotency metadata.
90        action: PreparedAction,
91    },
92}
93
94/// Immediate response to command admission for the selected execution attempt.
95#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
96#[serde(tag = "type", rename_all = "snake_case")]
97#[serde(deny_unknown_fields)]
98pub enum CommandResponse {
99    /// Command was admitted; final state is delivered through events.
100    Accepted {
101        /// Authenticated context of the admitted command.
102        context: InvocationContext,
103        /// Stable admission disposition.
104        disposition: AdmissionDisposition,
105        /// Queue revision when the command mutated the admission queue.
106        queue_revision: Option<Revision>,
107    },
108    /// Command was rejected before admission.
109    Rejected {
110        /// Authenticated context of the rejected command.
111        context: InvocationContext,
112        /// Stable machine-readable error code.
113        code: String,
114        /// Safe diagnostic message.
115        message: String,
116    },
117}
118
119/// Stable wire representation of command admission.
120#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
121#[serde(rename_all = "snake_case")]
122pub enum AdmissionDisposition {
123    /// A Lion-dispatched request entered the Agent process's execution queue.
124    Queued,
125    /// Interaction decisions are durable and ready to resume.
126    InteractionReady,
127    /// Cancellation has been signaled to the live Agent process.
128    CancelRequested,
129}
130
131/// Routed best-effort observation sent outside the durable event outbox.
132#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
133pub struct ObservationMessage {
134    /// Monotonic within the bound execution; gaps are allowed and never replayed.
135    pub sequence: u64,
136    /// Authenticated invocation and routing context.
137    pub context: InvocationContext,
138    /// Request-scoped progress payload.
139    pub observation: AgentObservation,
140}
141
142/// Terminal transport failure for an admitted execution that could not produce a checkpoint event.
143#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
144pub struct ExecutionFailure {
145    pub context: InvocationContext,
146    pub code: String,
147    pub message: String,
148}
149
150/// Acknowledges one durable event after the receiver has applied or deduplicated it.
151#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
152pub struct EventAck {
153    pub event_id: EventId,
154    /// Checkpoint reference Lion actually selected while applying (or deduplicating) the event.
155    pub checkpoint: CheckpointReference,
156}
157
158/// Lion-owned archive GC request for all Runtime checkpoints under one private thread.
159#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
160pub struct CheckpointPurgeRequest {
161    pub purge_id: String,
162    pub conversation_key: String,
163    pub scope: Scope,
164    pub surface_id: ConversationSurface,
165    pub thread_id: ThreadId,
166}
167
168/// Result of one idempotent checkpoint-prefix purge.
169#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
170#[serde(tag = "status", rename_all = "snake_case")]
171#[serde(deny_unknown_fields)]
172pub enum CheckpointPurgeAck {
173    Deleted {
174        purge_id: String,
175        conversation_key: String,
176        thread_id: ThreadId,
177        deleted_objects: u32,
178    },
179    Failed {
180        purge_id: String,
181        conversation_key: String,
182        thread_id: ThreadId,
183        code: String,
184        message: String,
185    },
186}
187
188/// Model facts captured by the deployment with the desired route parameters.
189/// This snapshot belongs to admission, not to the provider-neutral LLM request.
190#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
191#[serde(deny_unknown_fields)]
192pub struct LlmModelSnapshot {
193    pub capabilities: llm_api::ModelCapabilities,
194    pub generation_support: LlmGenerationSupport,
195}
196
197/// Catalog parameter support travels with the plan so a run never resolves against a
198/// catalog that changed after admission. Resolution itself belongs to `llm_api::resolve`.
199pub use llm_api::GenerationSupport as LlmGenerationSupport;
200
201/// One concrete cloud model route frozen by the deployment before command admission.
202/// The generation fields are the deployment's desired preferences, not resolved values:
203/// the runtime resolves them once against `model_snapshot` and the backend protocol.
204#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
205pub struct LlmRoute {
206    pub backend: String,
207    pub profile: String,
208    pub model: String,
209    pub provider_preferences: Vec<String>,
210    pub temperature: Option<f64>,
211    pub cache_control: Option<bool>,
212    pub reasoning_effort: Option<String>,
213    #[serde(default, skip_serializing_if = "Option::is_none")]
214    pub thinking: Option<bool>,
215    #[serde(default, skip_serializing_if = "Option::is_none")]
216    pub fast_mode: Option<bool>,
217    /// Total context window of the exact frozen model route.
218    pub context_window_tokens: u32,
219    /// Required immutable facts; old plans must not silently resolve against a new catalog.
220    pub model_snapshot: LlmModelSnapshot,
221}
222
223/// One resolved model configuration frozen for the entire admitted run.
224/// `model_mode` records the entry-point selection; internal calls do not re-resolve it.
225#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
226pub struct LlmExecutionPlan {
227    pub model_mode: ModelMode,
228    pub revision: Option<String>,
229    pub route: LlmRoute,
230}
231
232/// Transport admission request. Environment bindings are consumed by the composition root and do
233/// not enter the provider-neutral Runtime command.
234#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
235pub struct CommandRequest {
236    pub command: AgentCommand,
237    /// Effective prompt-context generation selected by Lion for this dispatch.
238    pub context_generation: Option<ContextGeneration>,
239    pub llm_plan: Option<LlmExecutionPlan>,
240    /// Stable Runtime checkpoint selected by Lion, absent for a new thread.
241    pub base_checkpoint: Option<CheckpointReference>,
242    /// Attempt already terminalized by Lion and settled silently before new work starts.
243    pub abandoned_run_id: Option<RunId>,
244}
245
246/// Checkpoint-coupled Runtime event. Lion commits the generation before projecting the event.
247#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
248pub struct EventMessage {
249    pub event: DurableEvent,
250    pub checkpoint: CheckpointReference,
251}
252
253/// Result of an App Facade request transported back to Runtime.
254#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
255#[serde(tag = "type", rename_all = "snake_case")]
256#[serde(deny_unknown_fields)]
257pub enum AppFacadeResponse {
258    /// Read-only or committed operation result.
259    Result {
260        /// Structured App Facade result.
261        result: FacadeResult,
262    },
263    /// Prepared operation awaiting a commit or rejection.
264    Prepared {
265        /// Prepared action.
266        action: PreparedAction,
267    },
268    /// Rejection completed successfully.
269    #[serde(deserialize_with = "crate::execution::deserialize_empty_variant")]
270    Rejected,
271    /// Stable error safe to expose across the transport.
272    Failed {
273        /// Machine-readable category.
274        code: String,
275        /// Safe diagnostic message.
276        message: String,
277        /// Optional retry delay.
278        retry_after_ms: Option<u64>,
279    },
280}
281
282/// Payload families carried by the versioned envelope.
283#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
284#[serde(tag = "family", content = "payload", rename_all = "snake_case")]
285pub enum Payload {
286    /// Command entering Runtime.
287    Command(Box<CommandRequest>),
288    /// Immediate command-admission response.
289    CommandResponse(CommandResponse),
290    /// Event leaving Runtime.
291    Event(EventMessage),
292    /// Durable receiver acknowledgement used to clear the Runtime outbox.
293    EventAck(EventAck),
294    /// Lion asks the cloud composition to remove an archived thread's Runtime checkpoints.
295    CheckpointPurge(CheckpointPurgeRequest),
296    /// Agent-cloud reports the idempotent S3 purge result.
297    CheckpointPurgeAck(CheckpointPurgeAck),
298    /// Non-durable progress leaving Runtime.
299    Observation(ObservationMessage),
300    /// One dispatch failed outside the checkpointed Runtime state machine.
301    ExecutionFailure(ExecutionFailure),
302    /// Runtime request to the App Facade.
303    AppFacadeRequest(AppFacadeRequest),
304    /// App Facade response returned to Runtime.
305    AppFacadeResponse(AppFacadeResponse),
306}
307
308/// Complete transport message.
309#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
310pub struct Envelope {
311    /// Correlation and version metadata.
312    pub metadata: Metadata,
313    /// Typed payload.
314    pub payload: Payload,
315}
316
317impl Envelope {
318    /// Decodes a JSON envelope against the protocol types.
319    ///
320    /// Transport adapters should use this entry point rather than calling `serde_json` directly.
321    /// Unknown keys are rejected. Canonical emit omits skip-optional `None`; an explicit `null` on
322    /// those keys is accepted as absent so accept follows the types instead of a serialize roundtrip.
323    ///
324    /// # Errors
325    ///
326    /// Returns `ProtocolError` for malformed JSON, unknown fields, or failed envelope validation.
327    pub fn decode_json(input: &str) -> Result<Self, ProtocolError> {
328        let mut deserializer = serde_json::Deserializer::from_str(input);
329        let mut unknown = Vec::new();
330        let envelope: Self = serde_ignored::deserialize(&mut deserializer, |path| {
331            unknown.push(path.to_string());
332        })
333        .map_err(|error| ProtocolError::InvalidJson {
334            message: error.to_string(),
335        })?;
336        deserializer
337            .end()
338            .map_err(|error| ProtocolError::InvalidJson {
339                message: error.to_string(),
340            })?;
341        if let Some(path) = unknown.into_iter().next() {
342            return Err(ProtocolError::UnknownField { path });
343        }
344        envelope.validate()?;
345        Ok(envelope)
346    }
347
348    /// Validates transport-level invariants before dispatch.
349    ///
350    /// # Errors
351    ///
352    /// Returns `ProtocolError` when the major version is unsupported or a required identifier is
353    /// blank.
354    pub fn validate(&self) -> Result<(), ProtocolError> {
355        if self.metadata.version != CURRENT_VERSION {
356            return Err(ProtocolError::UnsupportedVersion {
357                actual: self.metadata.version,
358            });
359        }
360        if self.metadata.message_id.trim().is_empty() {
361            return Err(ProtocolError::MissingIdentifier("message_id"));
362        }
363        if self.metadata.correlation_id.trim().is_empty() {
364            return Err(ProtocolError::MissingIdentifier("correlation_id"));
365        }
366        validate_dispatch(&self.metadata.dispatch)?;
367        if matches!(&self.payload, Payload::Observation(message) if message.sequence == 0) {
368            return Err(ProtocolError::InvalidBinding(
369                "observation sequence must be positive".into(),
370            ));
371        }
372        if let Some(context) = self.context() {
373            validate_context(context)?;
374        }
375        if let Payload::Command(request) = &self.payload {
376            validate_command_binding(request)?;
377        }
378        if let Payload::CheckpointPurge(request) = &self.payload {
379            validate_purge_request(request)?;
380        }
381        if let Payload::CheckpointPurgeAck(ack) = &self.payload {
382            validate_purge_ack(ack)?;
383        }
384        if let Payload::EventAck(ack) = &self.payload {
385            validate_checkpoint_reference(&ack.checkpoint)?;
386        }
387        match &self.payload {
388            Payload::Command(request) => {
389                if let Some(reference) = &request.base_checkpoint {
390                    validate_checkpoint_reference(reference)?;
391                }
392                if matches!(&request.command, AgentCommand::Cancel { .. })
393                    && (request.base_checkpoint.is_some() || request.abandoned_run_id.is_some())
394                {
395                    return Err(ProtocolError::InvalidBinding(
396                        "cancel must target live execution without checkpoint recovery metadata"
397                            .to_owned(),
398                    ));
399                }
400                if request.abandoned_run_id.is_some() && request.base_checkpoint.is_none() {
401                    return Err(ProtocolError::InvalidBinding(
402                        "an abandoned run requires a base checkpoint".to_owned(),
403                    ));
404                }
405            }
406            Payload::Event(message) => validate_checkpoint_reference(&message.checkpoint)?,
407            _ => {}
408        }
409        Ok(())
410    }
411
412    fn context(&self) -> Option<&InvocationContext> {
413        match &self.payload {
414            Payload::Command(request) => match &request.command {
415                AgentCommand::Enqueue { context, .. }
416                | AgentCommand::SubmitInteraction { context, .. }
417                | AgentCommand::Cancel { context } => Some(context),
418            },
419            Payload::Event(message) => Some(&message.event.context),
420            Payload::Observation(message) => Some(&message.context),
421            Payload::ExecutionFailure(message) => Some(&message.context),
422            Payload::AppFacadeRequest(request) => match request {
423                AppFacadeRequest::Invoke { context, .. }
424                | AppFacadeRequest::Prepare { context, .. }
425                | AppFacadeRequest::Commit { context, .. }
426                | AppFacadeRequest::Reject { context, .. } => Some(context),
427            },
428            Payload::CommandResponse(response) => match response {
429                CommandResponse::Accepted { context, .. }
430                | CommandResponse::Rejected { context, .. } => Some(context),
431            },
432            Payload::EventAck(_)
433            | Payload::CheckpointPurge(_)
434            | Payload::CheckpointPurgeAck(_)
435            | Payload::AppFacadeResponse(_) => None,
436        }
437    }
438}
439
440fn validate_purge_request(request: &CheckpointPurgeRequest) -> Result<(), ProtocolError> {
441    if request.purge_id.trim().is_empty() {
442        return Err(ProtocolError::MissingIdentifier("purge_id"));
443    }
444    if request.conversation_key.trim().is_empty() {
445        return Err(ProtocolError::MissingIdentifier("conversation_key"));
446    }
447    if request.scope.scope_id.as_str().trim().is_empty() {
448        return Err(ProtocolError::MissingIdentifier("scope_id"));
449    }
450    if request.thread_id.as_str().trim().is_empty() {
451        return Err(ProtocolError::MissingIdentifier("thread_id"));
452    }
453    Ok(())
454}
455
456fn validate_purge_ack(ack: &CheckpointPurgeAck) -> Result<(), ProtocolError> {
457    let (purge_id, conversation_key, thread_id) = match ack {
458        CheckpointPurgeAck::Deleted {
459            purge_id,
460            conversation_key,
461            thread_id,
462            ..
463        }
464        | CheckpointPurgeAck::Failed {
465            purge_id,
466            conversation_key,
467            thread_id,
468            ..
469        } => (purge_id, conversation_key, thread_id),
470    };
471    if purge_id.trim().is_empty() {
472        return Err(ProtocolError::MissingIdentifier("purge_id"));
473    }
474    if conversation_key.trim().is_empty() {
475        return Err(ProtocolError::MissingIdentifier("conversation_key"));
476    }
477    if thread_id.as_str().trim().is_empty() {
478        return Err(ProtocolError::MissingIdentifier("thread_id"));
479    }
480    if let CheckpointPurgeAck::Failed { code, .. } = ack
481        && code.trim().is_empty()
482    {
483        return Err(ProtocolError::MissingIdentifier("code"));
484    }
485    Ok(())
486}
487
488fn validate_dispatch(dispatch: &DispatchBinding) -> Result<(), ProtocolError> {
489    if dispatch.dispatch_id.trim().is_empty() {
490        return Err(ProtocolError::MissingIdentifier("dispatch_id"));
491    }
492    if dispatch.agent_session_id.trim().is_empty() {
493        return Err(ProtocolError::MissingIdentifier("agent_session_id"));
494    }
495    Ok(())
496}
497
498fn validate_checkpoint_reference(reference: &CheckpointReference) -> Result<(), ProtocolError> {
499    if reference.object_key.trim().is_empty() {
500        return Err(ProtocolError::MissingIdentifier("checkpoint_object_key"));
501    }
502    if reference.format_version == 0
503        || reference.sha256.len() != 64
504        || !reference
505            .sha256
506            .bytes()
507            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
508    {
509        return Err(ProtocolError::InvalidBinding(
510            "checkpoint reference is invalid".to_owned(),
511        ));
512    }
513    Ok(())
514}
515
516fn validate_command_binding(request: &CommandRequest) -> Result<(), ProtocolError> {
517    match (&request.command, &request.context_generation) {
518        (AgentCommand::Enqueue { .. } | AgentCommand::SubmitInteraction { .. }, Some(value))
519            if value.is_complete() => {}
520        (AgentCommand::Enqueue { .. } | AgentCommand::SubmitInteraction { .. }, _) => {
521            return Err(ProtocolError::InvalidBinding(
522                "enqueue and submit_interaction require a context generation".to_owned(),
523            ));
524        }
525        (AgentCommand::Cancel { .. }, None) => {}
526        (AgentCommand::Cancel { .. }, Some(_)) => {
527            return Err(ProtocolError::InvalidBinding(
528                "cancel must not carry a context generation".to_owned(),
529            ));
530        }
531    }
532    match (&request.command, &request.llm_plan) {
533        (AgentCommand::Enqueue { options, .. }, Some(plan)) => {
534            validate_llm_plan(plan)?;
535            if plan.model_mode != options.model_mode {
536                return Err(ProtocolError::InvalidBinding(
537                    "LLM plan mode differs from the command's selected mode".to_owned(),
538                ));
539            }
540            Ok(())
541        }
542        (AgentCommand::Enqueue { .. }, None) => Err(ProtocolError::InvalidBinding(
543            "enqueue requires a frozen LLM plan".to_owned(),
544        )),
545        (AgentCommand::SubmitInteraction { .. }, Some(plan)) => validate_llm_plan(plan),
546        (AgentCommand::SubmitInteraction { .. }, None) => Err(ProtocolError::InvalidBinding(
547            "submit_interaction requires the run's frozen LLM plan".to_owned(),
548        )),
549        (_, Some(_)) => Err(ProtocolError::InvalidBinding(
550            "only commands that invoke the model may carry an LLM plan".to_owned(),
551        )),
552        (_, None) => Ok(()),
553    }
554}
555
556fn validate_llm_plan(plan: &LlmExecutionPlan) -> Result<(), ProtocolError> {
557    let route = &plan.route;
558    route
559        .model_snapshot
560        .generation_support
561        .validate()
562        .map_err(|error| ProtocolError::InvalidBinding(error.into()))?;
563    llm_api::GenerationParameters {
564        temperature: route.temperature,
565        reasoning_effort: route.reasoning_effort.clone(),
566        thinking: route.thinking,
567        fast_mode: route.fast_mode,
568    }
569    .validate()
570    .map_err(|error| ProtocolError::InvalidBinding(error.into()))?;
571    if plan.model_mode.0.trim().is_empty()
572        || route.backend.trim().is_empty()
573        || route.profile.trim().is_empty()
574        || route.model.trim().is_empty()
575    {
576        return Err(ProtocolError::InvalidBinding(
577            "LLM model mode, backend, profile, and model are required".to_owned(),
578        ));
579    }
580    if plan
581        .revision
582        .as_ref()
583        .is_some_and(|value| value.trim().is_empty())
584        || route
585            .provider_preferences
586            .iter()
587            .any(|value| value.trim().is_empty())
588        || route
589            .reasoning_effort
590            .as_ref()
591            .is_some_and(|value| value.trim().is_empty())
592        || route.temperature.is_some_and(|value| !value.is_finite())
593        || route.context_window_tokens <= 20_000
594    {
595        return Err(ProtocolError::InvalidBinding(
596            "LLM route contains invalid optional settings".to_owned(),
597        ));
598    }
599    Ok(())
600}
601
602fn validate_context(context: &InvocationContext) -> Result<(), ProtocolError> {
603    validate_scope(&context.scope)?;
604    for (name, value) in [
605        ("thread_id", context.thread_id.as_str()),
606        ("run_id", context.run_id.as_str()),
607        ("operation_id", context.operation_id.as_str()),
608    ] {
609        if value.trim().is_empty() {
610            return Err(ProtocolError::MissingIdentifier(name));
611        }
612    }
613    if !context.actor.valid_for(&context.scope, &context.surface_id) {
614        return Err(ProtocolError::InvalidBinding(
615            "Actor does not match the conversation scope and surface".into(),
616        ));
617    }
618    Ok(())
619}
620
621fn validate_scope(scope: &Scope) -> Result<(), ProtocolError> {
622    if scope.scope_id.as_str().trim().is_empty() {
623        return Err(ProtocolError::MissingIdentifier("scope_id"));
624    }
625    if scope
626        .tenant_id
627        .as_ref()
628        .is_some_and(|tenant| tenant.as_str().trim().is_empty())
629    {
630        return Err(ProtocolError::MissingIdentifier("tenant_id"));
631    }
632    Ok(())
633}
634
635/// Invalid wire envelope.
636#[derive(Clone, Debug, Error, Eq, PartialEq)]
637pub enum ProtocolError {
638    /// JSON could not be decoded into an envelope.
639    #[error("invalid protocol JSON: {message}")]
640    InvalidJson {
641        /// Safe parser diagnostic.
642        message: String,
643    },
644    /// The payload contained a field outside the normative schema.
645    #[error("unknown protocol field: {path}")]
646    UnknownField {
647        /// Serde path to the unexpected field.
648        path: String,
649    },
650    /// The sender used a protocol major version this crate does not understand.
651    #[error("unsupported protocol version {actual}")]
652    UnsupportedVersion {
653        /// Received major version.
654        actual: u16,
655    },
656    /// A required identifier was blank.
657    #[error("missing required identifier: {0}")]
658    MissingIdentifier(&'static str),
659    /// Environment binding is absent or disagrees with its Runtime command.
660    #[error("invalid command binding: {0}")]
661    InvalidBinding(String),
662}
663
664#[cfg(test)]
665mod tests {
666    use crate::execution::{
667        AccessMode, Actor, AgentEvent, ClientId, ContentPart, EventId, Message, MessageRole,
668        ModelMode, OperationId, RunId, RunOptions, Scope, ScopeId, TenantId, ThreadId, UseCase,
669        UserId,
670    };
671
672    use super::*;
673
674    #[test]
675    fn rejects_unknown_protocol_version() {
676        let envelope = Envelope {
677            metadata: Metadata {
678                version: CURRENT_VERSION + 1,
679                message_id: "message-1".to_owned(),
680                correlation_id: "correlation-1".to_owned(),
681                causation_id: None,
682                sent_at_unix_ms: 0,
683                dispatch: dispatch(),
684            },
685            payload: Payload::Event(EventMessage {
686                event: DurableEvent {
687                    id: EventId::from("event-1"),
688                    sequence: 1,
689                    context: InvocationContext {
690                        scope: Scope {
691                            tenant_id: None,
692                            scope_id: ScopeId::from("scope-1"),
693                        },
694                        actor: Actor::user(
695                            UserId::from("user-1"),
696                            Some(ClientId::from("client-1")),
697                        ),
698                        surface_id: ConversationSurface::client_personal("user-1").unwrap(),
699                        thread_id: ThreadId::from("thread-1"),
700                        run_id: RunId::from("run-1"),
701                        operation_id: OperationId::from("operation-1"),
702                        deadline_unix_ms: None,
703                        traceparent: None,
704                    },
705                    event: AgentEvent::Started,
706                },
707                checkpoint: checkpoint(),
708            }),
709        };
710
711        assert_eq!(
712            envelope.validate(),
713            Err(ProtocolError::UnsupportedVersion {
714                actual: CURRENT_VERSION + 1
715            })
716        );
717    }
718
719    #[test]
720    fn command_fixture_round_trips_without_shape_drift() {
721        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
722        let envelope = Envelope::decode_json(fixture).expect("valid golden fixture");
723
724        let expected = Envelope {
725            metadata: Metadata {
726                version: CURRENT_VERSION,
727                message_id: "message-1".to_owned(),
728                correlation_id: "correlation-1".to_owned(),
729                causation_id: None,
730                sent_at_unix_ms: 1_700_000_000_000,
731                dispatch: dispatch(),
732            },
733            payload: Payload::Command(Box::new(CommandRequest {
734                command: AgentCommand::Enqueue {
735                    context: InvocationContext {
736                        scope: Scope {
737                            tenant_id: Some(TenantId::from("tenant-1")),
738                            scope_id: ScopeId::from("scope-1"),
739                        },
740                        actor: Actor::user(
741                            UserId::from("user-1"),
742                            Some(ClientId::from("client-1")),
743                        ),
744                        surface_id: ConversationSurface::client_personal("user-1").unwrap(),
745                        thread_id: ThreadId::from("thread-1"),
746                        run_id: RunId::from("run-1"),
747                        operation_id: OperationId::from("operation-1"),
748                        deadline_unix_ms: None,
749                        traceparent: None,
750                    },
751                    message: Message {
752                        continuation: None,
753                        role: MessageRole::User,
754                        content: vec![ContentPart::Text {
755                            text: "hello".to_owned(),
756                        }],
757                    },
758                    options: RunOptions {
759                        use_case: UseCase("chat".to_owned()),
760                        model_mode: ModelMode("auto".to_owned()),
761                        access_mode: AccessMode::Interactive,
762                        allow_tools: true,
763                        max_steps: 8,
764                        credit_budget: None,
765                        debug: false,
766                    },
767                },
768                context_generation: Some(ContextGeneration {
769                    user_scope: "scope-1".to_owned(),
770                    identity: "identity-1".to_owned(),
771                    memory: "memory-1".to_owned(),
772                    integration_guide: "guide-1".to_owned(),
773                    installed_integrations: "installed-1".to_owned(),
774                    scope_integrations: "scope-integrations-1".to_owned(),
775                }),
776                llm_plan: Some(LlmExecutionPlan {
777                    model_mode: ModelMode("auto".to_owned()),
778                    revision: Some("1700000000000".to_owned()),
779                    route: LlmRoute {
780                        backend: "openrouter".to_owned(),
781                        profile: "cloud_openrouter".to_owned(),
782                        model: "openai/gpt-5.4-mini".to_owned(),
783                        provider_preferences: vec!["OpenAI".to_owned()],
784                        temperature: Some(0.5),
785                        cache_control: Some(true),
786                        reasoning_effort: None,
787                        thinking: None,
788                        fast_mode: None,
789                        context_window_tokens: 128_000,
790                        model_snapshot: LlmModelSnapshot {
791                            capabilities: llm_api::ModelCapabilities {
792                                input: vec!["image".to_owned()],
793                                tool_calling: true,
794                                structured_output: true,
795                            },
796                            generation_support: LlmGenerationSupport {
797                                temperature: Some(true),
798                                max_tokens: Some(true),
799                                thinking: Some(true),
800                                reasoning_efforts: Some(vec!["high".into()]),
801                                temperature_with_reasoning: Some(true),
802                                ..LlmGenerationSupport::default()
803                            },
804                        },
805                    },
806                }),
807                base_checkpoint: None,
808                abandoned_run_id: None,
809            })),
810        };
811        assert_eq!(envelope, expected);
812        assert_eq!(
813            serde_json::to_value(envelope).expect("serializes"),
814            serde_json::from_str::<serde_json::Value>(fixture).expect("fixture JSON")
815        );
816    }
817
818    #[test]
819    fn snapshot_is_required_and_support_metadata_is_strict() {
820        let mut fixture: serde_json::Value = serde_json::from_str(ENQUEUE_FIXTURE).unwrap();
821        fixture["payload"]["payload"]["llm_plan"]["route"]
822            .as_object_mut()
823            .unwrap()
824            .remove("model_snapshot");
825        assert!(Envelope::decode_json(&fixture.to_string()).is_err());
826        let mut envelope = Envelope::decode_json(ENQUEUE_FIXTURE).unwrap();
827        let Payload::Command(command) = &mut envelope.payload else {
828            panic!("command");
829        };
830        command
831            .llm_plan
832            .as_mut()
833            .unwrap()
834            .route
835            .model_snapshot
836            .generation_support
837            .max_output_tokens = Some(0);
838        assert!(envelope.validate().is_err());
839    }
840
841    #[test]
842    fn generation_support_travels_in_catalog_spelling() {
843        let support = LlmGenerationSupport {
844            temperature: Some(true),
845            max_tokens: Some(true),
846            thinking: Some(true),
847            reasoning_efforts: Some(vec!["low".into(), "high".into()]),
848            reasoning_effort_default: Some("high".into()),
849            temperature_with_reasoning: Some(false),
850            temperature_max: Some(2.0),
851            max_output_tokens: Some(8_192),
852            fast_mode: Some(true),
853        };
854        let encoded = serde_json::to_value(&support).unwrap();
855        assert_eq!(
856            encoded["reasoningEfforts"],
857            serde_json::json!(["low", "high"])
858        );
859        assert_eq!(
860            encoded["temperatureWithReasoning"],
861            serde_json::json!(false)
862        );
863        assert_eq!(
864            serde_json::from_value::<LlmGenerationSupport>(encoded).unwrap(),
865            support
866        );
867        let ignored = serde_json::from_value::<LlmGenerationSupport>(serde_json::json!({
868            "temperature": true, "maxTokens": true, "reasoningEfforts": ["low"],
869            "temperatureWithReasoning": null, "temperatureMax": null, "maxOutputTokens": null,
870            "reasoningIntensity": "high"
871        }))
872        .unwrap();
873        assert_eq!(ignored.temperature, Some(true));
874        assert_eq!(
875            ignored.reasoning_efforts.as_deref(),
876            Some(["low".to_string()].as_slice())
877        );
878    }
879
880    fn dispatch() -> DispatchBinding {
881        DispatchBinding {
882            dispatch_id: "dispatch-1".to_owned(),
883            agent_session_id: "agent-session-1".to_owned(),
884        }
885    }
886
887    fn checkpoint() -> CheckpointReference {
888        CheckpointReference {
889            object_key: "checkpoints/abc.json".to_owned(),
890            sha256: "a".repeat(64),
891            format_version: 1,
892        }
893    }
894
895    #[test]
896    fn every_v1_golden_fixture_round_trips() {
897        let schema: serde_json::Value = serde_json::from_str(include_str!(
898            "../../../schema/execution/envelope.v1.schema.json"
899        ))
900        .expect("valid JSON Schema document");
901        let validator = jsonschema::validator_for(&schema).expect("valid JSON Schema semantics");
902        for fixture in [
903            include_str!("../../../fixtures/execution/enqueue.v1.json"),
904            include_str!("../../../fixtures/execution/completed-event.v1.json"),
905            include_str!("../../../fixtures/execution/interaction-required-event.v1.json"),
906            include_str!("../../../fixtures/execution/failed-event.v1.json"),
907            include_str!("../../../fixtures/execution/app-facade-prepare.v1.json"),
908            include_str!("../../../fixtures/execution/observation.v1.json"),
909            include_str!("../../../fixtures/execution/admission.v1.json"),
910            include_str!("../../../fixtures/execution/app-facade-prepared.v1.json"),
911            include_str!("../../../fixtures/execution/app-facade-user-action.v1.json"),
912            include_str!("../../../fixtures/execution/event-ack.v1.json"),
913            include_str!("../../../fixtures/execution/checkpoint-purge.v1.json"),
914            include_str!("../../../fixtures/execution/checkpoint-purge-ack.v1.json"),
915        ] {
916            let json: serde_json::Value =
917                serde_json::from_str(fixture).expect("valid golden fixture JSON");
918            validator
919                .validate(&json)
920                .expect("golden fixture matches the normative schema");
921            let envelope = Envelope::decode_json(fixture).expect("fixture matches Rust DTOs");
922            assert_eq!(serde_json::to_value(envelope).expect("serializes"), json);
923        }
924    }
925
926    #[test]
927    fn rejects_blank_scoped_identifiers() {
928        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
929        let mut envelope: Envelope = serde_json::from_str(fixture).expect("valid golden fixture");
930        let Payload::Command(request) = &mut envelope.payload else {
931            panic!("enqueue fixture changed family");
932        };
933        let CommandRequest {
934            command: AgentCommand::Enqueue { context, .. },
935            ..
936        } = request.as_mut()
937        else {
938            panic!("enqueue fixture changed family");
939        };
940        context.scope.scope_id = ScopeId::from(" ");
941        assert_eq!(
942            envelope.validate(),
943            Err(ProtocolError::MissingIdentifier("scope_id"))
944        );
945    }
946
947    #[test]
948    fn strict_decoder_checks_other_commands_and_message_parts() {
949        let golden: serde_json::Value = serde_json::from_str(ENQUEUE_FIXTURE).unwrap();
950        let context = golden["payload"]["payload"]["command"]["context"].clone();
951        for command in [
952            serde_json::json!({"type": "cancel", "context": context}),
953            serde_json::json!({"type": "submit_interaction", "context": context,
954                "interaction": {"batch_id": "batch-1", "decisions": [{"action_id": "action-1", "proceed": true}]}}),
955        ] {
956            let mut input = golden.clone();
957            if command["type"] == "cancel" {
958                input["payload"]["payload"]["llm_plan"] = serde_json::Value::Null;
959                input["payload"]["payload"]["context_generation"] = serde_json::Value::Null;
960            }
961            input["payload"]["payload"]["command"] = command;
962            Envelope::decode_json(&input.to_string()).expect("valid command");
963            input["payload"]["payload"]["command"]["unexpected"] = serde_json::Value::Null;
964            assert!(Envelope::decode_json(&input.to_string()).is_err());
965        }
966        for part in [
967            serde_json::json!({"type": "text", "text": "hello"}),
968            serde_json::json!({"type": "artifact", "uri": crate::execution::ArtifactReference::new("tenant-1", "scope-1", "a".repeat(64), crate::execution::ArtifactMetadata::image("image/png", 1, 1, 1).unwrap()).unwrap().uri().unwrap()}),
969            serde_json::json!({"type": "tool_call", "id": "call-1", "name": "test", "arguments": {"arbitrary": {"nested": null}}}),
970            serde_json::json!({"type": "tool_result", "call_id": "call-1", "result": {"arbitrary": true}, "is_error": false}),
971        ] {
972            let mut input = golden.clone();
973            input["payload"]["payload"]["command"]["message"]["content"][0] = part;
974            Envelope::decode_json(&input.to_string()).expect("valid content and opaque JSON");
975            input["payload"]["payload"]["command"]["message"]["content"][0]["unexpected"] =
976                serde_json::Value::Bool(true);
977            assert!(Envelope::decode_json(&input.to_string()).is_err());
978        }
979        assert!(
980            serde_json::from_value::<crate::execution::AgentEvent>(
981                serde_json::json!({"type": "started", "unexpected": true})
982            )
983            .is_err()
984        );
985        assert!(
986            serde_json::from_value::<AppFacadeResponse>(
987                serde_json::json!({"type": "rejected", "unexpected": null})
988            )
989            .is_err()
990        );
991        assert!(
992            serde_json::from_value::<crate::execution::AgentEvent>(
993                serde_json::json!({"type": "started"})
994            )
995            .is_ok()
996        );
997        assert!(
998            serde_json::from_value::<AppFacadeResponse>(serde_json::json!({"type": "rejected"}))
999                .is_ok()
1000        );
1001    }
1002
1003    #[test]
1004    fn decoder_checks_every_closed_object_in_golden_envelopes() {
1005        fn object_paths(value: &serde_json::Value, pointer: &str, paths: &mut Vec<String>) {
1006            match value {
1007                serde_json::Value::Object(fields) => {
1008                    paths.push(pointer.to_owned());
1009                    for (key, child) in fields {
1010                        let key = key.replace('~', "~0").replace('/', "~1");
1011                        object_paths(child, &format!("{pointer}/{key}"), paths);
1012                    }
1013                }
1014                serde_json::Value::Array(items) => {
1015                    for (index, child) in items.iter().enumerate() {
1016                        object_paths(child, &format!("{pointer}/{index}"), paths);
1017                    }
1018                }
1019                _ => {}
1020            }
1021        }
1022
1023        let schema: serde_json::Value = serde_json::from_str(include_str!(
1024            "../../../schema/execution/envelope.v1.schema.json"
1025        ))
1026        .unwrap();
1027        let validator = jsonschema::validator_for(&schema).unwrap();
1028        let fixtures = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("fixtures/execution");
1029        let mut accepted_unknown = Vec::new();
1030        let mut checked = 0;
1031        for file in std::fs::read_dir(fixtures).unwrap() {
1032            let file = file.unwrap().path();
1033            if file.extension().is_none_or(|extension| extension != "json") {
1034                continue;
1035            }
1036            let input = std::fs::read_to_string(&file).unwrap();
1037            let golden: serde_json::Value = serde_json::from_str(&input).unwrap();
1038            validator
1039                .validate(&golden)
1040                .expect("valid canonical fixture");
1041            Envelope::decode_json(&input).expect("valid fixture decodes");
1042            let mut paths = Vec::new();
1043            object_paths(&golden, "", &mut paths);
1044            for pointer in paths {
1045                for extra in [serde_json::Value::Bool(true), serde_json::Value::Null] {
1046                    let mut input = golden.clone();
1047                    input
1048                        .pointer_mut(&pointer)
1049                        .unwrap()
1050                        .as_object_mut()
1051                        .unwrap()
1052                        .insert("__unknown_probe".to_owned(), extra);
1053                    let decoded = Envelope::decode_json(&input.to_string());
1054                    if validator.is_valid(&input) {
1055                        decoded.expect("protocol-owned opaque JSON must remain extensible");
1056                    } else {
1057                        checked += 1;
1058                        if decoded.is_ok() {
1059                            accepted_unknown.push(format!(
1060                                "{}:{pointer}",
1061                                file.file_name().unwrap().to_string_lossy()
1062                            ));
1063                        }
1064                    }
1065                }
1066            }
1067        }
1068        assert!(checked > 100, "exercise the entire fixture graph");
1069        assert!(
1070            accepted_unknown.is_empty(),
1071            "unknown fields accepted at {accepted_unknown:#?}"
1072        );
1073    }
1074
1075    #[test]
1076    fn admission_requires_one_model_matching_the_selected_mode() {
1077        let mut input: serde_json::Value =
1078            serde_json::from_str(include_str!("../../../fixtures/execution/enqueue.v1.json"))
1079                .unwrap();
1080        input["payload"]["payload"]["llm_plan"]["model_mode"] = "smart".into();
1081        assert!(matches!(
1082            Envelope::decode_json(&input.to_string()),
1083            Err(ProtocolError::InvalidBinding(_))
1084        ));
1085        input["payload"]["payload"]["command"]["options"]["model_mode"] = "smart".into();
1086        assert!(Envelope::decode_json(&input.to_string()).is_ok());
1087        let plan = input["payload"]["payload"]["llm_plan"].take();
1088        input["payload"]["payload"]["llm_plan"] = serde_json::json!({"routes": [plan]});
1089        assert!(Envelope::decode_json(&input.to_string()).is_err());
1090    }
1091
1092    #[test]
1093    fn strict_decoder_rejects_unknown_nested_fields() {
1094        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1095        let input = fixture.replacen(
1096            "\"sent_at_unix_ms\": 1700000000000",
1097            "\"sent_at_unix_ms\": 1700000000000, \"unexpected\": true",
1098            1,
1099        );
1100        assert!(matches!(
1101            Envelope::decode_json(&input),
1102            Err(ProtocolError::UnknownField { .. })
1103        ));
1104
1105        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1106        input["payload"]["payload"]["options"]["unexpected"] = serde_json::Value::Bool(true);
1107        assert!(matches!(
1108            Envelope::decode_json(&input.to_string()),
1109            Err(ProtocolError::UnknownField { .. })
1110        ));
1111    }
1112
1113    #[test]
1114    fn strict_decoder_treats_skip_optional_nulls_as_absent() {
1115        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1116        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1117        input["payload"]["payload"]["llm_plan"]["route"]["fast_mode"] = serde_json::Value::Null;
1118        input["payload"]["payload"]["llm_plan"]["route"]["thinking"] = serde_json::Value::Null;
1119        input["payload"]["payload"]["llm_plan"]["route"]["model_snapshot"]["generation_support"]
1120            ["fastMode"] = serde_json::Value::Null;
1121        input["payload"]["payload"]["llm_plan"]["route"]["model_snapshot"]["generation_support"]
1122            ["reasoningEffortDefault"] = serde_json::Value::Null;
1123        input["payload"]["payload"]["command"]["message"]["continuation"] = serde_json::Value::Null;
1124
1125        let envelope = Envelope::decode_json(&input.to_string())
1126            .expect("null on skip-optional fields deserializes as absent");
1127        let canonical = serde_json::to_value(&envelope).expect("serializes");
1128        let canonical_route = &canonical["payload"]["payload"]["llm_plan"]["route"];
1129        assert!(canonical_route.get("fast_mode").is_none());
1130        assert!(canonical_route.get("thinking").is_none());
1131        let canonical_support = &canonical_route["model_snapshot"]["generation_support"];
1132        assert!(canonical_support.get("fastMode").is_none());
1133        assert!(canonical_support.get("reasoningEffortDefault").is_none());
1134        assert!(
1135            canonical["payload"]["payload"]["command"]["message"]
1136                .get("continuation")
1137                .is_none()
1138        );
1139
1140        let schema: serde_json::Value = serde_json::from_str(include_str!(
1141            "../../../schema/execution/envelope.v1.schema.json"
1142        ))
1143        .expect("schema");
1144        let validator = jsonschema::validator_for(&schema).expect("schema");
1145        assert!(
1146            validator.validate(&input).is_err(),
1147            "canonical emit omits skip-optional keys; schema must not advertise null"
1148        );
1149    }
1150
1151    #[test]
1152    fn schema_rejects_generation_support_and_continuation_nulls() {
1153        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1154        let schema: serde_json::Value = serde_json::from_str(include_str!(
1155            "../../../schema/execution/envelope.v1.schema.json"
1156        ))
1157        .expect("schema");
1158        let validator = jsonschema::validator_for(&schema).expect("schema");
1159        let golden: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1160        assert!(
1161            validator.validate(&golden).is_ok(),
1162            "golden enqueue fixture must remain emit-valid"
1163        );
1164
1165        let mut support_null = golden.clone();
1166        support_null["payload"]["payload"]["llm_plan"]["route"]["model_snapshot"]["generation_support"]
1167            ["fastMode"] = serde_json::Value::Null;
1168        assert!(
1169            validator.validate(&support_null).is_err(),
1170            "skip-optional generation_support.fastMode must omit, not null"
1171        );
1172
1173        let mut continuation_null = golden;
1174        continuation_null["payload"]["payload"]["command"]["message"]["continuation"] =
1175            serde_json::Value::Null;
1176        assert!(
1177            validator.validate(&continuation_null).is_err(),
1178            "skip-optional message.continuation must omit, not null"
1179        );
1180    }
1181
1182    #[test]
1183    fn strict_decoder_still_rejects_unknown_null_fields() {
1184        let fixture = include_str!("../../../fixtures/execution/enqueue.v1.json");
1185        let mut input: serde_json::Value = serde_json::from_str(fixture).expect("fixture JSON");
1186        input["payload"]["payload"]["llm_plan"]["route"]["unexpected"] = serde_json::Value::Null;
1187        assert!(matches!(
1188            Envelope::decode_json(&input.to_string()),
1189            Err(ProtocolError::UnknownField { .. })
1190        ));
1191    }
1192
1193    #[test]
1194    fn abandoned_run_requires_a_base_checkpoint() {
1195        let mut value: serde_json::Value =
1196            serde_json::from_str(include_str!("../../../fixtures/execution/enqueue.v1.json"))
1197                .unwrap();
1198        value["payload"]["payload"]["abandoned_run_id"] =
1199            serde_json::Value::String("run-abandoned".to_owned());
1200
1201        assert!(matches!(
1202            Envelope::decode_json(&value.to_string()),
1203            Err(ProtocolError::InvalidBinding(_))
1204        ));
1205    }
1206}
conversation_api/execution/wire/mod.rs

conversation_api/execution/wire/
mod.rs
