liminal_cli/network/
flow.rs

1//! `flow` (SDK-017 R4): steps fired in order, each a routed request fed by
2//! the previous reply through `{"$prev": "<JSON pointer>"}` objects; the
3//! first failure stops the flow by index.
4
5use serde_json::{Map, Value, json};
6
7use super::Rail;
8use super::call::{Outcome, request};
9use crate::mcp::tools::ToolResult;
10
11const PREV: &str = "$prev";
12
13/// One step as the host wrote it.
14struct Step<'a> {
15    subject: &'a str,
16    input: &'a Value,
17    reply: bool,
18    pass_for: Option<&'a str>,
19    /// Whether its reply is answered (SRV-052 R3); it feeds `$prev` either way.
20    keep: bool,
21}
22
23fn step(value: &Value, index: usize) -> Result<Step<'_>, String> {
24    let subject = value
25        .get("subject")
26        .and_then(Value::as_str)
27        .ok_or_else(|| format!("step {index} is missing subject"))?;
28    let Some(input @ Value::Object(_)) = value.get("input") else {
29        return Err(format!("step {index}'s input is missing or not an object"));
30    };
31    let reply = value.get("reply").and_then(Value::as_bool).unwrap_or(true);
32    let pass_for = match value.get("pass_for") {
33        None => None,
34        Some(Value::String(audience)) => Some(audience.as_str()),
35        Some(_) => return Err(format!("step {index}'s pass_for is not a string")),
36    };
37    let keep = match value.get("keep") {
38        None => true,
39        Some(Value::Bool(keep)) => *keep,
40        Some(_) => return Err(format!("step {index}'s keep is not a boolean")),
41    };
42    Ok(Step {
43        subject,
44        input,
45        reply,
46        pass_for,
47        keep,
48    })
49}
50
51/// `input` with every `{"$prev": pointer}` object replaced from `prev`.
52///
53/// # Errors
54/// Returns the pointer that names nothing in `prev`.
55pub fn fill(input: &Value, prev: Option<&Value>) -> Result<Value, String> {
56    match input {
57        Value::Object(object) => {
58            if let (1, Some(Value::String(pointer))) = (object.len(), object.get(PREV)) {
59                let prev = prev.ok_or_else(|| format!("{PREV} {pointer:?} in the first step"))?;
60                return prev.pointer(pointer).cloned().ok_or_else(|| {
61                    format!("{PREV} {pointer:?} names nothing in the previous reply")
62                });
63            }
64            let mut filled = Map::with_capacity(object.len());
65            for (key, value) in object {
66                filled.insert(key.clone(), fill(value, prev)?);
67            }
68            Ok(Value::Object(filled))
69        }
70        Value::Array(items) => items
71            .iter()
72            .map(|item| fill(item, prev))
73            .collect::<Result<_, _>>()
74            .map(Value::Array),
75        other => Ok(other.clone()),
76    }
77}
78
79/// The `flow` tool over `steps`, answering every kept reply.
80pub fn flow<R: Rail>(rail: &R, steps: &[Value]) -> ToolResult {
81    flow_answering(rail, steps, false)
82}
83
84/// A step's entry in the answer: its reply when kept, else its size, so a
85/// large intermediate reply feeds the next step without coming back
86/// (SRV-052 R3).
87fn entry(subject: &str, reply: &Value, kept: bool) -> Value {
88    if kept {
89        json!({ "subject": subject, "reply": reply })
90    } else {
91        let bytes = serde_json::to_vec(reply).map_or(0, |bytes| bytes.len());
92        json!({ "subject": subject, "kept": false, "bytes": bytes })
93    }
94}
95
96/// The `flow` tool over `steps`, at most the rail's configured count; with
97/// `last`, only the last step's reply is answered, as `reply`.
98pub fn flow_answering<R: Rail>(rail: &R, steps: &[Value], last: bool) -> ToolResult {
99    let limit = rail.flow_steps();
100    if steps.is_empty() {
101        return ToolResult::failed("the flow has no steps", json!({ "steps": [] }));
102    }
103    if steps.len() > limit {
104        return ToolResult::failed(
105            &format!(
106                "the flow has {} steps, over the limit of {limit}",
107                steps.len()
108            ),
109            json!({ "steps": [] }),
110        );
111    }
112    let mut answered: Vec<Value> = Vec::with_capacity(steps.len());
113    let mut prev: Option<Value> = None;
114    for (index, value) in steps.iter().enumerate() {
115        let failed = |words: String, answered: &[Value]| {
116            ToolResult::failed(
117                &format!("step {index} stopped the flow: {words}"),
118                json!({ "failed_step": index, "steps": answered }),
119            )
120        };
121        let step = match step(value, index) {
122            Ok(step) => step,
123            Err(words) => return failed(words, &answered),
124        };
125        let input = match fill(step.input, prev.as_ref()) {
126            Ok(input) => input,
127            Err(words) => return failed(words, &answered),
128        };
129        match request(rail, step.subject, &input, step.reply, step.pass_for) {
130            Outcome::Reply(reply) => {
131                answered.push(entry(step.subject, &reply, step.keep && !last));
132                prev = Some(reply);
133            }
134            Outcome::Sent => {
135                answered.push(json!({ "subject": step.subject, "sent": true }));
136                prev = Some(Value::Null);
137            }
138            Outcome::Refused(words) => {
139                return failed(format!("{}: {words}", step.subject), &answered);
140            }
141        }
142    }
143    if last {
144        return ToolResult::ok(json!({ "steps": answered, "reply": prev.unwrap_or(Value::Null) }));
145    }
146    ToolResult::ok(json!({ "steps": answered }))
147}
148
149#[cfg(test)]
150#[path = "flow_tests.rs"]
151mod tests;
liminal_cli/network/flow.rs

liminal_cli/network/
flow.rs
