robonix_executor/
verification.rs

1// SPDX-License-Identifier: MulanPSL-2.0
2//
3// Executor-owned result verification. A configured rule turns one successful
4// capability result into a synchronous call to a provider implementing the
5// shared `robonix/service/verifier/verify` contract.
6
7use std::time::Duration;
8
9use serde::Deserialize;
10
11use crate::config::VerificationRule;
12use crate::dispatch;
13use crate::pb::pilot::{CapabilityCall, CapabilityCallResult};
14use crate::plan_runtime::PlanRuntime;
15use crate::rtdl_wire::NodeEventContext;
16use robonix_atlas::client::AtlasClient;
17use robonix_scribe::{info, warn};
18
19pub const VERIFY_CONTRACT_ID: &str = "robonix/service/verifier/verify";
20const VERIFICATION_TIMEOUT: Duration = Duration::from_secs(60);
21
22/// Immutable rule set shared by all concurrently executing plans.
23#[derive(Clone, Debug, Default)]
24pub struct VerificationPolicy {
25    rules: Vec<VerificationRule>,
26}
27
28impl VerificationPolicy {
29    pub fn new(rules: Vec<VerificationRule>) -> Self {
30        Self { rules }
31    }
32
33    pub fn len(&self) -> usize {
34        self.rules.len()
35    }
36
37    /// Exact provider+contract rules override a contract-only fallback.
38    pub fn rule_for(&self, call: &CapabilityCall) -> Option<&VerificationRule> {
39        self.rules
40            .iter()
41            .find(|rule| {
42                rule.target_contract_id == call.contract_id
43                    && rule.target_provider_id.as_deref() == Some(call.provider_id.as_str())
44            })
45            .or_else(|| {
46                self.rules.iter().find(|rule| {
47                    rule.target_contract_id == call.contract_id && rule.target_provider_id.is_none()
48                })
49            })
50    }
51}
52
53#[derive(Debug, Deserialize)]
54struct VerifyResponse {
55    passed: bool,
56    detail: String,
57}
58
59/// Verify a successful result when a rule matches. The verifier call itself
60/// goes straight through dispatch and is never recursively verified.
61pub async fn verify_result(
62    policy: &VerificationPolicy,
63    call: &CapabilityCall,
64    node: &NodeEventContext,
65    original: CapabilityCallResult,
66    self_provider_id: &str,
67    atlas: &mut AtlasClient,
68    runtime: &PlanRuntime,
69) -> CapabilityCallResult {
70    let Some(rule) = policy.rule_for(call) else {
71        return original;
72    };
73
74    let verifier_call = build_verifier_call(rule, call, &node.description, &original);
75    info!(
76        "[executor/verification] call_id={} target='{}' verifier='{}'",
77        call.call_id, call.contract_id, rule.verifier_provider_id
78    );
79    let response = dispatch::dispatch_with_timeout(
80        &verifier_call,
81        self_provider_id,
82        atlas,
83        runtime,
84        &node.plan_id,
85        Some(VERIFICATION_TIMEOUT),
86    )
87    .await;
88    if !response.success {
89        return unavailable(
90            original,
91            format!(
92                "verifier '{}' failed: {}",
93                rule.verifier_provider_id, response.error
94            ),
95        );
96    }
97
98    let parsed: VerifyResponse = match serde_json::from_str(&response.output) {
99        Ok(parsed) => parsed,
100        Err(error) => {
101            return unavailable(
102                original,
103                format!(
104                    "verifier '{}' returned invalid response: {error}",
105                    rule.verifier_provider_id
106                ),
107            );
108        }
109    };
110    apply_verdict(original, parsed)
111}
112
113/// Build the common two-field Verify request. The nested payload stays opaque
114/// to Executor beyond carrying the original call context and configured args.
115fn build_verifier_call(
116    rule: &VerificationRule,
117    call: &CapabilityCall,
118    description: &str,
119    result: &CapabilityCallResult,
120) -> CapabilityCall {
121    let payload = serde_json::json!({
122        "target_provider_id": call.provider_id,
123        "target_contract_id": call.contract_id,
124        "target_description": description,
125        "target_args": json_or_string(&call.args_json),
126        "target_output": json_or_string(&result.output),
127        "verifier_args": rule.verifier_args,
128    });
129    CapabilityCall {
130        call_id: format!("{}:verify", call.call_id),
131        provider_id: rule.verifier_provider_id.clone(),
132        contract_id: VERIFY_CONTRACT_ID.to_string(),
133        args_json: serde_json::json!({
134            "call_id": call.call_id,
135            "args_json": payload.to_string(),
136        })
137        .to_string(),
138    }
139}
140
141fn json_or_string(raw: &str) -> serde_json::Value {
142    serde_json::from_str(raw).unwrap_or_else(|_| serde_json::Value::String(raw.to_string()))
143}
144
145fn apply_verdict(
146    mut original: CapabilityCallResult,
147    verdict: VerifyResponse,
148) -> CapabilityCallResult {
149    if verdict.passed {
150        info!(
151            "[executor/verification] call_id={} passed: {}",
152            original.call_id, verdict.detail
153        );
154        return original;
155    }
156    let detail = if verdict.detail.trim().is_empty() {
157        "verifier rejected the capability result".to_string()
158    } else {
159        verdict.detail
160    };
161    warn!(
162        "[executor/verification] call_id={} rejected: {}",
163        original.call_id, detail
164    );
165    original.success = false;
166    original.error = format!("result verification failed: {detail}");
167    original
168}
169
170fn unavailable(mut original: CapabilityCallResult, reason: String) -> CapabilityCallResult {
171    warn!(
172        "[executor/verification] call_id={} unavailable: {}",
173        original.call_id, reason
174    );
175    original.success = false;
176    original.error = format!("result verification unavailable: {reason}");
177    original
178}
179
180#[cfg(test)]
181mod tests {
182    use super::*;
183    use robonix_atlas::service::{AtlasRegistry, serve_atlas};
184    use std::net::SocketAddr;
185    use std::sync::Arc;
186
187    fn rule(provider: Option<&str>, verifier: &str) -> VerificationRule {
188        VerificationRule {
189            target_contract_id: "cap/target".to_string(),
190            target_provider_id: provider.map(str::to_string),
191            verifier_provider_id: verifier.to_string(),
192            verifier_args: serde_json::json!({"camera_provider_id":"front_camera"}),
193        }
194    }
195
196    fn call() -> CapabilityCall {
197        CapabilityCall {
198            call_id: "p:0".to_string(),
199            provider_id: "arm".to_string(),
200            contract_id: "cap/target".to_string(),
201            args_json: r#"{"object":"bottle"}"#.to_string(),
202        }
203    }
204
205    fn result() -> CapabilityCallResult {
206        CapabilityCallResult {
207            call_id: "p:0".to_string(),
208            provider_id: "arm".to_string(),
209            contract_id: "cap/target".to_string(),
210            success: true,
211            output: r#"{"accepted":true}"#.to_string(),
212            error: String::new(),
213        }
214    }
215
216    fn node() -> NodeEventContext {
217        NodeEventContext {
218            plan_id: "p".to_string(),
219            node_index: 0,
220            node_kind: 0,
221            op_id: "op".to_string(),
222            description: "place bottle".to_string(),
223        }
224    }
225
226    fn reserve_address() -> SocketAddr {
227        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve port");
228        listener.local_addr().expect("reserved address")
229    }
230
231    #[test]
232    fn exact_provider_rule_wins_over_contract_fallback() {
233        let policy =
234            VerificationPolicy::new(vec![rule(None, "fallback"), rule(Some("arm"), "exact")]);
235        assert_eq!(
236            policy.rule_for(&call()).unwrap().verifier_provider_id,
237            "exact"
238        );
239    }
240
241    #[test]
242    fn verifier_request_contains_runtime_context_and_opaque_args() {
243        let verify = build_verifier_call(
244            &rule(None, "vlm_verifier"),
245            &call(),
246            "place bottle",
247            &result(),
248        );
249        let outer: serde_json::Value = serde_json::from_str(&verify.args_json).unwrap();
250        let payload: serde_json::Value =
251            serde_json::from_str(outer["args_json"].as_str().unwrap()).unwrap();
252        assert_eq!(verify.contract_id, VERIFY_CONTRACT_ID);
253        assert_eq!(payload["target_args"]["object"], "bottle");
254        assert_eq!(payload["target_output"]["accepted"], true);
255        assert_eq!(payload["target_description"], "place bottle");
256        assert_eq!(
257            payload["verifier_args"]["camera_provider_id"],
258            "front_camera"
259        );
260    }
261
262    #[test]
263    fn rejected_verdict_preserves_output_and_fails_original_call() {
264        let output = result().output;
265        let verified = apply_verdict(
266            result(),
267            VerifyResponse {
268                passed: false,
269                detail: "bottle is outside the box".to_string(),
270            },
271        );
272        assert!(!verified.success);
273        assert_eq!(verified.output, output);
274        assert!(verified.error.contains("result verification failed"));
275    }
276
277    #[test]
278    fn passed_verdict_keeps_original_result_unchanged() {
279        let original = result();
280        let verified = apply_verdict(
281            original.clone(),
282            VerifyResponse {
283                passed: true,
284                detail: "visible".to_string(),
285            },
286        );
287        assert_eq!(verified, original);
288    }
289
290    #[test]
291    fn unavailable_verifier_preserves_output_but_fails_closed() {
292        let output = result().output;
293        let verified = unavailable(result(), "not registered".to_string());
294        assert!(!verified.success);
295        assert_eq!(verified.output, output);
296        assert!(verified.error.contains("result verification unavailable"));
297    }
298
299    #[tokio::test]
300    async fn missing_verifier_provider_fails_closed_over_atlas() {
301        let atlas_addr = reserve_address();
302        let registry = Arc::new(AtlasRegistry::default());
303        let atlas_server = tokio::spawn(serve_atlas(registry, atlas_addr));
304        let mut atlas = AtlasClient::connect_with_retry(
305            format!("http://{atlas_addr}"),
306            50,
307            Duration::from_millis(10),
308        )
309        .await
310        .expect("connect test Atlas");
311        let policy = VerificationPolicy::new(vec![rule(None, "missing_verifier")]);
312        let verified = verify_result(
313            &policy,
314            &call(),
315            &node(),
316            result(),
317            "executor",
318            &mut atlas,
319            &PlanRuntime::default(),
320        )
321        .await;
322        assert!(!verified.success);
323        assert!(verified.error.contains("result verification unavailable"));
324        assert!(verified.error.contains("ConnectCapability failed"));
325        atlas_server.abort();
326    }
327}
robonix_executor/verification.rs

robonix_executor/
verification.rs
