meerkat_runtime/
peer_handling_mode.rs

1//! Peer handling-mode validation — reject handling_mode on response progress conventions.
2//!
3//! Rules:
4//! - handling_mode is FORBIDDEN for: PeerInput(ResponseProgress)
5//! - handling_mode is ALLOWED for: PeerInput(Message), PeerInput(Request), PeerInput(ResponseTerminal), PeerInput(no convention)
6//! - Steer handling_mode is FORBIDDEN on peer inputs carrying injected
7//!   context: steer realization stages live system-context appends only, so
8//!   the injected-context transcript appends would be silently dropped
9//! - Non-peer inputs are always accepted (validation is a no-op)
10
11use crate::input::{Input, PeerConvention};
12use meerkat_core::types::HandlingMode;
13
14/// Errors from peer handling-mode validation.
15#[derive(Debug, Clone, thiserror::Error)]
16#[non_exhaustive]
17pub enum PeerHandlingModeError {
18    /// handling_mode is not allowed on ResponseProgress peer inputs.
19    #[error("handling_mode is forbidden on ResponseProgress peer inputs")]
20    ForbiddenForResponseProgress,
21    /// Steer deliveries realize as live system-context appends, which carry
22    /// no transcript boundary for injected context to precede. Fail closed
23    /// rather than silently dropping host-attached context.
24    #[error("steer handling_mode cannot carry injected context on peer inputs")]
25    SteerCannotCarryInjectedContext,
26}
27
28/// Validate that a peer input does not carry handling_mode on response progress.
29pub fn validate_peer_handling_mode(input: &Input) -> Result<(), PeerHandlingModeError> {
30    let Input::Peer(peer) = input else {
31        return Ok(());
32    };
33    if peer.handling_mode == Some(HandlingMode::Steer) && !peer.injected_context.is_empty() {
34        return Err(PeerHandlingModeError::SteerCannotCarryInjectedContext);
35    }
36    if peer.handling_mode.is_none() {
37        return Ok(());
38    }
39    match &peer.convention {
40        Some(PeerConvention::ResponseProgress { .. }) => {
41            Err(PeerHandlingModeError::ForbiddenForResponseProgress)
42        }
43        _ => Ok(()),
44    }
45}
46
47#[cfg(test)]
48#[allow(clippy::unwrap_used)]
49mod tests {
50    use super::*;
51    use crate::input::*;
52    use chrono::Utc;
53    use meerkat_core::lifecycle::InputId;
54    use meerkat_core::types::HandlingMode;
55
56    fn make_header() -> InputHeader {
57        InputHeader {
58            id: InputId::new(),
59            timestamp: Utc::now(),
60            source: InputOrigin::Peer {
61                peer_id: "peer-1".into(),
62                display_identity: None,
63                runtime_id: None,
64            },
65            durability: InputDurability::Durable,
66            visibility: InputVisibility::default(),
67            idempotency_key: None,
68            supersession_key: None,
69            correlation_id: None,
70        }
71    }
72
73    #[test]
74    fn response_progress_with_handling_mode_rejected() {
75        let input = Input::Peer(PeerInput {
76            directed_interaction_id: None,
77            objective_id: None,
78            system_prompts: Vec::new(),
79            injected_context: Vec::new(),
80            sender_taint: None,
81            header: make_header(),
82            convention: Some(PeerConvention::ResponseProgress {
83                request_id: "r".into(),
84                phase: ResponseProgressPhase::InProgress,
85            }),
86            content: "working".into(),
87            payload: Some(serde_json::json!({"progress": "working"})),
88            handling_mode: Some(HandlingMode::Queue),
89        });
90        let err = validate_peer_handling_mode(&input).unwrap_err();
91        assert!(matches!(
92            err,
93            PeerHandlingModeError::ForbiddenForResponseProgress
94        ));
95    }
96
97    #[test]
98    fn response_terminal_with_handling_mode_accepted() {
99        let input = Input::Peer(PeerInput {
100            directed_interaction_id: None,
101            objective_id: None,
102            system_prompts: Vec::new(),
103            injected_context: Vec::new(),
104            sender_taint: None,
105            header: make_header(),
106            convention: Some(PeerConvention::ResponseTerminal {
107                request_id: "r".into(),
108                status: ResponseTerminalStatus::Completed,
109            }),
110            content: "done".into(),
111            payload: Some(serde_json::json!({"ok": true})),
112            handling_mode: Some(HandlingMode::Steer),
113        });
114        assert!(validate_peer_handling_mode(&input).is_ok());
115    }
116
117    #[test]
118    fn response_terminal_with_queue_handling_mode_accepted() {
119        let input = Input::Peer(PeerInput {
120            directed_interaction_id: None,
121            objective_id: None,
122            system_prompts: Vec::new(),
123            injected_context: Vec::new(),
124            sender_taint: None,
125            header: make_header(),
126            convention: Some(PeerConvention::ResponseTerminal {
127                request_id: "r".into(),
128                status: ResponseTerminalStatus::Completed,
129            }),
130            content: "done".into(),
131            payload: Some(serde_json::json!({"ok": true})),
132            handling_mode: Some(HandlingMode::Queue),
133        });
134        assert!(validate_peer_handling_mode(&input).is_ok());
135    }
136
137    #[test]
138    fn message_with_handling_mode_accepted() {
139        let input = Input::Peer(PeerInput {
140            directed_interaction_id: None,
141            objective_id: None,
142            system_prompts: Vec::new(),
143            injected_context: Vec::new(),
144            sender_taint: None,
145            header: make_header(),
146            convention: Some(PeerConvention::Message),
147            content: "hi".into(),
148            payload: None,
149            handling_mode: Some(HandlingMode::Queue),
150        });
151        assert!(validate_peer_handling_mode(&input).is_ok());
152    }
153
154    #[test]
155    fn request_with_handling_mode_accepted() {
156        let input = Input::Peer(PeerInput {
157            directed_interaction_id: None,
158            objective_id: None,
159            system_prompts: Vec::new(),
160            injected_context: Vec::new(),
161            sender_taint: None,
162            header: make_header(),
163            convention: Some(PeerConvention::Request {
164                request_id: "r".into(),
165                intent: "i".into(),
166            }),
167            content: "do it".into(),
168            payload: Some(serde_json::json!({"subject": "x"})),
169            handling_mode: Some(HandlingMode::Steer),
170        });
171        assert!(validate_peer_handling_mode(&input).is_ok());
172    }
173
174    #[test]
175    fn no_convention_with_handling_mode_accepted() {
176        let input = Input::Peer(PeerInput {
177            directed_interaction_id: None,
178            objective_id: None,
179            system_prompts: Vec::new(),
180            injected_context: Vec::new(),
181            sender_taint: None,
182            header: make_header(),
183            convention: None,
184            content: "hi".into(),
185            payload: None,
186            handling_mode: Some(HandlingMode::Queue),
187        });
188        assert!(validate_peer_handling_mode(&input).is_ok());
189    }
190
191    /// Steer realization stages live system-context appends only — injected
192    /// context would silently vanish. The accept boundary fails closed.
193    #[test]
194    fn steer_with_injected_context_rejected() {
195        let input = Input::Peer(PeerInput {
196            directed_interaction_id: None,
197            objective_id: None,
198            system_prompts: Vec::new(),
199            injected_context: vec![meerkat_core::types::ContentInput::Text(
200                "ambient".to_string(),
201            )],
202            sender_taint: None,
203            header: make_header(),
204            convention: Some(PeerConvention::Message),
205            content: "urgent".into(),
206            payload: None,
207            handling_mode: Some(HandlingMode::Steer),
208        });
209        let err = validate_peer_handling_mode(&input).unwrap_err();
210        assert!(matches!(
211            err,
212            PeerHandlingModeError::SteerCannotCarryInjectedContext
213        ));
214    }
215
216    /// Queue-mode deliveries stage at a run boundary, which carries the
217    /// transcript appends — injected context is deliverable there.
218    #[test]
219    fn queue_with_injected_context_accepted() {
220        let input = Input::Peer(PeerInput {
221            directed_interaction_id: None,
222            objective_id: None,
223            system_prompts: Vec::new(),
224            injected_context: vec![meerkat_core::types::ContentInput::Text(
225                "ambient".to_string(),
226            )],
227            sender_taint: None,
228            header: make_header(),
229            convention: Some(PeerConvention::Message),
230            content: "work".into(),
231            payload: None,
232            handling_mode: Some(HandlingMode::Queue),
233        });
234        assert!(validate_peer_handling_mode(&input).is_ok());
235    }
236
237    #[test]
238    fn peer_without_handling_mode_always_accepted() {
239        for convention in [
240            Some(PeerConvention::Message),
241            Some(PeerConvention::Request {
242                request_id: "r".into(),
243                intent: "i".into(),
244            }),
245            Some(PeerConvention::ResponseProgress {
246                request_id: "r".into(),
247                phase: ResponseProgressPhase::InProgress,
248            }),
249            Some(PeerConvention::ResponseTerminal {
250                request_id: "r".into(),
251                status: ResponseTerminalStatus::Completed,
252            }),
253            None,
254        ] {
255            let input = Input::Peer(PeerInput {
256                directed_interaction_id: None,
257                objective_id: None,
258                system_prompts: Vec::new(),
259                injected_context: Vec::new(),
260                sender_taint: None,
261                header: make_header(),
262                convention,
263                content: "hi".into(),
264                payload: None,
265                handling_mode: None,
266            });
267            assert!(
268                validate_peer_handling_mode(&input).is_ok(),
269                "should accept peer without handling_mode"
270            );
271        }
272    }
273
274    #[test]
275    fn non_peer_input_always_accepted() {
276        let input = Input::Prompt(PromptInput {
277            injected_context: Vec::new(),
278            header: InputHeader {
279                id: InputId::new(),
280                timestamp: Utc::now(),
281                source: InputOrigin::Operator,
282                durability: InputDurability::Durable,
283                visibility: InputVisibility::default(),
284                idempotency_key: None,
285                supersession_key: None,
286                correlation_id: None,
287            },
288            content: "hi".into(),
289            typed_turn_appends: Vec::new(),
290            turn_metadata: None,
291        });
292        assert!(validate_peer_handling_mode(&input).is_ok());
293    }
294}
meerkat_runtime/peer_handling_mode.rs

meerkat_runtime/
peer_handling_mode.rs
