Vakyartha documentation

1//! Runs→flows adoption (docs/design/10-flows.md): convert proven work —
2//! planner plan ledgers, flow-run snapshots, or a session's settled shell
3//! commands — into a governed, hand-editable flow file. Provider/model are
4//! NEVER baked in from caller hints; attribution stays in the receipts.
5
6use crate::types::FlowState;
7
8#[derive(Debug, Clone)]
9pub struct AdoptedFlow {
10 pub name: String,
11 pub toml: String,
12 pub warnings: Vec<String>,
13}
14
15fn header_comment(source: &str, note: &str) -> String {
16 format!(
17 "# adopted by `flow adopt` from {source} at {}\n# {note}\n",
18 chrono::Utc::now().to_rfc3339(),
19 )
20}
21
22/// Adopt a frozen plan/flow-run snapshot: its definition TOML is already
23/// validated and proven — reuse it byte-for-byte under a new name.
24pub fn from_flow_state(
25 state_json: &str,
26 new_name: &str,
27 source_label: &str,
28) -> Result<AdoptedFlow, String> {
29 let state: FlowState =
30 serde_json::from_str(state_json).map_err(|e| format!("not a flow-run ledger: {e}"))?;
31 if state.nodes.is_empty() {
32 return Err("run has no node results to adopt".into());
33 }
34 let mut warnings = Vec::new();
35 // Rename inside the TOML so the adopted copy does not collide with the
36 // source flow's own identity.
37 let renamed = state.definition_toml.replacen(
38 &format!("name = \"{}\"", state.flow_name),
39 &format!("name = \"{new_name}\""),
40 1,
41 );
42 if renamed == state.definition_toml
43 && !state
44 .definition_toml
45 .contains(&format!("name = \"{new_name}\""))
46 {
47 warnings.push("could not rewrite [flow] name; edit the header manually".into());
48 }
49 let toml = format!(
50 "{}{}",
51 header_comment(source_label, "frozen snapshot reused verbatim"),
52 renamed
53 );
54 Ok(AdoptedFlow {
55 name: new_name.to_string(),
56 toml,
57 warnings,
58 })
59}
60
61/// Adopt a session's settled shell work: every DISTINCT command that ran
62/// green, in first-run order, chained linearly with a merge node that
63/// reports partial failure. This is the honest extractable core of a
64/// coding session — prompts and model turns stay behind.
65pub fn from_green_commands(
66 name: &str,
67 description: &str,
68 commands: &[String],
69) -> Result<AdoptedFlow, String> {
70 if commands.is_empty() {
71 return Err("no green bash commands found to adopt".into());
72 }
73 let mut toml = String::new();
74 toml.push_str(&header_comment(
75 "session commands (green only)",
76 description,
77 ));
78 toml.push_str(&format!("[flow]\nname = \"{name}\"\n"));
79 if !description.trim().is_empty() {
80 // Single-line safe description.
81 let d: String = description
82 .chars()
83 .map(|c| if c == '\n' { ' ' } else { c })
84 .collect();
85 toml.push_str(&format!("description = \"{d}\"\n"));
86 }
87 toml.push('\n');
88
89 let mut prev: Option<String> = None;
90 for (i, cmd) in commands.iter().enumerate() {
91 let id = format!("step_{}", i + 1);
92 toml.push_str("[[nodes]]\n");
93 toml.push_str(&format!("id = \"{id}\"\n"));
94 toml.push_str("type = \"bash\"\n");
95 if let Some(p) = &prev {
96 toml.push_str(&format!("deps = [\"{p}\"]\n"));
97 }
98 // Escape for basic TOML string: backslashes and quotes.
99 let escaped = cmd.replace('\\', "\\\\").replace('"', "\\\"");
100 toml.push_str(&format!("command = \"{escaped}\"\n\n"));
101 prev = Some(id);
102 }
103
104 // Merge tail always runs so a broken chain still reports partial output.
105 let Some(last) = prev else {
106 return Err("no commands to chain".into());
107 };
108 toml.push_str("[[nodes]]\n");
109 toml.push_str("id = \"summary\"\ntype = \"merge\"\n");
110 toml.push_str(&format!("deps = [\"{last}\"]\nrequired = false\n"));
111
112 Ok(AdoptedFlow {
113 name: name.to_string(),
114 toml,
115 warnings: Vec::new(),
116 })
117}
118
119/// Deterministic run-vs-run comparison over two ledger JSON strings:
120/// per-node status deltas and output differences. Pure; no model calls.
121pub fn diff_flow_states(a_json: &str, b_json: &str) -> Result<String, String> {
122 let a: FlowState = serde_json::from_str(a_json).map_err(|e| format!("ledger A: {e}"))?;
123 let b: FlowState = serde_json::from_str(b_json).map_err(|e| format!("ledger B: {e}"))?;
124
125 let mut out = String::new();
126 out.push_str(&format!(
127 "comparing {} vs {} (same definition: {})\n",
128 a.run_id,
129 b.run_id,
130 a.definition_toml == b.definition_toml
131 ));
132
133 let mut keys: Vec<&String> = a.nodes.keys().chain(b.nodes.keys()).collect();
134 keys.sort();
135 keys.dedup();
136
137 let mut diffs = 0usize;
138 for k in keys {
139 let sa = a.nodes.get(k);
140 let sb = b.nodes.get(k);
141 match (sa, sb) {
142 (Some(x), Some(y)) => {
143 if x.status != y.status {
144 out.push_str(&format!(
145 " ~ {k}: status {:?} → {:?}\n",
146 x.status, y.status
147 ));
148 diffs += 1;
149 } else if x.output != y.output {
150 out.push_str(&format!(
151 " ~ {k}: output differs ({} vs {} bytes)\n",
152 x.output.len(),
153 y.output.len()
154 ));
155 diffs += 1;
156 }
157 }
158 (Some(x), None) => {
159 out.push_str(&format!(" - {k}: only in A ({:?})\n", x.status));
160 diffs += 1;
161 }
162 (None, Some(y)) => {
163 out.push_str(&format!(" + {k}: only in B ({:?})\n", y.status));
164 diffs += 1;
165 }
166 (None, None) => unreachable!(),
167 }
168 }
169 if diffs == 0 {
170 out.push_str(" identical (no node status/output differences)\n");
171 } else {
172 out.insert_str(0, &format!("{diffs} difference(s):\n"));
173 }
174 Ok(out)
175}
176
177/// Recovery audit for `--resume`: classify the snapshot relationship and
178/// return the typed action. Fail-closed on drift unless accepted.
179pub fn recovery_audit(
180 state_json: &str,
181 live_definition: Option<&str>,
182) -> Result<(String, &'static str), String> {
183 let state: FlowState = serde_json::from_str(state_json).map_err(|e| e.to_string())?;
184 match live_definition {
185 None => Ok(("missing".into(), "rebuild")),
186 Some(live) if live.trim() == state.definition_toml.trim() => {
187 Ok(("frozen".into(), "resume"))
188 }
189 Some(_) => Ok(("drifted".into(), "repair")),
190 }
191}
192
193#[cfg(test)]
194#[allow(clippy::unwrap_used, clippy::expect_used)]
195mod tests {
196 use super::*;
197
198 const SAMPLE_STATE: &str = r#"{
199 "run_id":"r1","flow_name":"old-name",
200 "definition_toml":"[flow]\nname = \"old-name\"\n\n[[nodes]]\nid = \"a\"\ntype = \"bash\"\ncommand = \"echo hi\"\n",
201 "started_at":"2026-08-24T00:00:00Z",
202 "nodes":{"a":{"status":"completed","output":"hi"}}
203 }"#;
204
205 #[test]
206 fn adopt_from_state_reuses_frozen_toml_and_renames() {
207 let f = from_flow_state(SAMPLE_STATE, "new-name", "plan-1.json").unwrap();
208 assert!(f.toml.contains("name = \"new-name\""));
209 assert!(
210 f.toml
211 .contains("# adopted by `flow adopt` from plan-1.json")
212 );
213 assert!(f.toml.contains("command = \"echo hi\""));
214 assert!(f.warnings.is_empty());
215 }
216
217 #[test]
218 fn adopt_rejects_empty_run() {
219 let empty = r#"{"run_id":"r","flow_name":"x","definition_toml":"","started_at":"2026-08-24T00:00:00Z","nodes":{}}"#;
220 assert!(from_flow_state(empty, "n", "s").is_err());
221 }
222
223 #[test]
224 fn green_commands_become_chained_bash_nodes_with_merge_tail() {
225 let f = from_green_commands(
226 "verify-loop",
227 "rebuild + test",
228 &["cargo build".into(), "cargo test -q".into()],
229 )
230 .unwrap();
231 assert!(f.toml.contains("id = \"step_1\""));
232 assert!(f.toml.contains("id = \"step_2\""));
233 assert!(f.toml.contains("deps = [\"step_1\"]"));
234 assert!(f.toml.contains("type = \"merge\""));
235 assert!(f.toml.contains("command = \"cargo test -q\""));
236 }
237
238 #[test]
239 fn escapes_quotes_and_backslashes_in_commands() {
240 let f = from_green_commands("x", "", &[r#"echo "a\b""#.into()]).unwrap();
241 assert!(f.toml.contains(r#"command = "echo \"a\\b\"""#));
242 }
243
244 #[test]
245 fn empty_command_list_is_an_error() {
246 assert!(from_green_commands("x", "", &[]).is_err());
247 }
248
249 const STATE_A: &str = r#"{"run_id":"A","flow_name":"f","definition_toml":"[flow]\nname=\"f\"","started_at":"2026-08-24T00:00:00Z","nodes":{"n1":{"status":"completed","output":"ok"}}}"#;
250 const STATE_B_SAME: &str = r#"{"run_id":"B","flow_name":"f","definition_toml":"[flow]\nname=\"f\"","started_at":"2026-08-24T01:00:00Z","nodes":{"n1":{"status":"completed","output":"ok"}}}"#;
251 const STATE_C_DIFF: &str = r#"{"run_id":"C","flow_name":"f","definition_toml":"[flow]\nname=\"f\"","started_at":"2026-08-24T02:00:00Z","nodes":{"n1":{"status":"failed","output":"boom"}}}"#;
252
253 #[test]
254 fn diff_reports_identical_and_differences() {
255 let same = diff_flow_states(STATE_A, STATE_B_SAME).unwrap();
256 assert!(same.contains("identical"));
257 let d = diff_flow_states(STATE_A, STATE_C_DIFF).unwrap();
258 assert!(d.contains("1 difference(s)"));
259 assert!(d.contains("status Completed → Failed"));
260 }
261
262 #[test]
263 fn recovery_audit_classifies_snapshot_relationship() {
264 let live_same = "[flow]\nname=\"f\"";
265 let (snap, action) = recovery_audit(STATE_A, Some(live_same)).unwrap();
266 assert_eq!((snap.as_str(), action), ("frozen", "resume"));
267
268 let (snap, action) = recovery_audit(STATE_A, Some("[flow]\nname=\"edited\"")).unwrap();
269 assert_eq!((snap.as_str(), action), ("drifted", "repair"));
270
271 let (snap, action) = recovery_audit(STATE_A, None).unwrap();
272 assert_eq!((snap.as_str(), action), ("missing", "rebuild"));
273 }
274}
275

1//! Runs→flows adoption (docs/design/10-flows.md): convert proven work —
2//! planner plan ledgers, flow-run snapshots, or a session's settled shell
3//! commands — into a governed, hand-editable flow file. Provider/model are
4//! NEVER baked in from caller hints; attribution stays in the receipts.
5
6use crate::types::FlowState;
7
8#[derive(Debug, Clone)]
9pub struct AdoptedFlow {
10 pub name: String,
11 pub toml: String,
12 pub warnings: Vec<String>,
13}
14
15fn header_comment(source: &str, note: &str) -> String {
16 format!(
17 "# adopted by `flow adopt` from {source} at {}\n# {note}\n",
18 chrono::Utc::now().to_rfc3339(),
19 )
20}
21
22/// Adopt a frozen plan/flow-run snapshot: its definition TOML is already
23/// validated and proven — reuse it byte-for-byte under a new name.
24pub fn from_flow_state(
25 state_json: &str,
26 new_name: &str,
27 source_label: &str,
28) -> Result<AdoptedFlow, String> {
29 let state: FlowState =
30 serde_json::from_str(state_json).map_err(|e| format!("not a flow-run ledger: {e}"))?;
31 if state.nodes.is_empty() {
32 return Err("run has no node results to adopt".into());
33 }
34 let mut warnings = Vec::new();
35 // Rename inside the TOML so the adopted copy does not collide with the
36 // source flow's own identity.
37 let renamed = state.definition_toml.replacen(
38 &format!("name = \"{}\"", state.flow_name),
39 &format!("name = \"{new_name}\""),
40 1,
41 );
42 if renamed == state.definition_toml
43 && !state
44 .definition_toml
45 .contains(&format!("name = \"{new_name}\""))
46 {
47 warnings.push("could not rewrite [flow] name; edit the header manually".into());
48 }
49 let toml = format!(
50 "{}{}",
51 header_comment(source_label, "frozen snapshot reused verbatim"),
52 renamed
53 );
54 Ok(AdoptedFlow {
55 name: new_name.to_string(),
56 toml,
57 warnings,
58 })
59}
60
61/// Adopt a session's settled shell work: every DISTINCT command that ran
62/// green, in first-run order, chained linearly with a merge node that
63/// reports partial failure. This is the honest extractable core of a
64/// coding session — prompts and model turns stay behind.
65pub fn from_green_commands(
66 name: &str,
67 description: &str,
68 commands: &[String],
69) -> Result<AdoptedFlow, String> {
70 if commands.is_empty() {
71 return Err("no green bash commands found to adopt".into());
72 }
73 let mut toml = String::new();
74 toml.push_str(&header_comment(
75 "session commands (green only)",
76 description,
77 ));
78 toml.push_str(&format!("[flow]\nname = \"{name}\"\n"));
79 if !description.trim().is_empty() {
80 // Single-line safe description.
81 let d: String = description
82 .chars()
83 .map(|c| if c == '\n' { ' ' } else { c })
84 .collect();
85 toml.push_str(&format!("description = \"{d}\"\n"));
86 }
87 toml.push('\n');
88
89 let mut prev: Option<String> = None;
90 for (i, cmd) in commands.iter().enumerate() {
91 let id = format!("step_{}", i + 1);
92 toml.push_str("[[nodes]]\n");
93 toml.push_str(&format!("id = \"{id}\"\n"));
94 toml.push_str("type = \"bash\"\n");
95 if let Some(p) = &prev {
96 toml.push_str(&format!("deps = [\"{p}\"]\n"));
97 }
98 // Escape for basic TOML string: backslashes and quotes.
99 let escaped = cmd.replace('\\', "\\\\").replace('"', "\\\"");
100 toml.push_str(&format!("command = \"{escaped}\"\n\n"));
101 prev = Some(id);
102 }
103
104 // Merge tail always runs so a broken chain still reports partial output.
105 let Some(last) = prev else {
106 return Err("no commands to chain".into());
107 };
108 toml.push_str("[[nodes]]\n");
109 toml.push_str("id = \"summary\"\ntype = \"merge\"\n");
110 toml.push_str(&format!("deps = [\"{last}\"]\nrequired = false\n"));
111
112 Ok(AdoptedFlow {
113 name: name.to_string(),
114 toml,
115 warnings: Vec::new(),
116 })
117}
118
119/// Deterministic run-vs-run comparison over two ledger JSON strings:
120/// per-node status deltas and output differences. Pure; no model calls.
121pub fn diff_flow_states(a_json: &str, b_json: &str) -> Result<String, String> {
122 let a: FlowState = serde_json::from_str(a_json).map_err(|e| format!("ledger A: {e}"))?;
123 let b: FlowState = serde_json::from_str(b_json).map_err(|e| format!("ledger B: {e}"))?;
124
125 let mut out = String::new();
126 out.push_str(&format!(
127 "comparing {} vs {} (same definition: {})\n",
128 a.run_id,
129 b.run_id,
130 a.definition_toml == b.definition_toml
131 ));
132
133 let mut keys: Vec<&String> = a.nodes.keys().chain(b.nodes.keys()).collect();
134 keys.sort();
135 keys.dedup();
136
137 let mut diffs = 0usize;
138 for k in keys {
139 let sa = a.nodes.get(k);
140 let sb = b.nodes.get(k);
141 match (sa, sb) {
142 (Some(x), Some(y)) => {
143 if x.status != y.status {
144 out.push_str(&format!(
145 " ~ {k}: status {:?} → {:?}\n",
146 x.status, y.status
147 ));
148 diffs += 1;
149 } else if x.output != y.output {
150 out.push_str(&format!(
151 " ~ {k}: output differs ({} vs {} bytes)\n",
152 x.output.len(),
153 y.output.len()
154 ));
155 diffs += 1;
156 }
157 }
158 (Some(x), None) => {
159 out.push_str(&format!(" - {k}: only in A ({:?})\n", x.status));
160 diffs += 1;
161 }
162 (None, Some(y)) => {
163 out.push_str(&format!(" + {k}: only in B ({:?})\n", y.status));
164 diffs += 1;
165 }
166 (None, None) => unreachable!(),
167 }
168 }
169 if diffs == 0 {
170 out.push_str(" identical (no node status/output differences)\n");
171 } else {
172 out.insert_str(0, &format!("{diffs} difference(s):\n"));
173 }
174 Ok(out)
175}
176
177/// Recovery audit for `--resume`: classify the snapshot relationship and
178/// return the typed action. Fail-closed on drift unless accepted.
179pub fn recovery_audit(
180 state_json: &str,
181 live_definition: Option<&str>,
182) -> Result<(String, &'static str), String> {
183 let state: FlowState = serde_json::from_str(state_json).map_err(|e| e.to_string())?;
184 match live_definition {
185 None => Ok(("missing".into(), "rebuild")),
186 Some(live) if live.trim() == state.definition_toml.trim() => {
187 Ok(("frozen".into(), "resume"))
188 }
189 Some(_) => Ok(("drifted".into(), "repair")),
190 }
191}
192
193#[cfg(test)]
194#[allow(clippy::unwrap_used, clippy::expect_used)]
195mod tests {
196 use super::*;
197
198 const SAMPLE_STATE: &str = r#"{
199 "run_id":"r1","flow_name":"old-name",
200 "definition_toml":"[flow]\nname = \"old-name\"\n\n[[nodes]]\nid = \"a\"\ntype = \"bash\"\ncommand = \"echo hi\"\n",
201 "started_at":"2026-08-24T00:00:00Z",
202 "nodes":{"a":{"status":"completed","output":"hi"}}
203 }"#;
204
205 #[test]
206 fn adopt_from_state_reuses_frozen_toml_and_renames() {
207 let f = from_flow_state(SAMPLE_STATE, "new-name", "plan-1.json").unwrap();
208 assert!(f.toml.contains("name = \"new-name\""));
209 assert!(
210 f.toml
211 .contains("# adopted by `flow adopt` from plan-1.json")
212 );
213 assert!(f.toml.contains("command = \"echo hi\""));
214 assert!(f.warnings.is_empty());
215 }
216
217 #[test]
218 fn adopt_rejects_empty_run() {
219 let empty = r#"{"run_id":"r","flow_name":"x","definition_toml":"","started_at":"2026-08-24T00:00:00Z","nodes":{}}"#;
220 assert!(from_flow_state(empty, "n", "s").is_err());
221 }
222
223 #[test]
224 fn green_commands_become_chained_bash_nodes_with_merge_tail() {
225 let f = from_green_commands(
226 "verify-loop",
227 "rebuild + test",
228 &["cargo build".into(), "cargo test -q".into()],
229 )
230 .unwrap();
231 assert!(f.toml.contains("id = \"step_1\""));
232 assert!(f.toml.contains("id = \"step_2\""));
233 assert!(f.toml.contains("deps = [\"step_1\"]"));
234 assert!(f.toml.contains("type = \"merge\""));
235 assert!(f.toml.contains("command = \"cargo test -q\""));
236 }
237
238 #[test]
239 fn escapes_quotes_and_backslashes_in_commands() {
240 let f = from_green_commands("x", "", &[r#"echo "a\b""#.into()]).unwrap();
241 assert!(f.toml.contains(r#"command = "echo \"a\\b\"""#));
242 }
243
244 #[test]
245 fn empty_command_list_is_an_error() {
246 assert!(from_green_commands("x", "", &[]).is_err());
247 }
248
249 const STATE_A: &str = r#"{"run_id":"A","flow_name":"f","definition_toml":"[flow]\nname=\"f\"","started_at":"2026-08-24T00:00:00Z","nodes":{"n1":{"status":"completed","output":"ok"}}}"#;
250 const STATE_B_SAME: &str = r#"{"run_id":"B","flow_name":"f","definition_toml":"[flow]\nname=\"f\"","started_at":"2026-08-24T01:00:00Z","nodes":{"n1":{"status":"completed","output":"ok"}}}"#;
251 const STATE_C_DIFF: &str = r#"{"run_id":"C","flow_name":"f","definition_toml":"[flow]\nname=\"f\"","started_at":"2026-08-24T02:00:00Z","nodes":{"n1":{"status":"failed","output":"boom"}}}"#;
252
253 #[test]
254 fn diff_reports_identical_and_differences() {
255 let same = diff_flow_states(STATE_A, STATE_B_SAME).unwrap();
256 assert!(same.contains("identical"));
257 let d = diff_flow_states(STATE_A, STATE_C_DIFF).unwrap();
258 assert!(d.contains("1 difference(s)"));
259 assert!(d.contains("status Completed → Failed"));
260 }
261
262 #[test]
263 fn recovery_audit_classifies_snapshot_relationship() {
264 let live_same = "[flow]\nname=\"f\"";
265 let (snap, action) = recovery_audit(STATE_A, Some(live_same)).unwrap();
266 assert_eq!((snap.as_str(), action), ("frozen", "resume"));
267
268 let (snap, action) = recovery_audit(STATE_A, Some("[flow]\nname=\"edited\"")).unwrap();
269 assert_eq!((snap.as_str(), action), ("drifted", "repair"));
270
271 let (snap, action) = recovery_audit(STATE_A, None).unwrap();
272 assert_eq!((snap.as_str(), action), ("missing", "rebuild"));
273 }
274}
275

Indexing the workspace…