sim_lib_bridge/
ask.rs

1use sim_codec::{Input, Output, decode_with_codec, encode_with_codec};
2use sim_codec_bridge::{
3    BridgeBook, BridgeCallArgument, BridgeCallPayload, BridgeHeader, BridgePacket, BridgePart,
4    BridgeProvenance, CallArgumentMedia, content_id_string, stamp_packet_cid,
5};
6use sim_kernel::{
7    Cx, Datum, EncodeOptions, EvalFabric, Expr, ReadPolicy, Result, Symbol, encode::EncodePosition,
8};
9use sim_lib_agent_runner_core::{
10    InjectionFence, ModelResponse, OutputContract, terminal_model_content,
11};
12use sim_shape::{check_value_report, shape_value};
13use sim_value::{access::field, build::entry};
14
15use crate::model::output_contract_for_packet;
16use crate::repair::{AskFailure, RepairPolicy};
17use crate::rx::{effective_caps, rx_check, shape_from_contract_expr};
18use crate::tx::{eval_request_for_checked_packet, prepare_packet};
19
20/// Default codec used to pack ASK call arguments and answers.
21pub fn ask_default_codec() -> Symbol {
22    Symbol::qualified("codec", "json")
23}
24
25/// Builds an ASK request packet with model parameters omitted.
26pub fn ask_packet(
27    cx: &mut Cx,
28    name: &str,
29    params: Vec<(String, Expr)>,
30    return_shape: Expr,
31    to: &str,
32) -> Result<BridgePacket> {
33    ask_packet_with_model_params(cx, name, params, Vec::new(), return_shape, to)
34}
35
36/// Builds an ASK request packet.
37///
38/// Argument values are encoded through the default ASK codec at data position,
39/// wrapped in deterministic injection fences, and only then stored in the
40/// packet.
41pub fn ask_packet_with_model_params(
42    cx: &mut Cx,
43    name: &str,
44    params: Vec<(String, Expr)>,
45    model_params: Vec<(String, Expr)>,
46    return_shape: Expr,
47    to: &str,
48) -> Result<BridgePacket> {
49    let codec = ask_default_codec();
50    let mut call = BridgeCallPayload::new(symbol_from_name(name));
51    for (name, value) in params {
52        call = call.with_arg(pack_argument(cx, &name, &codec, &value)?);
53    }
54    for (name, value) in model_params {
55        call = call.with_model_param(symbol_from_name(&name), value);
56    }
57    Ok(BridgePacket {
58        header: BridgeHeader {
59            cid: None,
60            move_kind: Symbol::new("request"),
61            from: "sim".to_owned(),
62            to: vec![to.to_owned()],
63            role: Symbol::new("implementer"),
64            parents: Vec::new(),
65            task: Symbol::new("C1"),
66            output: Symbol::new("O1"),
67            ceiling: vec![Symbol::qualified("ai", "run")],
68            context: Vec::new(),
69            provenance: BridgeProvenance::default(),
70        },
71        body: vec![
72            BridgePart {
73                id: Symbol::new("C1"),
74                kind: Symbol::qualified("bridge", "Call"),
75                payload: call.to_expr(),
76            },
77            BridgePart {
78                id: Symbol::new("O1"),
79                kind: Symbol::qualified("bridge", "Return"),
80                payload: Expr::Map(vec![
81                    entry("codec", Expr::Symbol(codec)),
82                    entry("shape", return_shape),
83                ]),
84            },
85        ],
86        warrant: None,
87    })
88}
89
90/// Runs an ASK packet with the default bounded repair policy.
91pub fn run_ask(cx: &mut Cx, target: &dyn EvalFabric, packet: BridgePacket) -> Result<BridgePacket> {
92    run_ask_with_policy(cx, target, packet, RepairPolicy::default())
93}
94
95/// Runs an ASK packet with an explicit bounded repair policy.
96pub fn run_ask_with_policy(
97    cx: &mut Cx,
98    target: &dyn EvalFabric,
99    mut packet: BridgePacket,
100    policy: RepairPolicy,
101) -> Result<BridgePacket> {
102    let book = BridgeBook::standard();
103    let max_retries = policy.retries();
104    for attempt in 0..=max_retries {
105        let checked = prepare_packet(cx, &book, &packet)?;
106        let request = eval_request_for_checked_packet(cx, &book, &checked)?;
107        let caps = effective_caps(cx, &checked)?;
108        let reply = cx.with_capabilities(caps, |cx| target.realize(cx, request))?;
109        let response = ModelResponse::try_from(reply.value.object().as_expr(cx)?)?;
110        match answer_packet(cx, &book, &checked, &response)? {
111            Ok(answer) => return Ok(answer),
112            Err(failure) if attempt < max_retries => {
113                packet = repair_packet_for_failure(cx, &checked, &failure, attempt + 1)?;
114            }
115            Err(failure) => {
116                return Err(sim_kernel::Error::Eval(format!(
117                    "bridge ask failed after {} attempt(s): {}",
118                    attempt + 1,
119                    failure.message()
120                )));
121            }
122        }
123    }
124    unreachable!("bounded ASK loop always returns inside the retry range")
125}
126
127pub(crate) fn pack_argument(
128    cx: &mut Cx,
129    name: &str,
130    codec: &Symbol,
131    value: &Expr,
132) -> Result<BridgeCallArgument> {
133    let output = encode_with_codec(
134        cx,
135        codec,
136        value,
137        EncodeOptions {
138            position: EncodePosition::Data,
139            ..EncodeOptions::default()
140        },
141    )?;
142    let (media, datum, body) = match output {
143        Output::Text(text) => (CallArgumentMedia::Text, Datum::String(text.clone()), text),
144        Output::Bytes(bytes) => (
145            CallArgumentMedia::Bytes,
146            Datum::Bytes(bytes.clone()),
147            hex_text(&bytes),
148        ),
149    };
150    let content_id = datum.content_id()?;
151    let fence = InjectionFence::for_content(&content_id);
152    Ok(BridgeCallArgument::new(
153        symbol_from_name(name),
154        codec.clone(),
155        media,
156        content_id_string(&content_id),
157        fence.wrap(name, &body),
158    ))
159}
160
161fn answer_packet(
162    cx: &mut Cx,
163    book: &BridgeBook,
164    parent: &BridgePacket,
165    response: &ModelResponse,
166) -> Result<std::result::Result<BridgePacket, AskFailure>> {
167    let contract = output_contract_for_packet(parent)?;
168    let answer = match decode_terminal_answer(cx, response, &contract)? {
169        Ok(answer) => answer,
170        Err(failure) => return Ok(Err(failure)),
171    };
172    if let Err(failure) = validate_answer(cx, &contract, &answer)? {
173        return Ok(Err(failure));
174    }
175    let packet = stamp_packet_cid(&BridgePacket {
176        header: BridgeHeader {
177            cid: None,
178            move_kind: Symbol::new("reply"),
179            from: parent
180                .header
181                .to
182                .first()
183                .cloned()
184                .unwrap_or_else(|| "model".to_owned()),
185            to: vec![parent.header.from.clone()],
186            role: Symbol::new("implementer"),
187            parents: vec![parent.header.cid.clone().unwrap_or_default()],
188            task: Symbol::new("A1"),
189            output: Symbol::new("A1"),
190            ceiling: Vec::new(),
191            context: Vec::new(),
192            provenance: BridgeProvenance::default(),
193        },
194        body: vec![BridgePart {
195            id: Symbol::new("A1"),
196            kind: Symbol::qualified("bridge", "Return"),
197            payload: answer,
198        }],
199        warrant: None,
200    })?;
201    let report = rx_check(cx, book, &packet, Some(parent))?;
202    if !report.accepted() {
203        return Err(sim_kernel::Error::Eval(format!(
204            "bridge ask reply failed rx check: {:?}",
205            report.obligations
206        )));
207    }
208    Ok(Ok(packet))
209}
210
211fn decode_terminal_answer(
212    cx: &mut Cx,
213    response: &ModelResponse,
214    contract: &OutputContract,
215) -> Result<std::result::Result<Expr, AskFailure>> {
216    let input = match terminal_model_content(response) {
217        Ok(Expr::String(text)) => Input::Text(text.clone()),
218        Ok(Expr::Bytes(bytes)) => Input::Bytes(bytes.clone()),
219        Ok(Expr::Map(_)) => match field(terminal_model_content(response)?, "text") {
220            Some(Expr::String(text)) => Input::Text(text.clone()),
221            _ => {
222                return Ok(Err(AskFailure::Decode {
223                    codec: contract.codec.clone(),
224                    message: "terminal content map must carry text".to_owned(),
225                }));
226            }
227        },
228        Ok(other) => {
229            return Ok(Err(AskFailure::Decode {
230                codec: contract.codec.clone(),
231                message: format!("terminal content must be text or bytes, found {other:?}"),
232            }));
233        }
234        Err(err) => {
235            return Ok(Err(AskFailure::Decode {
236                codec: contract.codec.clone(),
237                message: err.to_string(),
238            }));
239        }
240    };
241    match decode_with_codec(cx, &contract.codec, input, ReadPolicy::default()) {
242        Ok(answer) => Ok(Ok(answer)),
243        Err(err) => Ok(Err(AskFailure::Decode {
244            codec: contract.codec.clone(),
245            message: err.to_string(),
246        })),
247    }
248}
249
250fn validate_answer(
251    cx: &mut Cx,
252    contract: &OutputContract,
253    answer: &Expr,
254) -> Result<std::result::Result<(), AskFailure>> {
255    let Some(shape) = shape_from_contract_expr(&contract.shape_expr) else {
256        return Ok(Err(AskFailure::Shape {
257            expected: format!("{:?}", contract.shape_expr),
258            diagnostics: vec!["unsupported return Shape expression".to_owned()],
259        }));
260    };
261    let shape_ref = shape_value(Symbol::qualified("bridge", "AskReturn"), shape);
262    let value = cx.factory().expr(answer.clone())?;
263    let matched = check_value_report(cx, &shape_ref, value)?;
264    if matched.accepted {
265        Ok(Ok(()))
266    } else {
267        Ok(Err(AskFailure::Shape {
268            expected: format!("{:?}", contract.shape_expr),
269            diagnostics: matched
270                .diagnostics
271                .iter()
272                .map(|diagnostic| diagnostic.message.clone())
273                .collect(),
274        }))
275    }
276}
277
278fn repair_packet_for_failure(
279    cx: &mut Cx,
280    packet: &BridgePacket,
281    failure: &AskFailure,
282    attempt: u8,
283) -> Result<BridgePacket> {
284    let mut repaired = packet.canonicalized();
285    for part in &mut repaired.body {
286        if part.kind != Symbol::qualified("bridge", "Call") {
287            continue;
288        }
289        let payload = BridgeCallPayload::from_expr(&part.payload)?.with_arg(pack_argument(
290            cx,
291            &format!("repair-{attempt}"),
292            &ask_default_codec(),
293            &failure.to_expr(),
294        )?);
295        part.payload = payload.to_expr();
296        return Ok(repaired);
297    }
298    Err(sim_kernel::Error::Eval(
299        "bridge ask repair requires a Call part".to_owned(),
300    ))
301}
302
303fn symbol_from_name(name: &str) -> Symbol {
304    match name.split_once('/') {
305        Some((namespace, name)) if !namespace.is_empty() && !name.is_empty() => {
306            Symbol::qualified(namespace, name)
307        }
308        _ => Symbol::new(name),
309    }
310}
311
312fn hex_text(bytes: &[u8]) -> String {
313    bytes
314        .iter()
315        .map(|byte| format!("{byte:02x}"))
316        .collect::<String>()
317}
sim_lib_bridge/ask.rs

sim_lib_bridge/
ask.rs
