OxygenFlow M0: pipeline parser and canonical job graph, receipts, the runner API types, flow-init, the Firecracker tier, flow check, run, rootfs import and runner e0918b887850

Author: Andy McCall @andymccall <[email protected]>
When: 1 month ago · 2026-09-28 11:36 UTC
Signature: ✔ signed by Andy McCall
Hash: e0918b887850318bed4bbbc69e984e037b20bdff738ad75435012e75be400c40
Depends on: nothing: a root change
Depended on by: e3fb0e9b5e6b f3b5b86e23ef 3029d39be772 1a50d4dae89c 412808f9f769 21a3ea4aab88 48ed8a1db369 060d459ff6c6 0d663e91c598 5544c8998aa7 9296a15494d3 1215f05ff8d0 32f54b393396 77c8c24066ef c8917cf1f931 eb8e445f27b6 a63d3dcd2aef cce63af9ec54 f4dfd742b30d 4dd392d16454 494a1b9a8ac2 6f66e397539c e6949d87205b 3554afefb6aa cc462b56fbbb cedd12342afa 732b4de405e6 85275e1ef3b7 67851a868a3b f201e1bc5bcd 1b160d372d9e 06a4c472b385 5cf1e8aefbb6 28cc603f737d 39e737fbc7a9 1408324d5d55 6a2d31568cf2 4be5e4cd0af6 8bf49b95b09e 077a65242359 da72c3b192c3 2594278f49b2 8efb7e14532b 32fbe8980585 fc2a6d1817c0 35f7a849bcc5 448568561f2a e87c1c138187 d65cad6bdeb5 a4b3f8cb9aae ed4aef8b70ca 2be1fe4ad345 f4e51e63582d e20897216221 46fe2c815cc3 ea81796bb757 1e6b9fb1e73f 0077c7488e5d de1d7926185f 82cf6e91d398 89171fca079c d60aac6c3f8a ecce3719da80 b14aeaae683a e38bb032e38b 3b9dddc92057 8f211e20bea5 62829d15ef09 4cf0c99a1486 59078d2b8d46 2814f4213e7c 6d53fb7d00ec ba8fb1bbcc0a 8f81a19252b1 9a6fac2b40a9 b11e30f5031d ea56c56eaf90 c5b702e59c1f f53e6d47b56a 1abadd4a2fa6 720eb8191059 e2c125610f43 c58efc6a74db 78c0a9446558 280ef951d193 f2609cf795d8 bf7bb1b80fa3 19728d93bef4 02b127d2c721 ab3204b9974b bda0d3c9468e 1daca0b6bb8c 8c0a4e3a8a30 3007d8f011fa 6b3d325df979 e576e08617dc f849c7889471 00b245154ee2 b647ab72d51e 2d025bbf9842 f98a8f7881e0 a67b97e9a717 47c1d64dec8a f3b4fdeb2bc2 a72abeeedd78 1b2229dab1a4 c4285535b03a bb643c252515 9e8f4a56630b 50354f576ea6 fe128fa67a06 c23d519168d8 eb5c4db673e5 fc10d9e80383 a083cc802892
In bookmarks: main newer-pipeline push-uses-login
In tags: v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
Edits: 22 atoms, 31 tree operations, 22 files

Browse the tree at this change

hide whitespace changes

added .oxyignore +2 −0

added Cargo.lock +1679 −0

added Cargo.toml +22 −0

added crates/flow-core/Cargo.toml +17 −0

added crates/flow-core/src/api.rs +68 −0

added crates/flow-core/src/canonical.rs +95 −0

added crates/flow-core/src/lib.rs +14 −0

added crates/flow-core/src/pipeline.rs +554 −0

added crates/flow-core/src/receipt.rs +100 −0

added crates/flow-core/src/tests.rs +178 −0

added crates/flow-init/Cargo.toml +11 −0

added crates/flow-init/src/main.rs +185 −0

added crates/flow/Cargo.toml +27 −0

added crates/flow/src/cpio.rs +56 −0

added crates/flow/src/home.rs +114 −0

added crates/flow/src/main.rs +256 −0

added crates/flow/src/microvm.rs +164 −0

added crates/flow/src/runner.rs +222 −0

added crates/flow/src/workspace.rs +117 −0

added crates/flow/tests/microvm.rs +58 −0

added docs/m0.md +156 −0

added rustfmt.toml +2 −0

1	+	[workspace]
2	+	members = ["crates/flow-core", "crates/flow-init", "crates/flow"]
3	+	resolver = "2"
4	+
5	+	[workspace.package]
6	+	version = "0.1.0"
7	+	edition = "2021"
8	+	license = "MIT OR Apache-2.0"
9	+
10	+	[workspace.dependencies]
11	+	oxy-core = { path = "../oxygen/crates/oxy-core" }
12	+	flow-core = { path = "crates/flow-core" }
13	+	serde = { version = "1", features = ["derive"] }
14	+	serde_json = "1"
15	+	blake3 = "1"
16	+	hex = "0.4"
17	+	thiserror = "1"
18	+	anyhow = "1"
19	+	tempfile = "3"
20	+
21	+	[profile.release]
22	+	debug = 1

1	+	[package]
2	+	name = "flow-core"
3	+	version.workspace = true
4	+	edition.workspace = true
5	+	license.workspace = true
6	+	description = "OxygenFlow pipelines: parsing, validation, canonical job graphs and receipts"
7	+
8	+	[dependencies]
9	+	oxy-core.workspace = true
10	+	serde.workspace = true
11	+	serde_json.workspace = true
12	+	blake3.workspace = true
13	+	hex.workspace = true
14	+	thiserror.workspace = true
15	+	kdl = "6"
16	+	miette = "7"
17	+	globset = "0.4"

1	+	//! OxygenFlow core (docs/spec.md, docs/m0.md): pipelines, their
2	+	//! canonical form and hashes, and execution receipts. No I/O.
3	+
4	+	pub mod api;
5	+	pub mod canonical;
6	+	pub mod pipeline;
7	+	pub mod receipt;
8	+
9	+	pub use canonical::{object_hash, scope_tree_hash, Hash};
10	+	pub use pipeline::{Job, Network, Pipeline, Problem, Problems, Runtime, Sandbox, PIPELINE_PATH};
11	+	pub use receipt::{Environment, Receipt};
12	+
13	+	#[cfg(test)]
14	+	mod tests;

1	+	use crate::*;
2	+
3	+	const ROOTFS: &str = "9b5c2a1e8f3d4e7a6b8c9d0e1f2a3b4c5d6e7f8a9b0c1d2e3f4a5b6c7d8e9f0a";
4	+
5	+	fn doc(body: &str) -> String {
6	+	format!(
7	+	"schema-version 1\n\nsandbox \"alpine\" {{\n runtime \"microvm\"\n rootfs \"cas://blob/{ROOTFS}\"\n memory-limit \"512MiB\"\n cpu-limit 2\n network \"isolated\"\n}}\n\n{body}"
8	+	)
9	+	}
10	+
11	+	fn problems(src: &str) -> Vec<(usize, String)> {
12	+	Pipeline::parse(src).unwrap_err().0.into_iter().map(\|p\| (p.line, p.message)).collect()
13	+	}
14	+
15	+	#[test]
16	+	fn parses_the_m0_example() {
17	+	let p = Pipeline::parse(&doc(r#"job "lint" {
18	+	sandbox "alpine"
19	+	scope "src/**" "README.md"
20	+	env {
21	+	LANG "C.UTF-8"
22	+	}
23	+	exec "grep" "-rn" "TODO" "src"
24	+	}
25	+
26	+	job "test" {
27	+	sandbox "alpine"
28	+	depends-on "lint"
29	+	exec "sh" "-eu" "test.sh"
30	+	}
31	+	"#))
32	+	.unwrap();
33	+	let sb = &p.sandboxes["alpine"];
34	+	assert_eq!((sb.runtime, sb.memory_mib, sb.cpus, sb.network), (Runtime::Microvm, 512, 2, Network::Isolated));
35	+	assert_eq!(sb.rootfs, ROOTFS);
36	+	let lint = &p.jobs["lint"];
37	+	assert_eq!(lint.exec, ["grep", "-rn", "TODO", "src"]);
38	+	assert_eq!(lint.scope, ["src/**", "README.md"]);
39	+	assert_eq!(lint.env["LANG"], "C.UTF-8");
40	+	assert_eq!(p.jobs["test"].depends_on, ["lint"]);
41	+	assert_eq!(p.order().iter().map(\|j\| j.name.as_str()).collect::<Vec<_>>(), ["lint", "test"]);
42	+	assert_eq!(p.depths()["test"], 1);
43	+	}
44	+
45	+	#[test]
46	+	fn defaults_and_gib() {
47	+	let p = Pipeline::parse(&format!(
48	+	"schema-version 1\nsandbox \"s\" {{\n rootfs \"cas://blob/{ROOTFS}\"\n memory-limit \"2GiB\"\n}}\njob \"a\" {{\n sandbox \"s\"\n exec \"true\"\n}}\n"
49	+	))
50	+	.unwrap();
51	+	let s = &p.sandboxes["s"];
52	+	assert_eq!((s.runtime, s.memory_mib, s.cpus, s.network), (Runtime::Microvm, 2048, 1, Network::Isolated));
53	+	}
54	+
55	+	#[test]
56	+	fn shell_metacharacters_stay_literal_argv() {
57	+	// Spec §10.1: every byte reaches argv as written; nothing adds a shell.
58	+	let p = Pipeline::parse(&doc(r#"job "x" { sandbox "alpine"; exec "echo" "a; rm -rf / && $(id) \| `x`" "b\nc"; }"#)).unwrap();
59	+	assert_eq!(p.jobs["x"].exec, ["echo", "a; rm -rf / && $(id) \| `x`", "b\nc"]);
60	+	}
61	+
62	+	#[test]
63	+	fn problems_carry_lines_and_are_all_reported() {
64	+	let src = doc(r#"job "a" {
65	+	sandbox "missing"
66	+	exec "true"
67	+	mount { source "oxy://workspace"; }
68	+	}
69	+	job "b" {
70	+	sandbox "alpine"
71	+	depends-on "nope"
72	+	exec
73	+	}
74	+	flavour "x"
75	+	"#);
76	+	let ps = problems(&src);
77	+	let has = \|line: usize, text: &str\| ps.iter().any(\|(l, m)\| *l == line && m.contains(text));
78	+	assert!(has(11, "which is not declared") && has(11, "\"missing\""), "{ps:?}");
79	+	assert!(has(14, "`mount` is not in M0"), "{ps:?}");
80	+	assert!(has(16, "\"nope\""), "{ps:?}");
81	+	assert!(has(19, "`exec` needs a program"), "{ps:?}");
82	+	assert!(has(21, "unknown node `flavour`"), "{ps:?}");
83	+	}
84	+
85	+	#[test]
86	+	fn typed_values_are_not_coerced() {
87	+	let src = format!(
88	+	"schema-version 1\nsandbox \"s\" {{\n rootfs \"cas://blob/{ROOTFS}\"\n cpu-limit \"2\"\n memory-limit \"1.5GiB\"\n}}\njob \"a\" {{\n sandbox \"s\"\n exec \"echo\" 1\n}}\n"
89	+	);
90	+	let ps = problems(&src);
91	+	assert!(ps.iter().any(\|(l, m)\| *l == 4 && m.contains("integer, not a string")), "{ps:?}");
92	+	assert!(ps.iter().any(\|(l, m)\| *l == 5 && m.contains("whole number")), "{ps:?}");
93	+	assert!(ps.iter().any(\|(l, m)\| *l == 9 && m.contains("strings, not an integer")), "{ps:?}");
94	+	}
95	+
96	+	#[test]
97	+	fn m0_refuses_what_it_cannot_honour() {
98	+	let src = format!("schema-version 1\nsandbox \"s\" {{\n runtime \"wasm\"\n rootfs \"cas://blob/{ROOTFS}\"\n network \"egress\"\n}}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}\n");
99	+	let ps = problems(&src);
100	+	assert!(ps.iter().any(\|(l, m)\| *l == 3 && m.contains("wasm")), "{ps:?}");
101	+	assert!(ps.iter().any(\|(l, m)\| *l == 5 && m.contains("isolated")), "{ps:?}");
102	+	assert!(problems("schema-version 2\n").iter().any(\|(_, m)\| m.contains("schema-version 2")));
103	+	assert!(problems(&format!(
104	+	"schema-version 1\nsandbox \"s\" {{ rootfs \"cas://blob/{}\"; }}\njob \"a\" {{ sandbox \"s\"; exec \"true\"; }}",
105	+	&ROOTFS[..60]
106	+	))
107	+	.iter()
108	+	.any(\|(_, m)\| m.contains("64 lower-case hex")));
109	+	}
110	+
111	+	#[test]
112	+	fn cycles_are_found() {
113	+	let ps = problems(&doc(
114	+	"job \"a\" { sandbox \"alpine\"; depends-on \"c\"; exec \"true\"; }\njob \"b\" { sandbox \"alpine\"; depends-on \"a\"; exec \"true\"; }\njob \"c\" { sandbox \"alpine\"; depends-on \"b\"; exec \"true\"; }\njob \"d\" { sandbox \"alpine\"; depends-on \"d\"; exec \"true\"; }\n",
115	+	));
116	+	assert!(ps.iter().any(\|(_, m)\| m.contains("cycle: a → c → b → a")), "{ps:?}");
117	+	assert!(ps.iter().any(\|(_, m)\| m.contains("\"d\" depends on itself")), "{ps:?}");
118	+	}
119	+
120	+	#[test]
121	+	fn invalid_kdl_has_a_line() {
122	+	let ps = problems("schema-version 1\njob \"a\" {\n exec \"unterminated\n}\n");
123	+	assert!(!ps.is_empty() && ps[0].0 >= 2, "{ps:?}");
124	+	}
125	+
126	+	#[test]
127	+	fn canonical_form_ignores_layout_and_order() {
128	+	let a = Pipeline::parse(&doc(
129	+	"job \"b\" { sandbox \"alpine\"; exec \"true\"; }\njob \"a\" {\n sandbox \"alpine\"\n depends-on \"b\"\n exec \"x\"\n}\n",
130	+	))
131	+	.unwrap();
132	+	let b = Pipeline::parse(&doc("// comment\njob \"a\" { exec \"x\"; depends-on \"b\"; sandbox \"alpine\"; }\n\n\njob \"b\" { exec \"true\"; sandbox \"alpine\"; }\n")).unwrap();
133	+	assert_eq!(a.canonical(), b.canonical());
134	+	assert_eq!(a.hash(), b.hash());
135	+	let c = Pipeline::parse(&doc(
136	+	"job \"b\" { sandbox \"alpine\"; exec \"true\"; }\njob \"a\" { sandbox \"alpine\"; depends-on \"b\"; exec \"y\"; }\n",
137	+	))
138	+	.unwrap();
139	+	assert_ne!(a.hash(), c.hash());
140	+	let sb = &a.sandboxes["alpine"];
141	+	assert_ne!(a.jobs["a"].hash(sb), c.jobs["a"].hash(sb));
142	+	assert_eq!(a.jobs["b"].hash(sb), c.jobs["b"].hash(sb), "an unchanged job keeps its hash");
143	+	// The canonical form round-trips through Oxygen's strict decoder.
144	+	oxy_core::cbor::decode(&a.canonical()).unwrap();
145	+	}
146	+
147	+	#[test]
148	+	fn scope_tree_hash_sees_only_the_scope() {
149	+	let f = \|b: u8\| [b; 32];
150	+	let scope = vec!["src/**".to_string()];
151	+	let base = scope_tree_hash(&scope, [("src/a.rs", f(1)), ("README.md", f(2))]);
152	+	assert_eq!(base, scope_tree_hash(&scope, [("README.md", f(9)), ("src/a.rs", f(1))]), "outside the scope, and order, do not count");
153	+	assert_ne!(base, scope_tree_hash(&scope, [("src/a.rs", f(3)), ("README.md", f(2))]));
154	+	assert_ne!(scope_tree_hash(&[], [("README.md", f(2))]), scope_tree_hash(&[], [("README.md", f(9))]), "no scope is the whole tree");
155	+	}
156	+
157	+	#[test]
158	+	fn receipts_hash_as_oxygen_objects() {
159	+	let r = Receipt {
160	+	job_name: "lint".into(),
161	+	pipeline_hash: [1; 32],
162	+	scope_tree_hash: [2; 32],
163	+	inputs: vec![],
164	+	environment: [3; 32],
165	+	outputs: vec![],
166	+	exit_code: 0,
167	+	duration_ms: 1234,
168	+	};
169	+	let body = r.encode();
170	+	let mut want = format!("oxy1 receipt {}\0", body.len()).into_bytes();
171	+	want.extend_from_slice(&body);
172	+	assert_eq!(r.hash(), *blake3::hash(&want).as_bytes());
173	+	let json = serde_json::to_string(&r).unwrap();
174	+	assert_eq!(serde_json::from_str::<Receipt>(&json).unwrap(), r);
175	+	let mut r2 = r.clone();
176	+	r2.exit_code = 1;
177	+	assert_ne!(r.hash(), r2.hash());
178	+	}

1	+	[package]
2	+	name = "flow-init"
3	+	version.workspace = true
4	+	edition.workspace = true
5	+	license.workspace = true
6	+	description = "OxygenFlow guest init: PID 1 of a job's microVM (docs/m0.md §2)"
7	+
8	+	[dependencies]
9	+	libc = "0.2"
10	+	serde.workspace = true
11	+	serde_json.workspace = true

1	+	//! The runner API between OxygenHub and `flow runner` (docs/m0.md §5).
2	+	//! Both sides use these types, so they cannot drift.
3	+
4	+	use crate::pipeline::{Job, Sandbox};
5	+	use crate::receipt::Receipt;
6	+	use serde::{Deserialize, Serialize};
7	+
8	+	/// `POST /api/v1/flow/claim`.
9	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
10	+	pub struct ClaimRequest {
11	+	/// The runner's name, shown on job pages.
12	+	pub runner: String,
13	+	/// Tiers this runner can run (`microvm`, `syscall`).
14	+	pub runtimes: Vec<String>,
15	+	}
16	+
17	+	/// A claimed job: everything the runner needs, nothing more.
18	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
19	+	pub struct Claim {
20	+	pub job_id: i64,
21	+	pub run_id: i64,
22	+	/// `owner/name`.
23	+	pub repo: String,
24	+	/// Where to fetch the repository from (`oxy` wire protocol).
25	+	pub clone_url: String,
26	+	/// Read access to this repository only, until the job finishes.
27	+	pub token: String,
28	+	/// The state to build, as full change hashes.
29	+	pub heads: Vec<String>,
30	+	/// Hex BLAKE3 of the canonical pipeline.
31	+	pub pipeline_hash: String,
32	+	pub job: Job,
33	+	pub sandbox: Sandbox,
34	+	/// The hub's limit for this job.
35	+	pub timeout_s: u64,
36	+	}
37	+
38	+	/// One line of output (docs/m0.md §2).
39	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
40	+	pub struct Record {
41	+	/// Milliseconds since the job started.
42	+	pub t: u64,
43	+	/// `out`, `err`, `console` (the VM outside the job) or `flow` (the runner).
44	+	pub stream: String,
45	+	pub text: String,
46	+	}
47	+
48	+	/// `POST /api/v1/flow/jobs/{id}/log`.
49	+	#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
50	+	pub struct LogBatch {
51	+	pub records: Vec<Record>,
52	+	}
53	+
54	+	/// `POST /api/v1/flow/jobs/{id}/finish`.
55	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
56	+	pub struct Finish {
57	+	/// The job's exit code; absent when the runner could not run it.
58	+	pub exit_code: Option<i32>,
59	+	pub duration_ms: u64,
60	+	pub receipt: Option<Receipt>,
61	+	/// Why the runner could not run the job (rootfs missing, VM failed).
62	+	pub error: Option<String>,
63	+	}
64	+
65	+	/// Longest text of one record the hub keeps; longer lines are cut.
66	+	pub const MAX_RECORD_TEXT: usize = 16 * 1024;
67	+	/// Most records the hub keeps per job; later ones are dropped with a note.
68	+	pub const MAX_RECORDS: usize = 50_000;

1	+	//! The canonical job graph (spec §3.2) and object hashes. Pipelines,
2	+	//! jobs and receipts are encoded with Oxygen's deterministic CBOR profile
3	+	//! (Oxygen spec §5.1: unsigned integer map keys in order) and hashed the
4	+	//! way Oxygen hashes objects: `BLAKE3("oxy1 " \|\| type \|\| " " \|\| len \|\| 0x00 \|\| body)`.
5	+
6	+	use crate::pipeline::{Job, Pipeline, Sandbox};
7	+	use oxy_core::cbor::{encode, Value};
8	+
9	+	/// A 32-byte object reference, shown as hex.
10	+	pub type Hash = [u8; 32];
11	+
12	+	/// Oxygen's object hash construction for a Flow object type.
13	+	pub fn object_hash(ty: &str, body: &[u8]) -> Hash {
14	+	let mut h = blake3::Hasher::new();
15	+	h.update(format!("oxy1 {ty} {}\0", body.len()).as_bytes());
16	+	h.update(body);
17	+	*h.finalize().as_bytes()
18	+	}
19	+
20	+	fn text(s: &str) -> Value {
21	+	Value::Text(s.to_string())
22	+	}
23	+
24	+	fn texts(v: &[String]) -> Value {
25	+	Value::Array(v.iter().map(\|s\| text(s)).collect())
26	+	}
27	+
28	+	fn hex32(s: &str) -> Value {
29	+	Value::Bytes(hex::decode(s).unwrap_or_default())
30	+	}
31	+
32	+	pub fn sandbox_value(s: &Sandbox) -> Value {
33	+	Value::map([
34	+	(0, text(&s.name)),
35	+	(1, text(s.runtime.as_str())),
36	+	(2, hex32(&s.rootfs)),
37	+	(3, Value::Uint(s.memory_mib as u64)),
38	+	(4, Value::Uint(s.cpus as u64)),
39	+	(5, text(s.network.as_str())),
40	+	])
41	+	}
42	+
43	+	pub fn job_value(j: &Job) -> Value {
44	+	Value::map([
45	+	(0, text(&j.name)),
46	+	(1, text(&j.sandbox)),
47	+	(2, texts(&j.scope)),
48	+	(3, texts(&j.depends_on)),
49	+	// Environment as sorted [name, value] pairs: the profile has integer map keys only.
50	+	(4, Value::Array(j.env.iter().map(\|(k, v)\| Value::Array(vec![text(k), text(v)])).collect())),
51	+	(5, texts(&j.exec)),
52	+	])
53	+	}
54	+
55	+	impl Pipeline {
56	+	/// The canonical CBOR job graph: sandboxes and jobs in name order.
57	+	pub fn canonical(&self) -> Vec<u8> {
58	+	encode(&Value::map([
59	+	(0, Value::Uint(self.schema_version as u64)),
60	+	(1, Value::Array(self.sandboxes.values().map(sandbox_value).collect())),
61	+	(2, Value::Array(self.jobs.values().map(job_value).collect())),
62	+	]))
63	+	}
64	+
65	+	/// `pipeline_hash` of receipts and provenance.
66	+	pub fn hash(&self) -> Hash {
67	+	object_hash("pipeline", &self.canonical())
68	+	}
69	+	}
70	+
71	+	impl Job {
72	+	/// `Hash(J)` of spec §6.2: the job with its sandbox, so a change to
73	+	/// either is a different job.
74	+	pub fn hash(&self, sandbox: &Sandbox) -> Hash {
75	+	object_hash("job", &encode(&Value::Array(vec![job_value(self), sandbox_value(sandbox)])))
76	+	}
77	+	}
78	+
79	+	/// BLAKE3 tree hash of the files a scope selects (spec §6.2): sorted
80	+	/// `(path, content hash)` pairs. An empty scope selects every file.
81	+	pub fn scope_tree_hash<'a>(scope: &[String], files: impl IntoIterator<Item = (&'a str, Hash)>) -> Hash {
82	+	let set = {
83	+	let mut b = globset::GlobSetBuilder::new();
84	+	for g in scope {
85	+	if let Ok(g) = globset::Glob::new(g) {
86	+	b.add(g);
87	+	}
88	+	}
89	+	b.build().unwrap_or_else(\|_\| globset::GlobSet::empty())
90	+	};
91	+	let mut picked: Vec<(&str, Hash)> = files.into_iter().filter(\|(p, _)\| scope.is_empty() \|\| set.is_match(p)).collect();
92	+	picked.sort();
93	+	let body = encode(&Value::Array(picked.iter().map(\|(p, h)\| Value::Array(vec![text(p), Value::Bytes(h.to_vec())])).collect()));
94	+	object_hash("scope", &body)
95	+	}

1	+	//! `.flow/pipeline.kdl`: the M0 subset of spec §3 (docs/m0.md §1), parsed
2	+	//! from KDL 2.0 and validated before anything runs. Every problem is
3	+	//! reported with its line, and all of them at once.
4	+
5	+	use serde::{Deserialize, Serialize};
6	+	use std::collections::{BTreeMap, BTreeSet};
7	+	use std::fmt;
8	+
9	+	/// Where a pipeline lives in the state being built.
10	+	pub const PIPELINE_PATH: &str = ".flow/pipeline.kdl";
11	+
12	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
13	+	pub struct Pipeline {
14	+	pub schema_version: u32,
15	+	/// By name.
16	+	pub sandboxes: BTreeMap<String, Sandbox>,
17	+	/// By name.
18	+	pub jobs: BTreeMap<String, Job>,
19	+	}
20	+
21	+	#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
22	+	#[serde(rename_all = "lowercase")]
23	+	pub enum Runtime {
24	+	/// Tier 0 (spec §4.1); refused in M0.
25	+	Wasm,
26	+	/// Tier 1 (spec §4.2): namespaces, Landlock, seccomp, cgroups.
27	+	Syscall,
28	+	/// Tier 2 (spec §4.3), the default.
29	+	Microvm,
30	+	}
31	+
32	+	impl Runtime {
33	+	pub fn as_str(self) -> &'static str {
34	+	match self {
35	+	Runtime::Wasm => "wasm",
36	+	Runtime::Syscall => "syscall",
37	+	Runtime::Microvm => "microvm",
38	+	}
39	+	}
40	+	}
41	+
42	+	#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
43	+	#[serde(rename_all = "lowercase")]
44	+	pub enum Network {
45	+	Isolated,
46	+	Loopback,
47	+	Egress,
48	+	}
49	+
50	+	impl Network {
51	+	pub fn as_str(self) -> &'static str {
52	+	match self {
53	+	Network::Isolated => "isolated",
54	+	Network::Loopback => "loopback",
55	+	Network::Egress => "egress",
56	+	}
57	+	}
58	+	}
59	+
60	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
61	+	pub struct Sandbox {
62	+	pub name: String,
63	+	pub runtime: Runtime,
64	+	/// BLAKE3 of the rootfs image, hex.
65	+	pub rootfs: String,
66	+	pub memory_mib: u32,
67	+	pub cpus: u32,
68	+	pub network: Network,
69	+	}
70	+
71	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
72	+	pub struct Job {
73	+	pub name: String,
74	+	pub sandbox: String,
75	+	/// Glob patterns over repository paths (spec §6.2).
76	+	pub scope: Vec<String>,
77	+	/// Sorted.
78	+	pub depends_on: Vec<String>,
79	+	pub env: BTreeMap<String, String>,
80	+	/// Executed as is: no shell, no expansion (spec §2.2 I2).
81	+	pub exec: Vec<String>,
82	+	}
83	+
84	+	/// One problem in a pipeline, at a 1-based line.
85	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
86	+	pub struct Problem {
87	+	pub line: usize,
88	+	pub message: String,
89	+	}
90	+
91	+	impl fmt::Display for Problem {
92	+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
93	+	write!(f, "line {}: {}", self.line, self.message)
94	+	}
95	+	}
96	+
97	+	/// Every problem found, in line order.
98	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
99	+	pub struct Problems(pub Vec<Problem>);
100	+
101	+	impl fmt::Display for Problems {
102	+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
103	+	let lines: Vec<String> = self.0.iter().map(\|p\| p.to_string()).collect();
104	+	f.write_str(&lines.join("\n"))
105	+	}
106	+	}
107	+
108	+	const MIN_MEMORY_MIB: u64 = 64;
109	+	const MAX_MEMORY_MIB: u64 = 16 * 1024;
110	+	const MAX_CPUS: i128 = 16;
111	+
112	+	/// Nodes of the full spec that M0 recognises and refuses.
113	+	const LATER_TOP: &[&str] = &["publish", "secret"];
114	+	const LATER_JOB: &[&str] = &["mount", "output", "secret", "attest"];
115	+
116	+	struct Ctx<'a> {
117	+	src: &'a str,
118	+	problems: Vec<Problem>,
119	+	}
120	+
121	+	impl Ctx<'_> {
122	+	fn line(&self, span: SourceSpan) -> usize {
123	+	let at = span.offset().min(self.src.len());
124	+	self.src[..at].bytes().filter(\|b\| *b == b'\n').count() + 1
125	+	}
126	+
127	+	fn err(&mut self, span: SourceSpan, message: impl Into<String>) {
128	+	let line = self.line(span);
129	+	self.problems.push(Problem { line, message: message.into() });
130	+	}
131	+
132	+	/// The node's arguments as strings; properties and other types are errors.
133	+	fn strings(&mut self, node: &KdlNode) -> Vec<String> {
134	+	let mut out = Vec::new();
135	+	for e in node.entries() {
136	+	if let Some(k) = e.name() {
137	+	self.err(e.span(), format!("`{}` takes no property `{}`", node.name().value(), k.value()));
138	+	continue;
139	+	}
140	+	match e.value() {
141	+	KdlValue::String(s) => out.push(s.clone()),
142	+	other => self.err(e.span(), format!("`{}` takes strings, not {}", node.name().value(), kind_of(other))),
143	+	}
144	+	}
145	+	out
146	+	}
147	+
148	+	/// Exactly one string argument.
149	+	fn one_string(&mut self, node: &KdlNode) -> Option<String> {
150	+	let v = self.strings(node);
151	+	if v.len() != 1 {
152	+	self.err(node.span(), format!("`{}` takes one string", node.name().value()));
153	+	return None;
154	+	}
155	+	v.into_iter().next()
156	+	}
157	+
158	+	fn one_integer(&mut self, node: &KdlNode) -> Option<i128> {
159	+	match node.entries() {
160	+	[e] if e.name().is_none() => match e.value() {
161	+	KdlValue::Integer(n) => Some(*n),
162	+	other => {
163	+	self.err(e.span(), format!("`{}` takes an integer, not {}", node.name().value(), kind_of(other)));
164	+	None
165	+	}
166	+	},
167	+	_ => {
168	+	self.err(node.span(), format!("`{}` takes one integer", node.name().value()));
169	+	None
170	+	}
171	+	}
172	+	}
173	+
174	+	fn no_children(&mut self, node: &KdlNode) {
175	+	if node.children().is_some() {
176	+	self.err(node.span(), format!("`{}` takes no block", node.name().value()));
177	+	}
178	+	}
179	+
180	+	fn name_ok(&mut self, span: SourceSpan, what: &str, name: &str) -> bool {
181	+	let ok = !name.is_empty() && name.len() <= 64 && name.bytes().all(\|b\| b.is_ascii_alphanumeric() \|\| b"._-".contains(&b));
182	+	if !ok {
183	+	self.err(span, format!("{what} name {name:?} must be 1 to 64 of A-Z a-z 0-9 . _ -"));
184	+	}
185	+	ok
186	+	}
187	+	}
188	+
189	+	use kdl::{KdlDocument, KdlNode, KdlValue};
190	+	use miette::SourceSpan;
191	+
192	+	fn kind_of(v: &KdlValue) -> &'static str {
193	+	match v {
194	+	KdlValue::String(_) => "a string",
195	+	KdlValue::Integer(_) => "an integer",
196	+	KdlValue::Float(_) => "a number",
197	+	KdlValue::Bool(_) => "a boolean",
198	+	KdlValue::Null => "null",
199	+	}
200	+	}
201	+
202	+	/// `"512MiB"`, `"4GiB"`, `"65536KiB"` to MiB.
203	+	fn parse_memory(s: &str) -> Option<u64> {
204	+	let (num, unit) = s.split_at(s.find(\|c: char\| !c.is_ascii_digit())?);
205	+	let n: u64 = num.parse().ok()?;
206	+	let mib = match unit {
207	+	"KiB" => n.is_multiple_of(1024).then_some(n / 1024)?,
208	+	"MiB" => n,
209	+	"GiB" => n.checked_mul(1024)?,
210	+	_ => return None,
211	+	};
212	+	Some(mib)
213	+	}
214	+
215	+	impl Pipeline {
216	+	/// Parse and validate a pipeline document.
217	+	pub fn parse(src: &str) -> Result<Pipeline, Problems> {
218	+	let doc = match KdlDocument::parse_v2(src) {
219	+	Ok(d) => d,
220	+	Err(e) => {
221	+	let mut ctx = Ctx { src, problems: Vec::new() };
222	+	for d in &e.diagnostics {
223	+	let msg = d.message.clone().or_else(\|\| d.label.clone()).unwrap_or_else(\|\| "not valid KDL 2.0".into());
224	+	ctx.err(d.span, msg);
225	+	}
226	+	if ctx.problems.is_empty() {
227	+	ctx.problems.push(Problem { line: 1, message: "not valid KDL 2.0".into() });
228	+	}
229	+	return Err(Problems(ctx.problems));
230	+	}
231	+	};
232	+	let mut ctx = Ctx { src, problems: Vec::new() };
233	+	let mut schema = None;
234	+	let mut sandboxes = BTreeMap::new();
235	+	let mut jobs: BTreeMap<String, (Job, SourceSpan)> = BTreeMap::new();
236	+	for node in doc.nodes() {
237	+	match node.name().value() {
238	+	"schema-version" => {
239	+	ctx.no_children(node);
240	+	match ctx.one_integer(node) {
241	+	Some(1) => schema = Some(1),
242	+	Some(n) => ctx.err(node.span(), format!("schema-version {n} is not supported; this runner reads 1")),
243	+	None => {}
244	+	}
245	+	}
246	+	"sandbox" => {
247	+	if let Some(sb) = parse_sandbox(&mut ctx, node) {
248	+	if sandboxes.contains_key(&sb.name) {
249	+	ctx.err(node.span(), format!("sandbox {:?} is declared twice", sb.name));
250	+	} else {
251	+	sandboxes.insert(sb.name.clone(), sb);
252	+	}
253	+	}
254	+	}
255	+	"job" => {
256	+	if let Some(job) = parse_job(&mut ctx, node) {
257	+	if jobs.contains_key(&job.name) {
258	+	ctx.err(node.span(), format!("job {:?} is declared twice", job.name));
259	+	} else {
260	+	jobs.insert(job.name.clone(), (job, node.span()));
261	+	}
262	+	}
263	+	}
264	+	n if LATER_TOP.contains(&n) => ctx.err(node.span(), format!("`{n}` is not in M0 yet")),
265	+	n => ctx.err(node.span(), format!("unknown node `{n}`; expected schema-version, sandbox or job")),
266	+	}
267	+	}
268	+	if schema.is_none() && !ctx.problems.iter().any(\|p\| p.message.contains("schema-version")) {
269	+	ctx.problems.push(Problem { line: 1, message: "`schema-version 1` is required".into() });
270	+	}
271	+	if jobs.is_empty() && ctx.problems.is_empty() {
272	+	ctx.problems.push(Problem { line: 1, message: "a pipeline needs at least one job".into() });
273	+	}
274	+	// References and cycles.
275	+	for (job, span) in jobs.values() {
276	+	if !job.sandbox.is_empty() && !sandboxes.contains_key(&job.sandbox) {
277	+	ctx.err(*span, format!("job {:?} uses sandbox {:?}, which is not declared", job.name, job.sandbox));
278	+	}
279	+	for d in &job.depends_on {
280	+	if !jobs.contains_key(d) {
281	+	ctx.err(*span, format!("job {:?} depends on {:?}, which is not declared", job.name, d));
282	+	} else if d == &job.name {
283	+	ctx.err(*span, format!("job {:?} depends on itself", job.name));
284	+	}
285	+	}
286	+	}
287	+	let graph: BTreeMap<&str, Vec<&str>> =
288	+	jobs.values().map(\|(j, _)\| (j.name.as_str(), j.depends_on.iter().map(String::as_str).collect())).collect();
289	+	if let Some(cycle) = find_cycle(&graph) {
290	+	let span = jobs[cycle[0]].1;
291	+	ctx.err(span, format!("jobs depend on each other in a cycle: {}", cycle.join(" → ")));
292	+	}
293	+	if !ctx.problems.is_empty() {
294	+	ctx.problems.sort_by_key(\|p\| p.line);
295	+	return Err(Problems(ctx.problems));
296	+	}
297	+	Ok(Pipeline { schema_version: 1, sandboxes, jobs: jobs.into_iter().map(\|(k, (j, _))\| (k, j)).collect() })
298	+	}
299	+
300	+	/// Jobs in an order where each comes after what it depends on (ties by name).
301	+	pub fn order(&self) -> Vec<&Job> {
302	+	let mut done: BTreeSet<&str> = BTreeSet::new();
303	+	let mut out = Vec::with_capacity(self.jobs.len());
304	+	while out.len() < self.jobs.len() {
305	+	let before = out.len();
306	+	for j in self.jobs.values() {
307	+	if !done.contains(j.name.as_str()) && j.depends_on.iter().all(\|d\| done.contains(d.as_str())) {
308	+	out.push(j);
309	+	}
310	+	}
311	+	for j in &out[before..] {
312	+	done.insert(j.name.as_str());
313	+	}
314	+	if out.len() == before {
315	+	break; // validated pipelines have no cycles
316	+	}
317	+	}
318	+	out
319	+	}
320	+
321	+	/// Each job's depth: 0 without dependencies, else one more than its deepest dependency.
322	+	pub fn depths(&self) -> BTreeMap<&str, usize> {
323	+	let mut depth: BTreeMap<&str, usize> = BTreeMap::new();
324	+	for j in self.order() {
325	+	let d = j.depends_on.iter().map(\|x\| depth.get(x.as_str()).copied().unwrap_or(0) + 1).max().unwrap_or(0);
326	+	depth.insert(j.name.as_str(), d);
327	+	}
328	+	depth
329	+	}
330	+	}
331	+
332	+	fn find_cycle<'a>(graph: &BTreeMap<&'a str, Vec<&'a str>>) -> Option<Vec<&'a str>> {
333	+	#[derive(Clone, Copy, PartialEq)]
334	+	enum Mark {
335	+	Visiting,
336	+	Done,
337	+	}
338	+	fn visit<'a>(
339	+	n: &'a str,
340	+	graph: &BTreeMap<&'a str, Vec<&'a str>>,
341	+	marks: &mut BTreeMap<&'a str, Mark>,
342	+	stack: &mut Vec<&'a str>,
343	+	) -> Option<Vec<&'a str>> {
344	+	match marks.get(n) {
345	+	Some(Mark::Done) => return None,
346	+	Some(Mark::Visiting) => {
347	+	let at = stack.iter().position(\|s\| *s == n).unwrap_or(0);
348	+	let mut c = stack[at..].to_vec();
349	+	c.push(n);
350	+	return Some(c);
351	+	}
352	+	None => {}
353	+	}
354	+	marks.insert(n, Mark::Visiting);
355	+	stack.push(n);
356	+	for d in graph.get(n).into_iter().flatten() {
357	+	if graph.contains_key(d) && *d != n {
358	+	if let Some(c) = visit(d, graph, marks, stack) {
359	+	return Some(c);
360	+	}
361	+	}
362	+	}
363	+	stack.pop();
364	+	marks.insert(n, Mark::Done);
365	+	None
366	+	}
367	+	let mut marks = BTreeMap::new();
368	+	for n in graph.keys() {
369	+	let mut stack = Vec::new();
370	+	if let Some(c) = visit(n, graph, &mut marks, &mut stack) {
371	+	return Some(c);
372	+	}
373	+	}
374	+	None
375	+	}
376	+
377	+	fn block<'a>(ctx: &mut Ctx, node: &'a KdlNode) -> &'a [KdlNode] {
378	+	match node.children() {
379	+	Some(d) => d.nodes(),
380	+	None => {
381	+	ctx.err(node.span(), format!("`{}` needs a block {{ … }}", node.name().value()));
382	+	&[]
383	+	}
384	+	}
385	+	}
386	+
387	+	fn parse_sandbox(ctx: &mut Ctx, node: &KdlNode) -> Option<Sandbox> {
388	+	let name = ctx.one_string(node)?;
389	+	let named = ctx.name_ok(node.span(), "sandbox", &name);
390	+	let (mut runtime, mut rootfs, mut memory, mut cpus, mut network) =
391	+	(Some(Runtime::Microvm), None, Some(1024u32), Some(1u32), Some(Network::Isolated));
392	+	let mut seen = BTreeSet::new();
393	+	for c in block(ctx, node) {
394	+	let key = c.name().value();
395	+	if !seen.insert(key.to_string()) {
396	+	ctx.err(c.span(), format!("`{key}` is given twice"));
397	+	}
398	+	ctx.no_children(c);
399	+	match key {
400	+	"runtime" => {
401	+	runtime = match ctx.one_string(c).as_deref() {
402	+	Some("microvm") => Some(Runtime::Microvm),
403	+	Some("syscall") \| Some("gvisor") => Some(Runtime::Syscall),
404	+	Some("wasm") => {
405	+	ctx.err(c.span(), "runtime \"wasm\" is not in M0 yet; use \"microvm\" or \"syscall\"");
406	+	None
407	+	}
408	+	Some(other) => {
409	+	ctx.err(c.span(), format!("runtime {other:?}: expected \"microvm\", \"syscall\" or \"wasm\""));
410	+	None
411	+	}
412	+	None => None,
413	+	}
414	+	}
415	+	"rootfs" => {
416	+	rootfs = match ctx.one_string(c) {
417	+	Some(r) => match r.strip_prefix("cas://blob/") {
418	+	Some(h) if h.len() == 64 && h.bytes().all(\|b\| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) => {
419	+	Some(h.to_string())
420	+	}
421	+	_ => {
422	+	ctx.err(c.span(), format!("rootfs {r:?}: expected \"cas://blob/\" and 64 lower-case hex digits"));
423	+	None
424	+	}
425	+	},
426	+	None => None,
427	+	}
428	+	}
429	+	"memory-limit" => {
430	+	memory = match ctx.one_string(c) {
431	+	Some(m) => match parse_memory(&m) {
432	+	Some(mib) if (MIN_MEMORY_MIB..=MAX_MEMORY_MIB).contains(&mib) => Some(mib as u32),
433	+	Some(_) => {
434	+	ctx.err(c.span(), format!("memory-limit {m:?} must be from 64MiB to 16GiB"));
435	+	None
436	+	}
437	+	None => {
438	+	ctx.err(c.span(), format!("memory-limit {m:?}: expected a whole number of KiB, MiB or GiB, as \"4GiB\""));
439	+	None
440	+	}
441	+	},
442	+	None => None,
443	+	}
444	+	}
445	+	"cpu-limit" => {
446	+	cpus = match ctx.one_integer(c) {
447	+	Some(n) if (1..=MAX_CPUS).contains(&n) => Some(n as u32),
448	+	Some(n) => {
449	+	ctx.err(c.span(), format!("cpu-limit {n} must be from 1 to {MAX_CPUS}"));
450	+	None
451	+	}
452	+	None => None,
453	+	}
454	+	}
455	+	"network" => {
456	+	network = match ctx.one_string(c).as_deref() {
457	+	Some("isolated") => Some(Network::Isolated),
458	+	Some("loopback") \| Some("egress") => {
459	+	ctx.err(c.span(), "only network \"isolated\" is in M0");
460	+	None
461	+	}
462	+	Some(other) => {
463	+	ctx.err(c.span(), format!("network {other:?}: expected \"isolated\", \"loopback\" or \"egress\""));
464	+	None
465	+	}
466	+	None => None,
467	+	}
468	+	}
469	+	other => ctx.err(c.span(), format!("unknown sandbox setting `{other}`")),
470	+	}
471	+	}
472	+	if rootfs.is_none() && !seen.contains("rootfs") {
473	+	ctx.err(node.span(), format!("sandbox {name:?} needs a rootfs"));
474	+	}
475	+	if !named {
476	+	return None;
477	+	}
478	+	Some(Sandbox { name, runtime: runtime?, rootfs: rootfs?, memory_mib: memory?, cpus: cpus?, network: network? })
479	+	}
480	+
481	+	fn parse_job(ctx: &mut Ctx, node: &KdlNode) -> Option<Job> {
482	+	let name = ctx.one_string(node)?;
483	+	let named = ctx.name_ok(node.span(), "job", &name);
484	+	let mut job = Job { name: name.clone(), sandbox: String::new(), scope: vec![], depends_on: vec![], env: BTreeMap::new(), exec: vec![] };
485	+	let mut seen = BTreeSet::new();
486	+	for c in block(ctx, node) {
487	+	let key = c.name().value();
488	+	if key != "scope" && key != "depends-on" && !seen.insert(key.to_string()) {
489	+	ctx.err(c.span(), format!("`{key}` is given twice"));
490	+	}
491	+	match key {
492	+	"sandbox" => {
493	+	ctx.no_children(c);
494	+	job.sandbox = ctx.one_string(c).unwrap_or_default();
495	+	}
496	+	"scope" => {
497	+	ctx.no_children(c);
498	+	for g in ctx.strings(c) {
499	+	match globset::Glob::new(&g) {
500	+	Ok(_) => job.scope.push(g),
501	+	Err(e) => ctx.err(c.span(), format!("scope {g:?}: {e}")),
502	+	}
503	+	}
504	+	}
505	+	"depends-on" => {
506	+	ctx.no_children(c);
507	+	job.depends_on.extend(ctx.strings(c));
508	+	}
509	+	"env" => {
510	+	if !c.entries().is_empty() {
511	+	ctx.err(c.span(), "`env` takes a block of NAME \"value\" lines");
512	+	}
513	+	for e in block(ctx, c) {
514	+	let k = e.name().value().to_string();
515	+	let valid = !k.is_empty()
516	+	&& !k.starts_with(\|c: char\| c.is_ascii_digit())
517	+	&& k.bytes().all(\|b\| b.is_ascii_alphanumeric() \|\| b == b'_');
518	+	if !valid {
519	+	ctx.err(e.span(), format!("environment name {k:?} must be letters, digits and _"));
520	+	continue;
521	+	}
522	+	ctx.no_children(e);
523	+	if let Some(v) = ctx.one_string(e) {
524	+	if job.env.insert(k.clone(), v).is_some() {
525	+	ctx.err(e.span(), format!("environment {k} is set twice"));
526	+	}
527	+	}
528	+	}
529	+	}
530	+	"exec" => {
531	+	ctx.no_children(c);
532	+	job.exec = ctx.strings(c);
533	+	if job.exec.is_empty() {
534	+	ctx.err(c.span(), "`exec` needs a program and its arguments");
535	+	} else if job.exec.iter().any(\|a\| a.contains('\0')) {
536	+	ctx.err(c.span(), "`exec` arguments cannot contain NUL");
537	+	}
538	+	}
539	+	n if LATER_JOB.contains(&n) => ctx.err(c.span(), format!("`{n}` is not in M0 yet")),
540	+	other => ctx.err(c.span(), format!("unknown job setting `{other}`")),
541	+	}
542	+	}
543	+	if job.sandbox.is_empty() && !seen.contains("sandbox") {
544	+	ctx.err(node.span(), format!("job {name:?} needs a sandbox"));
545	+	}
546	+	if job.exec.is_empty() && !seen.contains("exec") {
547	+	ctx.err(node.span(), format!("job {name:?} needs an exec"));
548	+	}
549	+	job.depends_on.sort();
550	+	job.depends_on.dedup();
551	+	// Kept even when incomplete, so its references are checked too; the
552	+	// problems above already make the pipeline invalid.
553	+	named.then_some(job)
554	+	}

1	+	//! Execution receipts (spec §6.1), written by the host after a job ends,
2	+	//! never by the job itself.
3	+
4	+	use crate::canonical::{object_hash, Hash};
5	+	use oxy_core::cbor::{encode, Value};
6	+	use serde::{Deserialize, Serialize};
7	+
8	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
9	+	pub struct Receipt {
10	+	pub job_name: String,
11	+	#[serde(with = "hex32")]
12	+	pub pipeline_hash: Hash,
13	+	#[serde(with = "hex32")]
14	+	pub scope_tree_hash: Hash,
15	+	#[serde(with = "hex32s")]
16	+	pub inputs: Vec<Hash>,
17	+	#[serde(with = "hex32")]
18	+	pub environment: Hash,
19	+	#[serde(with = "hex32s")]
20	+	pub outputs: Vec<Hash>,
21	+	pub exit_code: u32,
22	+	pub duration_ms: u64,
23	+	}
24	+
25	+	impl Receipt {
26	+	pub fn encode(&self) -> Vec<u8> {
27	+	let refs = \|v: &[Hash]\| Value::Array(v.iter().map(\|h\| Value::Bytes(h.to_vec())).collect());
28	+	encode(&Value::map([
29	+	(0, Value::Uint(1)),
30	+	(1, Value::Text(self.job_name.clone())),
31	+	(2, Value::Bytes(self.pipeline_hash.to_vec())),
32	+	(3, Value::Bytes(self.scope_tree_hash.to_vec())),
33	+	(4, refs(&self.inputs)),
34	+	(5, Value::Bytes(self.environment.to_vec())),
35	+	(6, refs(&self.outputs)),
36	+	(7, Value::Uint(self.exit_code as u64)),
37	+	(8, Value::Uint(self.duration_ms)),
38	+	]))
39	+	}
40	+
41	+	/// `BLAKE3("oxy1 receipt " \|\| len \|\| 0x00 \|\| body)`.
42	+	pub fn hash(&self) -> Hash {
43	+	object_hash("receipt", &self.encode())
44	+	}
45	+	}
46	+
47	+	/// What a job ran in (the receipt's `environment`): the pinned pieces of
48	+	/// the isolation tier, the rootfs, argv and env.
49	+	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
50	+	pub struct Environment {
51	+	pub runtime: String,
52	+	/// Hashes of the tier's pinned parts (hypervisor, kernel, init), by name.
53	+	pub parts: Vec<(String, String)>,
54	+	pub rootfs: String,
55	+	pub argv: Vec<String>,
56	+	pub env: Vec<(String, String)>,
57	+	}
58	+
59	+	impl Environment {
60	+	pub fn hash(&self) -> Hash {
61	+	let t = \|s: &str\| Value::Text(s.to_string());
62	+	let pairs = \|v: &[(String, String)]\| Value::Array(v.iter().map(\|(a, b)\| Value::Array(vec![t(a), t(b)])).collect());
63	+	let body = encode(&Value::map([
64	+	(0, t(&self.runtime)),
65	+	(1, pairs(&self.parts)),
66	+	(2, t(&self.rootfs)),
67	+	(3, Value::Array(self.argv.iter().map(\|a\| t(a)).collect())),
68	+	(4, pairs(&self.env)),
69	+	]));
70	+	object_hash("environment", &body)
71	+	}
72	+	}
73	+
74	+	mod hex32 {
75	+	use serde::{Deserialize, Deserializer, Serializer};
76	+	pub fn serialize<S: Serializer>(h: &[u8; 32], s: S) -> Result<S::Ok, S::Error> {
77	+	s.serialize_str(&hex::encode(h))
78	+	}
79	+	pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 32], D::Error> {
80	+	let s = String::deserialize(d)?;
81	+	let v = hex::decode(&s).map_err(serde::de::Error::custom)?;
82	+	v.try_into().map_err(\|_\| serde::de::Error::custom("expected 32 bytes"))
83	+	}
84	+	}
85	+
86	+	mod hex32s {
87	+	use serde::{Deserialize, Deserializer, Serializer};
88	+	pub fn serialize<S: Serializer>(v: &[[u8; 32]], s: S) -> Result<S::Ok, S::Error> {
89	+	s.collect_seq(v.iter().map(hex::encode))
90	+	}
91	+	pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Vec<[u8; 32]>, D::Error> {
92	+	let v = Vec::<String>::deserialize(d)?;
93	+	v.into_iter()
94	+	.map(\|s\| {
95	+	let b = hex::decode(&s).map_err(serde::de::Error::custom)?;
96	+	b.try_into().map_err(\|_\| serde::de::Error::custom("expected 32 bytes"))
97	+	})
98	+	.collect()
99	+	}
100	+	}

1	+	//! PID 1 of a job's microVM (docs/m0.md §2). It mounts the read-only
2	+	//! rootfs and the workspace, runs the job's argv as an unprivileged user
3	+	//! with exactly the job's environment, frames the output on the serial
4	+	//! console for the host, and powers off. It never decides anything the
5	+	//! host has not: the job comes from `/job.json` in the initramfs the host
6	+	//! built, and the host enforces the time limit.
7	+	//!
8	+	//! Frames, one per line on the console: `\x1eO ` stdout, `\x1eE ` stderr,
9	+	//! `\x1eX <code>` the exit code (128 + signal when killed), `\x1eF <text>`
10	+	//! a failure of init itself.
11	+
12	+	use serde::Deserialize;
13	+	use std::ffi::CString;
14	+	use std::io::{Read, Write};
15	+	use std::os::fd::FromRawFd;
16	+	use std::sync::{Arc, Mutex};
17	+
18	+	#[derive(Deserialize)]
19	+	struct JobSpec {
20	+	argv: Vec<String>,
21	+	env: Vec<(String, String)>,
22	+	}
23	+
24	+	const PATH: &str = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin";
25	+
26	+	fn main() {
27	+	let console = Arc::new(Mutex::new(std::io::stdout()));
28	+	let code = match run(&console) {
29	+	Ok(code) => code,
30	+	Err(e) => {
31	+	frame(&console, b'F', e.as_bytes());
32	+	125
33	+	}
34	+	};
35	+	frame(&console, b'X', code.to_string().as_bytes());
36	+	unsafe {
37	+	libc::sync();
38	+	// Firecracker ends the VM when the guest reboots (reboot=k).
39	+	libc::reboot(libc::LINUX_REBOOT_CMD_RESTART);
40	+	}
41	+	}
42	+
43	+	fn frame(console: &Mutex<std::io::Stdout>, kind: u8, text: &[u8]) {
44	+	let mut out = console.lock().unwrap_or_else(\|p\| p.into_inner());
45	+	let mut line = Vec::with_capacity(text.len() + 4);
46	+	line.extend_from_slice(&[0x1e, kind, b' ']);
47	+	line.extend_from_slice(text);
48	+	line.push(b'\n');
49	+	let _ = out.write_all(&line);
50	+	let _ = out.flush();
51	+	}
52	+
53	+	fn cstr(s: &str) -> Result<CString, String> {
54	+	CString::new(s).map_err(\|_\| format!("{s:?} contains NUL"))
55	+	}
56	+
57	+	fn mount(src: &str, target: &str, fstype: &str, flags: libc::c_ulong, data: &str) -> Result<(), String> {
58	+	let (s, t, f, d) = (cstr(src)?, cstr(target)?, cstr(fstype)?, cstr(data)?);
59	+	let rc = unsafe { libc::mount(s.as_ptr(), t.as_ptr(), f.as_ptr(), flags, d.as_ptr() as *const libc::c_void) };
60	+	if rc != 0 {
61	+	return Err(format!("mount {src} on {target}: {}", std::io::Error::last_os_error()));
62	+	}
63	+	Ok(())
64	+	}
65	+
66	+	fn run(console: &Arc<Mutex<std::io::Stdout>>) -> Result<i32, String> {
67	+	let job: JobSpec = serde_json::from_slice(&std::fs::read("/job.json").map_err(\|e\| format!("read /job.json: {e}"))?)
68	+	.map_err(\|e\| format!("parse /job.json: {e}"))?;
69	+	if job.argv.is_empty() {
70	+	return Err("empty argv".into());
71	+	}
72	+	let nosuid = libc::MS_NOSUID \| libc::MS_NODEV;
73	+	std::fs::create_dir_all("/dev").ok();
74	+	mount("devtmpfs", "/dev", "devtmpfs", libc::MS_NOSUID, "")?;
75	+	std::fs::create_dir_all("/newroot").map_err(\|e\| e.to_string())?;
76	+	mount("/dev/vda", "/newroot", "ext4", libc::MS_RDONLY \| libc::MS_NODEV, "")?;
77	+	mount("/dev/vdb", "/newroot/src", "ext4", nosuid, "")?;
78	+	mount("tmpfs", "/newroot/tmp", "tmpfs", nosuid, "mode=1777")?;
79	+	mount("proc", "/newroot/proc", "proc", nosuid \| libc::MS_NOEXEC, "")?;
80	+	mount("devtmpfs", "/newroot/dev", "devtmpfs", libc::MS_NOSUID, "")?;
81	+	if std::path::Path::new("/newroot/run").is_dir() {
82	+	mount("tmpfs", "/newroot/run", "tmpfs", nosuid, "mode=755")?;
83	+	}
84	+	// The job runs as whoever owns the workspace (the host's builder), never root.
85	+	let (uid, gid) = {
86	+	let mut st: libc::stat = unsafe { std::mem::zeroed() };
87	+	let p = cstr("/newroot/src")?;
88	+	if unsafe { libc::stat(p.as_ptr(), &mut st) } != 0 {
89	+	return Err("stat /src".into());
90	+	}
91	+	if st.st_uid == 0 {
92	+	(1000, 1000)
93	+	} else {
94	+	(st.st_uid, st.st_gid)
95	+	}
96	+	};
97	+	let root = cstr("/newroot")?;
98	+	if unsafe { libc::chroot(root.as_ptr()) } != 0 \|\| std::env::set_current_dir("/src").is_err() {
99	+	return Err(format!("enter the rootfs: {}", std::io::Error::last_os_error()));
100	+	}
101	+
102	+	let argv: Vec<CString> = job.argv.iter().map(\|a\| cstr(a)).collect::<Result<_, _>>()?;
103	+	let mut env: Vec<CString> = vec![cstr(&format!("PATH={PATH}"))?, cstr("HOME=/tmp")?];
104	+	for (k, v) in &job.env {
105	+	env.push(cstr(&format!("{k}={v}"))?);
106	+	}
107	+	let mut out = [0; 2];
108	+	let mut err = [0; 2];
109	+	if unsafe { libc::pipe2(out.as_mut_ptr(), libc::O_CLOEXEC) } != 0 \|\| unsafe { libc::pipe2(err.as_mut_ptr(), libc::O_CLOEXEC) } != 0 {
110	+	return Err("pipe".into());
111	+	}
112	+	let pid = unsafe { libc::fork() };
113	+	if pid < 0 {
114	+	return Err("fork".into());
115	+	}
116	+	if pid == 0 {
117	+	// Child: no way back to root, then the program itself (execvpe: no shell).
118	+	unsafe {
119	+	let null = libc::open(c"/dev/null".as_ptr(), libc::O_RDONLY);
120	+	libc::dup2(null, 0);
121	+	libc::dup2(out[1], 1);
122	+	libc::dup2(err[1], 2);
123	+	libc::setsid();
124	+	if libc::setgroups(0, std::ptr::null()) != 0 \|\| libc::setgid(gid) != 0 \|\| libc::setuid(uid) != 0 {
125	+	libc::_exit(126);
126	+	}
127	+	let mut a: Vec<*const libc::c_char> = argv.iter().map(\|c\| c.as_ptr()).collect();
128	+	a.push(std::ptr::null());
129	+	let mut e: Vec<*const libc::c_char> = env.iter().map(\|c\| c.as_ptr()).collect();
130	+	e.push(std::ptr::null());
131	+	libc::execvpe(a[0], a.as_ptr(), e.as_ptr());
132	+	let msg = format!("flow: cannot run {:?}: {}\n", job.argv[0], std::io::Error::last_os_error());
133	+	libc::write(2, msg.as_ptr() as *const libc::c_void, msg.len());
134	+	libc::_exit(127);
135	+	}
136	+	}
137	+	unsafe {
138	+	libc::close(out[1]);
139	+	libc::close(err[1]);
140	+	}
141	+	let readers: Vec<_> = [(out[0], b'O'), (err[0], b'E')]
142	+	.into_iter()
143	+	.map(\|(fd, kind)\| {
144	+	let console = console.clone();
145	+	std::thread::spawn(move \|\| pump(unsafe { std::fs::File::from_raw_fd(fd) }, kind, &console))
146	+	})
147	+	.collect();
148	+	let mut status = 0;
149	+	// Reap everything (we are PID 1) until the job itself exits.
150	+	loop {
151	+	let r = unsafe { libc::waitpid(-1, &mut status, 0) };
152	+	if r == pid \|\| r < 0 {
153	+	break;
154	+	}
155	+	}
156	+	for r in readers {
157	+	let _ = r.join();
158	+	}
159	+	Ok(if libc::WIFEXITED(status) { libc::WEXITSTATUS(status) } else { 128 + libc::WTERMSIG(status) })
160	+	}
161	+
162	+	/// Frame a stream line by line; a long line is cut at 16 KiB.
163	+	fn pump(mut f: std::fs::File, kind: u8, console: &Mutex<std::io::Stdout>) {
164	+	let mut buf = [0u8; 8192];
165	+	let mut line: Vec<u8> = Vec::new();
166	+	loop {
167	+	let n = match f.read(&mut buf) {
168	+	Ok(0) \| Err(_) => break,
169	+	Ok(n) => n,
170	+	};
171	+	for &b in &buf[..n] {
172	+	if b == b'\n' \|\| line.len() >= 16 * 1024 {
173	+	frame(console, kind, &line);
174	+	line.clear();
175	+	if b == b'\n' {
176	+	continue;
177	+	}
178	+	}
179	+	line.push(b);
180	+	}
181	+	}
182	+	if !line.is_empty() {
183	+	frame(console, kind, &line);
184	+	}
185	+	}

1	+	[package]
2	+	name = "flow"
3	+	version.workspace = true
4	+	edition.workspace = true
5	+	license.workspace = true
6	+	description = "OxygenFlow: check pipelines, run them locally, and run jobs for OxygenHub"
7	+
8	+	[[bin]]
9	+	name = "flow"
10	+	path = "src/main.rs"
11	+
12	+	[dependencies]
13	+	flow-core.workspace = true
14	+	oxy-core.workspace = true
15	+	serde.workspace = true
16	+	serde_json.workspace = true
17	+	blake3.workspace = true
18	+	hex.workspace = true
19	+	anyhow.workspace = true
20	+	clap = { version = "4", features = ["derive"] }
21	+	toml = "0.8"
22	+	ureq = { version = "2", features = ["json"] }
23	+	ignore = "0.4"
24	+	libc = "0.2"
25	+
26	+	[dev-dependencies]
27	+	tempfile.workspace = true

1	+	//! A minimal `newc` cpio writer for the per-job initramfs: regular files
2	+	//! at the archive root, owned by root.
3	+
4	+	pub struct Entry<'a> {
5	+	pub name: &'a str,
6	+	pub mode: u32,
7	+	pub data: &'a [u8],
8	+	}
9	+
10	+	fn header(out: &mut Vec<u8>, ino: u32, mode: u32, size: usize, name: &str) {
11	+	let fields = [ino, mode, 0, 0, 1, 0, size as u32, 0, 0, 0, 0, (name.len() + 1) as u32, 0];
12	+	out.extend_from_slice(b"070701");
13	+	for f in fields {
14	+	out.extend_from_slice(format!("{f:08X}").as_bytes());
15	+	}
16	+	out.extend_from_slice(name.as_bytes());
17	+	out.push(0);
18	+	pad(out);
19	+	}
20	+
21	+	fn pad(out: &mut Vec<u8>) {
22	+	while !out.len().is_multiple_of(4) {
23	+	out.push(0);
24	+	}
25	+	}
26	+
27	+	pub fn newc(entries: &[Entry]) -> Vec<u8> {
28	+	let mut out = Vec::new();
29	+	for (i, e) in entries.iter().enumerate() {
30	+	header(&mut out, i as u32 + 1, 0o100000 \| e.mode, e.data.len(), e.name);
31	+	out.extend_from_slice(e.data);
32	+	pad(&mut out);
33	+	}
34	+	header(&mut out, 0, 0, 0, "TRAILER!!!");
35	+	out
36	+	}
37	+
38	+	#[cfg(test)]
39	+	mod tests {
40	+	use super::*;
41	+
42	+	#[test]
43	+	fn system_cpio_reads_it() {
44	+	let dir = tempfile::tempdir().unwrap();
45	+	let archive = newc(&[Entry { name: "init", mode: 0o755, data: b"#!x\n" }, Entry { name: "job.json", mode: 0o644, data: b"{}" }]);
46	+	std::fs::write(dir.path().join("a.cpio"), &archive).unwrap();
47	+	let listed = std::process::Command::new("cpio").args(["-it", "--quiet", "-F"]).arg(dir.path().join("a.cpio")).output();
48	+	if let Ok(o) = listed {
49	+	if o.status.success() {
50	+	assert_eq!(String::from_utf8_lossy(&o.stdout), "init\njob.json\n");
51	+	}
52	+	}
53	+	assert!(archive.len().is_multiple_of(4));
54	+	assert!(archive.windows(10).any(\|w\| w == b"TRAILER!!!"));
55	+	}
56	+	}

1	+	//! `$FLOW_HOME` (default `~/.local/share/flow`): the pinned parts of each
2	+	//! isolation tier, rootfs images by hash, scratch space and the runner's
3	+	//! configuration, all described by `flow.toml`.
4	+	//!
5	+	//! ```toml
6	+	//! [microvm]
7	+	//! firecracker = { path = "bin/firecracker", blake3 = "…" }
8	+	//! kernel = { path = "kernel/vmlinux", blake3 = "…" }
9	+	//! init = { path = "bin/flow-init", blake3 = "…" }
10	+	//!
11	+	//! [runner] # only for `flow runner`
12	+	//! hub = "https://hub.example:7343"
13	+	//! token = "…" # a runner token from the hub's [flow] runner_tokens
14	+	//! ca = "/path/to/hub-ca.pem" # when the hub's certificate is self-signed
15	+	//! name = "earth-1"
16	+	//! ```
17	+
18	+	use anyhow::{bail, Context, Result};
19	+	use serde::Deserialize;
20	+	use std::collections::HashMap;
21	+	use std::path::{Path, PathBuf};
22	+	use std::sync::Mutex;
23	+
24	+	#[derive(Clone, Debug, Deserialize)]
25	+	pub struct Pinned {
26	+	pub path: PathBuf,
27	+	pub blake3: String,
28	+	}
29	+
30	+	#[derive(Clone, Debug, Deserialize)]
31	+	pub struct MicrovmConfig {
32	+	pub firecracker: Pinned,
33	+	pub kernel: Pinned,
34	+	pub init: Pinned,
35	+	}
36	+
37	+	#[derive(Clone, Debug, Deserialize)]
38	+	pub struct RunnerConfig {
39	+	pub hub: String,
40	+	pub token: String,
41	+	pub ca: Option<PathBuf>,
42	+	pub name: Option<String>,
43	+	}
44	+
45	+	#[derive(Clone, Debug, Default, Deserialize)]
46	+	pub struct Config {
47	+	pub microvm: Option<MicrovmConfig>,
48	+	pub runner: Option<RunnerConfig>,
49	+	}
50	+
51	+	pub struct Home {
52	+	pub root: PathBuf,
53	+	pub config: Config,
54	+	/// Files already verified, by path, with the (len, mtime) they had then.
55	+	verified: Mutex<HashMap<PathBuf, (u64, std::time::SystemTime)>>,
56	+	}
57	+
58	+	impl Home {
59	+	pub fn locate(explicit: Option<&Path>) -> Result<Home> {
60	+	let root = match explicit {
61	+	Some(p) => p.to_path_buf(),
62	+	None => match std::env::var_os("FLOW_HOME") {
63	+	Some(p) => PathBuf::from(p),
64	+	None => PathBuf::from(std::env::var_os("HOME").context("HOME is not set")?).join(".local/share/flow"),
65	+	},
66	+	};
67	+	let file = root.join("flow.toml");
68	+	let config = if file.is_file() {
69	+	let text = std::fs::read_to_string(&file).with_context(\|\| format!("read {}", file.display()))?;
70	+	toml::from_str(&text).with_context(\|\| format!("parse {}", file.display()))?
71	+	} else {
72	+	Config::default()
73	+	};
74	+	for d in ["cas", "work", "repos"] {
75	+	std::fs::create_dir_all(root.join(d)).with_context(\|\| format!("create {}", root.join(d).display()))?;
76	+	}
77	+	Ok(Home { root, config, verified: Mutex::new(HashMap::new()) })
78	+	}
79	+
80	+	pub fn resolve(&self, p: &Path) -> PathBuf {
81	+	if p.is_absolute() {
82	+	p.to_path_buf()
83	+	} else {
84	+	self.root.join(p)
85	+	}
86	+	}
87	+
88	+	/// Where a rootfs image with this hash lives.
89	+	pub fn rootfs(&self, hash: &str) -> PathBuf {
90	+	self.root.join("cas").join(format!("{hash}.ext4"))
91	+	}
92	+
93	+	/// A pinned file, checked against its BLAKE3 before use (again only
94	+	/// when its size or modification time changed since).
95	+	pub fn verified(&self, path: &Path, blake3_hex: &str) -> Result<PathBuf> {
96	+	let meta = std::fs::metadata(path).with_context(\|\| format!("{} is missing", path.display()))?;
97	+	let stamp = (meta.len(), meta.modified()?);
98	+	if self.verified.lock().unwrap().get(path) == Some(&stamp) {
99	+	return Ok(path.to_path_buf());
100	+	}
101	+	let got = hash_file(path)?;
102	+	if got != blake3_hex {
103	+	bail!("{} has BLAKE3 {got}, but {blake3_hex} is pinned", path.display());
104	+	}
105	+	self.verified.lock().unwrap().insert(path.to_path_buf(), stamp);
106	+	Ok(path.to_path_buf())
107	+	}
108	+	}
109	+
110	+	pub fn hash_file(path: &Path) -> Result<String> {
111	+	let mut h = blake3::Hasher::new();
112	+	h.update_reader(std::fs::File::open(path).with_context(\|\| format!("open {}", path.display()))?)?;
113	+	Ok(h.finalize().to_hex().to_string())
114	+	}

1	+	//! `flow`: check pipelines, run them locally, and run jobs for OxygenHub
2	+	//! (docs/spec.md, docs/m0.md).
3	+
4	+	mod cpio;
5	+	mod home;
6	+	mod microvm;
7	+	mod runner;
8	+	mod workspace;
9	+
10	+	use anyhow::{bail, Context, Result};
11	+	use clap::{Parser, Subcommand};
12	+	use flow_core::api::Record;
13	+	use flow_core::{Pipeline, Runtime, PIPELINE_PATH};
14	+	use home::Home;
15	+	use std::collections::BTreeSet;
16	+	use std::path::{Path, PathBuf};
17	+	use std::time::Duration;
18	+
19	+	#[derive(Parser)]
20	+	#[command(name = "flow", version, about = "OxygenFlow: hermetic pipelines for Oxygen repositories")]
21	+	struct Cli {
22	+	/// The Flow home (default $FLOW_HOME, else ~/.local/share/flow).
23	+	#[arg(long, global = true)]
24	+	home: Option<PathBuf>,
25	+	#[command(subcommand)]
26	+	cmd: Cmd,
27	+	}
28	+
29	+	#[derive(Subcommand)]
30	+	enum Cmd {
31	+	/// Validate a pipeline and print its jobs and hash
32	+	Check {
33	+	/// The pipeline file (default: .flow/pipeline.kdl)
34	+	file: Option<PathBuf>,
35	+	},
36	+	/// Print a file's BLAKE3, for pinning in flow.toml
37	+	Hash { file: PathBuf },
38	+	/// Rootfs images for sandboxes
39	+	Rootfs {
40	+	#[command(subcommand)]
41	+	cmd: RootfsCmd,
42	+	},
43	+	/// Run the pipeline on this directory's files, locally
44	+	Run {
45	+	/// Only these jobs (and what they depend on)
46	+	jobs: Vec<String>,
47	+	/// The directory to build (default: the current one)
48	+	#[arg(long, default_value = ".")]
49	+	dir: PathBuf,
50	+	/// Minutes before a job is stopped
51	+	#[arg(long, default_value_t = 60)]
52	+	timeout: u64,
53	+	},
54	+	/// Claim and run jobs for the hub in flow.toml's [runner]
55	+	Runner,
56	+	}
57	+
58	+	#[derive(Subcommand)]
59	+	enum RootfsCmd {
60	+	/// Make an ext4 rootfs image from a root file system tarball and store it by hash
61	+	Import {
62	+	tarball: PathBuf,
63	+	/// Refuse the tarball unless its SHA-256 matches (as its publisher states it)
64	+	#[arg(long)]
65	+	sha256: Option<String>,
66	+	},
67	+	/// List the images this home holds
68	+	List,
69	+	}
70	+
71	+	fn main() {
72	+	let cli = Cli::parse();
73	+	if let Err(e) = run(cli) {
74	+	eprintln!("flow: {e:#}");
75	+	std::process::exit(1);
76	+	}
77	+	}
78	+
79	+	fn run(cli: Cli) -> Result<()> {
80	+	match cli.cmd {
81	+	Cmd::Check { file } => check(&file.unwrap_or_else(\|\| PathBuf::from(PIPELINE_PATH))),
82	+	Cmd::Hash { file } => {
83	+	println!("{}", home::hash_file(&file)?);
84	+	Ok(())
85	+	}
86	+	Cmd::Rootfs { cmd } => {
87	+	let home = Home::locate(cli.home.as_deref())?;
88	+	match cmd {
89	+	RootfsCmd::Import { tarball, sha256 } => rootfs_import(&home, &tarball, sha256.as_deref()),
90	+	RootfsCmd::List => {
91	+	for e in std::fs::read_dir(home.root.join("cas"))?.flatten() {
92	+	let size = e.metadata().map(\|m\| m.len()).unwrap_or(0);
93	+	if let Some(h) = e.file_name().to_string_lossy().strip_suffix(".ext4") {
94	+	println!("cas://blob/{h} {} MiB", size >> 20);
95	+	}
96	+	}
97	+	Ok(())
98	+	}
99	+	}
100	+	}
101	+	Cmd::Run { jobs, dir, timeout } => {
102	+	let home = Home::locate(cli.home.as_deref())?;
103	+	run_local(&home, &dir, &jobs, Duration::from_secs(timeout * 60))
104	+	}
105	+	Cmd::Runner => {
106	+	let home = Home::locate(cli.home.as_deref())?;
107	+	// The runner's oxy state (identity, user config) stays in its home.
108	+	if std::env::var_os("OXY_HOME").is_none() {
109	+	std::env::set_var("OXY_HOME", home.root.join("oxy"));
110	+	}
111	+	runner::serve(&home)
112	+	}
113	+	}
114	+	}
115	+
116	+	fn load(file: &Path) -> Result<Pipeline> {
117	+	let src = std::fs::read_to_string(file).with_context(\|\| format!("read {}", file.display()))?;
118	+	Pipeline::parse(&src).map_err(\|p\| anyhow::anyhow!("{} is not valid:\n{p}", file.display()))
119	+	}
120	+
121	+	fn check(file: &Path) -> Result<()> {
122	+	let p = load(file)?;
123	+	let depths = p.depths();
124	+	for j in p.order() {
125	+	let sb = &p.sandboxes[&j.sandbox];
126	+	let after = if j.depends_on.is_empty() { String::new() } else { format!(" after {}", j.depends_on.join(", ")) };
127	+	println!("{}{} {} [{}]{after}", " ".repeat(depths[j.name.as_str()]), j.name, j.exec.join(" "), sb.runtime.as_str());
128	+	}
129	+	println!("pipeline {}", hex::encode(p.hash()));
130	+	Ok(())
131	+	}
132	+
133	+	fn rootfs_import(home: &Home, tarball: &Path, sha256: Option<&str>) -> Result<()> {
134	+	if let Some(want) = sha256 {
135	+	let out = std::process::Command::new("sha256sum").arg(tarball).output().context("run sha256sum")?;
136	+	let got = String::from_utf8_lossy(&out.stdout).split_whitespace().next().unwrap_or("").to_string();
137	+	if got != want.to_lowercase() {
138	+	bail!("{} has SHA-256 {got}, expected {want}", tarball.display());
139	+	}
140	+	}
141	+	let tmp = home.root.join("work").join(format!("rootfs-{}", std::process::id()));
142	+	let _ = std::fs::remove_dir_all(&tmp);
143	+	let tree = tmp.join("tree");
144	+	std::fs::create_dir_all(&tree)?;
145	+	let result = (\|\| -> Result<String> {
146	+	let status = std::process::Command::new("tar").arg("-xf").arg(tarball).arg("-C").arg(&tree).arg("--no-same-owner").status()?;
147	+	if !status.success() {
148	+	bail!("tar could not unpack {}", tarball.display());
149	+	}
150	+	// Mount points flow-init needs (docs/m0.md §2).
151	+	for d in ["src", "tmp", "proc", "dev", "run"] {
152	+	std::fs::create_dir_all(tree.join(d))?;
153	+	}
154	+	let used: u64 = ignore::WalkBuilder::new(&tree)
155	+	.hidden(false)
156	+	.ignore(false)
157	+	.git_ignore(false)
158	+	.build()
159	+	.flatten()
160	+	.filter_map(\|e\| e.metadata().ok())
161	+	.map(\|m\| m.len().max(4096))
162	+	.sum();
163	+	let mib = used * 13 / 10 / (1 << 20) + 64;
164	+	let img = tmp.join("rootfs.ext4");
165	+	let status = std::process::Command::new("mkfs.ext4")
166	+	.args(["-q", "-F", "-L", "rootfs", "-E", "root_owner=0:0", "-d"])
167	+	.arg(&tree)
168	+	.arg(&img)
169	+	.arg(format!("{mib}M"))
170	+	.status()
171	+	.context("run mkfs.ext4")?;
172	+	if !status.success() {
173	+	bail!("mkfs.ext4 failed");
174	+	}
175	+	let hash = home::hash_file(&img)?;
176	+	std::fs::rename(&img, home.rootfs(&hash))?;
177	+	Ok(hash)
178	+	})();
179	+	let _ = std::fs::remove_dir_all(&tmp);
180	+	let hash = result?;
181	+	println!("rootfs \"cas://blob/{hash}\"");
182	+	Ok(())
183	+	}
184	+
185	+	fn run_local(home: &Home, dir: &Path, only: &[String], timeout: Duration) -> Result<()> {
186	+	let p = load(&dir.join(PIPELINE_PATH))?;
187	+	// The named jobs and everything they depend on.
188	+	let mut wanted: BTreeSet<String> = only.iter().cloned().collect();
189	+	for name in only {
190	+	if !p.jobs.contains_key(name) {
191	+	bail!("no job {name:?} in the pipeline");
192	+	}
193	+	}
194	+	if !only.is_empty() {
195	+	loop {
196	+	let more: Vec<String> = wanted.iter().flat_map(\|j\| p.jobs[j].depends_on.clone()).filter(\|d\| !wanted.contains(d)).collect();
197	+	if more.is_empty() {
198	+	break;
199	+	}
200	+	wanted.extend(more);
201	+	}
202	+	}
203	+	let scratch = home.root.join("work").join(format!("local-{}", std::process::id()));
204	+	let _ = std::fs::remove_dir_all(&scratch);
205	+	std::fs::create_dir_all(scratch.join("tree"))?;
206	+	workspace::copy_tree(dir, &scratch.join("tree"))?;
207	+	let mut failed: BTreeSet<String> = BTreeSet::new();
208	+	let mut worst = 0;
209	+	for job in p.order() {
210	+	if !only.is_empty() && !wanted.contains(&job.name) {
211	+	continue;
212	+	}
213	+	if let Some(d) = job.depends_on.iter().find(\|d\| failed.contains(*d)) {
214	+	println!("── {} skipped: {d} failed", job.name);
215	+	failed.insert(job.name.clone());
216	+	continue;
217	+	}
218	+	let sb = &p.sandboxes[&job.sandbox];
219	+	if sb.runtime != Runtime::Microvm {
220	+	bail!("job {}: runtime {} is not available in this build yet", job.name, sb.runtime.as_str());
221	+	}
222	+	println!("── {} ({})", job.name, job.exec.join(" "));
223	+	let js = scratch.join(&job.name);
224	+	std::fs::create_dir_all(&js)?;
225	+	// Each job starts from the same files: jobs do not share writes.
226	+	workspace::image(&scratch.join("tree"), &js.join("workspace.ext4"))?;
227	+	let out = microvm::run(home, sb, job, &js.join("workspace.ext4"), &js, timeout, &mut \|r: Record\| print_record(&r))?;
228	+	let _ = std::fs::remove_dir_all(&js);
229	+	match out.exit_code {
230	+	Some(0) => println!("── {} passed in {} ms", job.name, out.duration_ms),
231	+	other => {
232	+	println!(
233	+	"── {} failed ({}) in {} ms",
234	+	job.name,
235	+	other.map(\|c\| format!("exit {c}")).unwrap_or_else(\|\| "no exit code".into()),
236	+	out.duration_ms
237	+	);
238	+	failed.insert(job.name.clone());
239	+	worst = 1;
240	+	}
241	+	}
242	+	}
243	+	let _ = std::fs::remove_dir_all(&scratch);
244	+	if worst != 0 {
245	+	std::process::exit(worst);
246	+	}
247	+	Ok(())
248	+	}
249	+
250	+	fn print_record(r: &Record) {
251	+	match r.stream.as_str() {
252	+	"out" => println!(" {}", r.text),
253	+	"err" => eprintln!(" {}", r.text),
254	+	s => eprintln!(" [{s}] {}", r.text),
255	+	}
256	+	}

1	+	//! Tier 2 (spec §4.3, docs/m0.md §2): one Firecracker microVM per job,
2	+	//! no network device, the rootfs read-only, the workspace read-write,
3	+	//! `flow-init` as PID 1. The host reads the framed console, enforces
4	+	//! the time limit, and writes the receipt's environment.
5	+
6	+	use crate::home::Home;
7	+	use anyhow::{bail, Context, Result};
8	+	use flow_core::api::Record;
9	+	use flow_core::{Environment, Job, Sandbox};
10	+	use std::io::{BufRead, BufReader, Read};
11	+	use std::path::Path;
12	+	use std::sync::mpsc;
13	+	use std::time::{Duration, Instant};
14	+
15	+	pub struct Outcome {
16	+	/// None when the job never reported an exit (killed at the time limit, or the VM failed).
17	+	pub exit_code: Option<i32>,
18	+	pub duration_ms: u64,
19	+	pub environment: Environment,
20	+	pub timed_out: bool,
21	+	}
22	+
23	+	/// The receipt environment for a job in this tier.
24	+	pub fn environment(home: &Home, sandbox: &Sandbox, job: &Job) -> Result<Environment> {
25	+	let cfg = home.config.microvm.as_ref().context("flow.toml has no [microvm] section")?;
26	+	Ok(Environment {
27	+	runtime: "microvm".into(),
28	+	parts: vec![
29	+	("firecracker".into(), cfg.firecracker.blake3.clone()),
30	+	("kernel".into(), cfg.kernel.blake3.clone()),
31	+	("init".into(), cfg.init.blake3.clone()),
32	+	("memory_mib".into(), sandbox.memory_mib.to_string()),
33	+	("cpus".into(), sandbox.cpus.to_string()),
34	+	],
35	+	rootfs: sandbox.rootfs.clone(),
36	+	argv: job.exec.clone(),
37	+	env: job.env.iter().map(\|(k, v)\| (k.clone(), v.clone())).collect(),
38	+	})
39	+	}
40	+
41	+	/// Boot, run, collect. `scratch` is an empty directory this call may fill.
42	+	pub fn run(
43	+	home: &Home,
44	+	sandbox: &Sandbox,
45	+	job: &Job,
46	+	workspace: &Path,
47	+	scratch: &Path,
48	+	timeout: Duration,
49	+	sink: &mut dyn FnMut(Record),
50	+	) -> Result<Outcome> {
51	+	let cfg = home.config.microvm.as_ref().context("flow.toml has no [microvm] section")?;
52	+	let firecracker = home.verified(&home.resolve(&cfg.firecracker.path), &cfg.firecracker.blake3)?;
53	+	let kernel = home.verified(&home.resolve(&cfg.kernel.path), &cfg.kernel.blake3)?;
54	+	let init = home.verified(&home.resolve(&cfg.init.path), &cfg.init.blake3)?;
55	+	let rootfs = home.rootfs(&sandbox.rootfs);
56	+	if !rootfs.is_file() {
57	+	bail!("rootfs {} is not on this runner (import it with `flow rootfs import`)", sandbox.rootfs);
58	+	}
59	+	let rootfs = home.verified(&rootfs, &sandbox.rootfs)?;
60	+	let environment = environment(home, sandbox, job)?;
61	+
62	+	let spec = serde_json::json!({ "argv": job.exec, "env": job.env.iter().collect::<Vec<_>>() });
63	+	let init_bytes = std::fs::read(&init)?;
64	+	let spec_bytes = serde_json::to_vec(&spec)?;
65	+	let initrd = crate::cpio::newc(&[
66	+	crate::cpio::Entry { name: "init", mode: 0o755, data: &init_bytes },
67	+	crate::cpio::Entry { name: "job.json", mode: 0o644, data: &spec_bytes },
68	+	]);
69	+	let initrd_path = scratch.join("initrd.cpio");
70	+	std::fs::write(&initrd_path, initrd)?;
71	+	let vm = serde_json::json!({
72	+	"boot-source": {
73	+	"kernel_image_path": kernel,
74	+	"initrd_path": initrd_path,
75	+	"boot_args": "console=ttyS0 reboot=k panic=-1 pci=off quiet loglevel=0 rdinit=/init",
76	+	},
77	+	"drives": [
78	+	{ "drive_id": "rootfs", "path_on_host": rootfs, "is_root_device": false, "is_read_only": true },
79	+	{ "drive_id": "workspace", "path_on_host": workspace, "is_root_device": false, "is_read_only": false },
80	+	],
81	+	"machine-config": { "vcpu_count": sandbox.cpus, "mem_size_mib": sandbox.memory_mib },
82	+	});
83	+	let vm_path = scratch.join("vm.json");
84	+	std::fs::write(&vm_path, serde_json::to_vec_pretty(&vm)?)?;
85	+
86	+	let started = Instant::now();
87	+	let mut child = std::process::Command::new(&firecracker)
88	+	.arg("--no-api")
89	+	.arg("--config-file")
90	+	.arg(&vm_path)
91	+	.args(["--level", "Error"])
92	+	.current_dir(scratch)
93	+	.stdin(std::process::Stdio::null())
94	+	.stdout(std::process::Stdio::piped())
95	+	.stderr(std::process::Stdio::piped())
96	+	.spawn()
97	+	.with_context(\|\| format!("start {}", firecracker.display()))?;
98	+	let (tx, rx) = mpsc::channel::<(bool, Vec<u8>)>();
99	+	let out = child.stdout.take().unwrap();
100	+	let err = child.stderr.take().unwrap();
101	+	let tx2 = tx.clone();
102	+	let readers = [std::thread::spawn(move \|\| lines(out, true, tx)), std::thread::spawn(move \|\| lines(err, false, tx2))];
103	+	let mut exit_code = None;
104	+	let mut timed_out = false;
105	+	let deadline = started + timeout;
106	+	let t = \|s: &Instant\| s.elapsed().as_millis() as u64;
107	+	loop {
108	+	match rx.recv_timeout(Duration::from_millis(200)) {
109	+	Ok((from_console, line)) => {
110	+	let text = String::from_utf8_lossy(&line).trim_end_matches('\r').to_string();
111	+	let rec = \|stream: &str, text: &str\| Record { t: t(&started), stream: stream.into(), text: text.into() };
112	+	if !from_console {
113	+	sink(rec("console", &format!("firecracker: {text}")));
114	+	} else if let Some(rest) = text.strip_prefix('\u{1e}') {
115	+	let (kind, body) = rest.split_at(rest.len().min(1));
116	+	let body = body.strip_prefix(' ').unwrap_or(body);
117	+	match kind {
118	+	"O" => sink(rec("out", body)),
119	+	"E" => sink(rec("err", body)),
120	+	"X" => exit_code = body.trim().parse().ok(),
121	+	"F" => sink(rec("flow", &format!("guest init failed: {body}"))),
122	+	_ => sink(rec("console", &text)),
123	+	}
124	+	} else if !text.is_empty() {
125	+	sink(rec("console", &text));
126	+	}
127	+	}
128	+	Err(mpsc::RecvTimeoutError::Timeout) => {}
129	+	Err(mpsc::RecvTimeoutError::Disconnected) => break,
130	+	}
131	+	if Instant::now() >= deadline && !timed_out {
132	+	timed_out = true;
133	+	let _ = child.kill();
134	+	sink(Record { t: t(&started), stream: "flow".into(), text: format!("stopped at the time limit of {} s", timeout.as_secs()) });
135	+	}
136	+	}
137	+	let status = child.wait()?;
138	+	for r in readers {
139	+	let _ = r.join();
140	+	}
141	+	if exit_code.is_none() && !timed_out {
142	+	sink(Record { t: t(&started), stream: "flow".into(), text: format!("the microVM ended without the job's exit code ({status})") });
143	+	}
144	+	Ok(Outcome { exit_code: if timed_out { None } else { exit_code }, duration_ms: t(&started), environment, timed_out })
145	+	}
146	+
147	+	fn lines(r: impl Read, console: bool, tx: mpsc::Sender<(bool, Vec<u8>)>) {
148	+	let mut r = BufReader::new(r);
149	+	let mut buf = Vec::new();
150	+	loop {
151	+	buf.clear();
152	+	match r.read_until(b'\n', &mut buf) {
153	+	Ok(0) \| Err(_) => break,
154	+	Ok(_) => {
155	+	if buf.last() == Some(&b'\n') {
156	+	buf.pop();
157	+	}
158	+	if tx.send((console, buf.clone())).is_err() {
159	+	break;
160	+	}
161	+	}
162	+	}
163	+	}
164	+	}

	1	+	target/
	2	+	*.vsix

1	+	# This file is automatically @generated by Cargo.
2	+	# It is not intended for manual editing.
3	+	version = 4
4	+
5	+	[[package]]
6	+	name = "adler2"
7	+	version = "2.0.1"
8	+	source = "registry+https://github.com/rust-lang/crates.io-index"
9	+	checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
10	+
11	+	[[package]]
12	+	name = "aho-corasick"
13	+	version = "1.1.5"
14	+	source = "registry+https://github.com/rust-lang/crates.io-index"
15	+	checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba"
16	+	dependencies = [
17	+	"memchr",
18	+	]
19	+
20	+	[[package]]
21	+	name = "anstream"
22	+	version = "1.0.0"
23	+	source = "registry+https://github.com/rust-lang/crates.io-index"
24	+	checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d"
25	+	dependencies = [
26	+	"anstyle",
27	+	"anstyle-parse",
28	+	"anstyle-query",
29	+	"anstyle-wincon",
30	+	"colorchoice",
31	+	"is_terminal_polyfill",
32	+	"utf8parse",
33	+	]
34	+
35	+	[[package]]
36	+	name = "anstyle"
37	+	version = "1.0.14"
38	+	source = "registry+https://github.com/rust-lang/crates.io-index"
39	+	checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000"
40	+

1	+	//! Tier 2 conformance on a real microVM (spec §10.1 and §10.5). Runs when
2	+	//! `FLOW_TEST_HOME` names a Flow home with `[microvm]` pinned and
3	+	//! `FLOW_TEST_ROOTFS` the hash of an Alpine rootfs in it; skipped otherwise.
4	+
5	+	use std::process::Command;
6	+
7	+	#[test]
8	+	fn a_pipeline_runs_isolated_in_microvms() {
9	+	let (Some(home), Some(rootfs)) = (std::env::var_os("FLOW_TEST_HOME"), std::env::var("FLOW_TEST_ROOTFS").ok()) else {
10	+	eprintln!("skipped: set FLOW_TEST_HOME and FLOW_TEST_ROOTFS to boot microVMs");
11	+	return;
12	+	};
13	+	let dir = tempfile::tempdir().unwrap();
14	+	std::fs::create_dir_all(dir.path().join(".flow")).unwrap();
15	+	std::fs::write(
16	+	dir.path().join(".flow/pipeline.kdl"),
17	+	format!(
18	+	r#"schema-version 1
19	+	sandbox "alpine" {{
20	+	rootfs "cas://blob/{rootfs}"
21	+	memory-limit "256MiB"
22	+	}}
23	+	job "literal" {{
24	+	sandbox "alpine"
25	+	exec "echo" "a; touch /src/pwned && $(id) \| `id`"
26	+	}}
27	+	job "isolated" {{
28	+	sandbox "alpine"
29	+	exec "sh" "-c" "cat /proc/net/dev \| tail -n +3 \| cut -d: -f1 \| tr -d ' '; wget -q -T 2 -O /dev/null http://1.1.1.1 && echo NETWORK \|\| echo no-network"
30	+	}}
31	+	job "readonly" {{
32	+	sandbox "alpine"
33	+	exec "sh" "-c" "touch /etc/x 2>/dev/null && echo ROOTFS-WRITABLE; echo ok > /src/w && cat /src/w; id -u"
34	+	}}
35	+	job "fails" {{
36	+	sandbox "alpine"
37	+	exec "false"
38	+	}}
39	+	job "after-failure" {{
40	+	sandbox "alpine"

	1	+	max_width = 140
	2	+	use_small_heuristics = "Max"

1	+	//! `flow runner` (docs/m0.md §5): claim a job from OxygenHub, fetch the
2	+	//! repository into a cache with the job's token, write the state into a
3	+	//! workspace, run it in its tier, stream the log, report the receipt.
4	+	//! One job at a time.
5	+
6	+	use crate::home::{Home, RunnerConfig};
7	+	use crate::{microvm, workspace};
8	+	use anyhow::{bail, Context, Result};
9	+	use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, Record, MAX_RECORD_TEXT};
10	+	use flow_core::{scope_tree_hash, Receipt};
11	+	use oxy_core::hash::ChangeHash;
12	+	use oxy_core::repo::Repo;
13	+	use std::path::PathBuf;
14	+	use std::time::{Duration, Instant};
15	+
16	+	pub struct Hub {
17	+	base: String,
18	+	token: String,
19	+	agent: ureq::Agent,
20	+	}
21	+
22	+	impl Hub {
23	+	pub fn new(cfg: &RunnerConfig) -> Result<Hub> {
24	+	let tls = oxy_core::protocol::tls_client_config(cfg.ca.as_deref()).map_err(\|e\| anyhow::anyhow!("{e}"))?;
25	+	let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(60)).build();
26	+	Ok(Hub { base: cfg.hub.trim_end_matches('/').to_string(), token: cfg.token.clone(), agent })
27	+	}
28	+
29	+	fn post(&self, path: &str, body: &impl serde::Serialize) -> Result<Option<ureq::Response>> {
30	+	let url = format!("{}/api/v1{path}", self.base);
31	+	match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(body) {
32	+	Ok(r) if r.status() == 204 => Ok(None),
33	+	Ok(r) => Ok(Some(r)),
34	+	Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),
35	+	Err(e) => bail!("{url}: {e}"),
36	+	}
37	+	}
38	+
39	+	pub fn claim(&self, req: &ClaimRequest) -> Result<Option<Claim>> {
40	+	match self.post("/flow/claim", req)? {
41	+	Some(r) => Ok(Some(r.into_json()?)),
42	+	None => Ok(None),
43	+	}
44	+	}
45	+
46	+	pub fn log(&self, job: i64, records: Vec<Record>) -> Result<()> {
47	+	self.post(&format!("/flow/jobs/{job}/log"), &LogBatch { records })?;
48	+	Ok(())
49	+	}
50	+
51	+	pub fn finish(&self, job: i64, f: &Finish) -> Result<()> {
52	+	self.post(&format!("/flow/jobs/{job}/finish"), f)?;
53	+	Ok(())
54	+	}
55	+	}
56	+
57	+	pub fn serve(home: &Home) -> Result<()> {
58	+	let cfg = home.config.runner.clone().context("flow.toml has no [runner] section")?;
59	+	let hub = Hub::new(&cfg)?;
60	+	let name = cfg.name.clone().unwrap_or_else(\|\| "flow".into());
61	+	let req = ClaimRequest { runner: name.clone(), runtimes: vec!["microvm".into()] };
62	+	eprintln!("flow runner {name}: polling {}", cfg.hub);
63	+	let mut idle = Duration::from_secs(2);
64	+	loop {
65	+	match hub.claim(&req) {
66	+	Ok(Some(claim)) => {
67	+	idle = Duration::from_secs(2);
68	+	let id = claim.job_id;
69	+	eprintln!("job {id}: {} {} on {}", claim.repo, claim.job.name, claim.heads.first().map(\|h\| &h[..12]).unwrap_or("-"));
70	+	let finish = run_claim(home, &hub, &cfg, &claim);
71	+	let finish =
72	+	finish.unwrap_or_else(\|e\| Finish { exit_code: None, duration_ms: 0, receipt: None, error: Some(format!("{e:#}")) });
73	+	eprintln!(
74	+	"job {id}: exit {:?}{}",
75	+	finish.exit_code,
76	+	finish.error.as_deref().map(\|e\| format!(" ({e})")).unwrap_or_default()
77	+	);
78	+	if let Err(e) = hub.finish(id, &finish) {
79	+	eprintln!("job {id}: reporting failed: {e:#}");
80	+	}
81	+	}
82	+	Ok(None) => std::thread::sleep(idle),
83	+	Err(e) => {
84	+	eprintln!("claim: {e:#}");
85	+	std::thread::sleep(idle);
86	+	idle = (idle * 2).min(Duration::from_secs(60));
87	+	}
88	+	}
89	+	}
90	+	}
91	+
92	+	/// Batches records to the hub every second (or every 200 lines).
93	+	struct LogSink<'a> {
94	+	hub: &'a Hub,
95	+	job: i64,
96	+	buf: Vec<Record>,
97	+	last: Instant,
98	+	}
99	+
100	+	impl LogSink<'_> {
101	+	fn push(&mut self, mut r: Record) {
102	+	if r.text.len() > MAX_RECORD_TEXT {
103	+	let mut cut = MAX_RECORD_TEXT;
104	+	while !r.text.is_char_boundary(cut) {
105	+	cut -= 1;
106	+	}
107	+	r.text.truncate(cut);
108	+	}
109	+	self.buf.push(r);
110	+	if self.buf.len() >= 200 \|\| self.last.elapsed() >= Duration::from_secs(1) {
111	+	self.flush();
112	+	}
113	+	}
114	+
115	+	fn flush(&mut self) {
116	+	if !self.buf.is_empty() {
117	+	if let Err(e) = self.hub.log(self.job, std::mem::take(&mut self.buf)) {
118	+	eprintln!("job {}: log: {e:#}", self.job);
119	+	}
120	+	}
121	+	self.last = Instant::now();
122	+	}
123	+	}
124	+
125	+	fn run_claim(home: &Home, hub: &Hub, cfg: &RunnerConfig, claim: &Claim) -> Result<Finish> {
126	+	let started = Instant::now();
127	+	let mut sink = LogSink { hub, job: claim.job_id, buf: Vec::new(), last: Instant::now() };
128	+	let note =
129	+	\|sink: &mut LogSink, text: String\| sink.push(Record { t: started.elapsed().as_millis() as u64, stream: "flow".into(), text });
130	+	let heads: Vec<ChangeHash> = claim.heads.iter().map(\|h\| ChangeHash::from_hex(h).context("bad head")).collect::<Result<_>>()?;
131	+	note(&mut sink, format!("fetching {} for state {}", claim.repo, claim.heads.iter().map(\|h\| &h[..12]).collect::<Vec<_>>().join("+")));
132	+	let repo = cached_repo(home, cfg, claim)?;
133	+	let scratch = home.root.join("work").join(format!("job-{}", claim.job_id));
134	+	let _ = std::fs::remove_dir_all(&scratch);
135	+	std::fs::create_dir_all(scratch.join("tree"))?;
136	+	let result = (\|\| -> Result<Finish> {
137	+	let files = workspace::export_state(&repo, &heads, &scratch.join("tree"))?;
138	+	let scope_hash = scope_tree_hash(&claim.job.scope, files.iter().map(\|(p, h)\| (p.as_str(), *h)));
139	+	workspace::image(&scratch.join("tree"), &scratch.join("workspace.ext4"))?;
140	+	std::fs::remove_dir_all(scratch.join("tree"))?;
141	+	note(
142	+	&mut sink,
143	+	format!(
144	+	"booting a microVM: {} vCPU, {} MiB, rootfs {}",
145	+	claim.sandbox.cpus,
146	+	claim.sandbox.memory_mib,
147	+	&claim.sandbox.rootfs[..12]
148	+	),
149	+	);
150	+	let outcome = microvm::run(
151	+	home,
152	+	&claim.sandbox,
153	+	&claim.job,
154	+	&scratch.join("workspace.ext4"),
155	+	&scratch,
156	+	Duration::from_secs(claim.timeout_s),
157	+	&mut \|r\| sink.push(r),
158	+	)?;
159	+	let receipt = outcome.exit_code.map(\|code\| Receipt {
160	+	job_name: claim.job.name.clone(),
161	+	pipeline_hash: hex32(&claim.pipeline_hash),
162	+	scope_tree_hash: scope_hash,
163	+	inputs: vec![],
164	+	environment: outcome.environment.hash(),
165	+	outputs: vec![],
166	+	exit_code: code as u32,
167	+	duration_ms: outcome.duration_ms,
168	+	});
169	+	Ok(Finish {
170	+	exit_code: outcome.exit_code,
171	+	duration_ms: started.elapsed().as_millis() as u64,
172	+	receipt,
173	+	error: outcome.timed_out.then(\|\| "stopped at the time limit".to_string()),
174	+	})
175	+	})();
176	+	let _ = std::fs::remove_dir_all(&scratch);
177	+	sink.flush();
178	+	result
179	+	}
180	+
181	+	fn hex32(s: &str) -> [u8; 32] {
182	+	let mut out = [0; 32];
183	+	if let Ok(v) = hex::decode(s) {
184	+	if v.len() == 32 {
185	+	out.copy_from_slice(&v);
186	+	}
187	+	}
188	+	out
189	+	}
190	+
191	+	/// The runner's copy of a repository, fetched up to date with the job's token.
192	+	fn cached_repo(home: &Home, cfg: &RunnerConfig, claim: &Claim) -> Result<Repo> {
193	+	let dir: PathBuf = home.root.join("repos").join(claim.repo.replace('/', "__"));
194	+	let set_remote = \|repo: &mut Repo\| -> Result<()> {
195	+	repo.config.set("remote.origin.url", &claim.clone_url).map_err(\|e\| anyhow::anyhow!("{e}"))?;
196	+	repo.config.set("remote.origin.token", &claim.token).map_err(\|e\| anyhow::anyhow!("{e}"))?;
197	+	if let Some(ca) = &cfg.ca {
198	+	repo.config.set("remote.origin.ca", &ca.to_string_lossy()).map_err(\|e\| anyhow::anyhow!("{e}"))?;
199	+	}
200	+	repo.save_config().map_err(\|e\| anyhow::anyhow!("{e}"))
201	+	};
202	+	if dir.join(".oxy").is_dir() {
203	+	let mut repo = Repo::open(&dir).map_err(\|e\| anyhow::anyhow!("{e}"))?;
204	+	set_remote(&mut repo)?;
205	+	repo.fetch("origin").map_err(\|e\| anyhow::anyhow!("fetch {}: {e}", claim.clone_url))?;
206	+	return Ok(repo);
207	+	}
208	+	// First job for this repository: clone with the job's token.
209	+	std::env::set_var("OXY_TOKEN", &claim.token);
210	+	if let Some(ca) = &cfg.ca {
211	+	std::env::set_var("OXY_CA", ca);
212	+	}
213	+	let cloned = Repo::clone(&claim.clone_url, &dir);
214	+	std::env::remove_var("OXY_TOKEN");
215	+	std::env::remove_var("OXY_CA");
216	+	let mut repo = cloned.map_err(\|e\| {
217	+	let _ = std::fs::remove_dir_all(&dir);
218	+	anyhow::anyhow!("clone {}: {e}", claim.clone_url)
219	+	})?;
220	+	set_remote(&mut repo)?;
221	+	Ok(repo)
222	+	}

1	+	//! A job's workspace: the files of the state being built, written to a
2	+	//! scratch directory and packed into an ext4 image for the microVM
3	+	//! (docs/m0.md §2). Returns each file's content hash for the scope hash.
4	+
5	+	use anyhow::{bail, Context, Result};
6	+	use flow_core::Hash;
7	+	use oxy_core::hash::ChangeHash;
8	+	use oxy_core::objects::NodeKind;
9	+	use oxy_core::repo::Repo;
10	+	use oxy_core::state::FileContent;
11	+	use std::path::Path;
12	+
13	+	pub type Files = Vec<(String, Hash)>;
14	+
15	+	/// Write the state `heads` of `repo` under `dir`.
16	+	pub fn export_state(repo: &Repo, heads: &[ChangeHash], dir: &Path) -> Result<Files> {
17	+	let snapshot = repo.load_state(heads).context("load the state")?.materialize();
18	+	let mut files = Vec::new();
19	+	for (path, f) in &snapshot.files {
20	+	if path.split('/').any(\|p\| p.is_empty() \|\| p == "." \|\| p == ".." \|\| p == ".oxy") {
21	+	continue;
22	+	}
23	+	let target = dir.join(path);
24	+	match (&f.kind, &f.content) {
25	+	(NodeKind::Dir, _) \| (_, FileContent::Dir) => std::fs::create_dir_all(&target)?,
26	+	(NodeKind::Subrepo, _) => {}
27	+	(kind, content) => {
28	+	let bytes = match content {
29	+	FileContent::Text { bytes, .. } => bytes.clone(),
30	+	FileContent::Opaque { blobs } => match blobs.first() {
31	+	Some(b) => repo.read_blob(b)?,
32	+	None => Vec::new(),
33	+	},
34	+	FileContent::Dir => unreachable!(),
35	+	};
36	+	if let Some(parent) = target.parent() {
37	+	std::fs::create_dir_all(parent)?;
38	+	}
39	+	if *kind == NodeKind::Symlink {
40	+	let to = String::from_utf8_lossy(&bytes).into_owned();
41	+	std::os::unix::fs::symlink(&to, &target).with_context(\|\| format!("symlink {path}"))?;
42	+	} else {
43	+	std::fs::write(&target, &bytes).with_context(\|\| format!("write {path}"))?;
44	+	if f.mode.exec {
45	+	use std::os::unix::fs::PermissionsExt;
46	+	std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o755))?;
47	+	}
48	+	}
49	+	files.push((path.clone(), *blake3::hash(&bytes).as_bytes()));
50	+	}
51	+	}
52	+	}
53	+	Ok(files)
54	+	}
55	+
56	+	/// Copy a working directory (without `.oxy` and what `.oxyignore` and
57	+	/// `.gitignore` exclude) under `dir`, for `flow run`.
58	+	pub fn copy_tree(src: &Path, dir: &Path) -> Result<Files> {
59	+	let mut files = Vec::new();
60	+	let walk = ignore::WalkBuilder::new(src)
61	+	.hidden(false)
62	+	.git_ignore(true)
63	+	.require_git(false)
64	+	.add_custom_ignore_filename(".oxyignore")
65	+	.filter_entry(\|e\| e.file_name() != ".oxy" && e.file_name() != ".git")
66	+	.build();
67	+	for entry in walk {
68	+	let entry = entry?;
69	+	let rel = entry.path().strip_prefix(src)?;
70	+	if rel.as_os_str().is_empty() {
71	+	continue;
72	+	}
73	+	let rel_s = rel.to_string_lossy().replace('\\', "/");
74	+	let target = dir.join(rel);
75	+	let ft = entry.file_type().context("file type")?;
76	+	if ft.is_dir() {
77	+	std::fs::create_dir_all(&target)?;
78	+	} else if ft.is_symlink() {
79	+	let to = std::fs::read_link(entry.path())?;
80	+	std::os::unix::fs::symlink(&to, &target)?;
81	+	files.push((rel_s, *blake3::hash(to.to_string_lossy().as_bytes()).as_bytes()));
82	+	} else if ft.is_file() {
83	+	std::fs::copy(entry.path(), &target)?;
84	+	files.push((rel_s, *blake3::hash(&std::fs::read(entry.path())?).as_bytes()));
85	+	}
86	+	}
87	+	Ok(files)
88	+	}
89	+
90	+	/// Pack `dir` into an ext4 image whose root belongs to this user, so the
91	+	/// job (which runs as the owner of `/src`) can write to it.
92	+	pub fn image(dir: &Path, out: &Path) -> Result<()> {
93	+	let used: u64 = ignore::WalkBuilder::new(dir)
94	+	.hidden(false)
95	+	.ignore(false)
96	+	.git_ignore(false)
97	+	.build()
98	+	.flatten()
99	+	.filter_map(\|e\| e.metadata().ok())
100	+	.map(\|m\| m.len().max(4096))
101	+	.sum();
102	+	let mib = used * 12 / 10 / (1 << 20) + 512;
103	+	let (uid, gid) = unsafe { (libc::getuid(), libc::getgid()) };
104	+	let status = std::process::Command::new("mkfs.ext4")
105	+	.args(["-q", "-F", "-L", "workspace", "-E"])
106	+	.arg(format!("root_owner={uid}:{gid}"))
107	+	.arg("-d")
108	+	.arg(dir)
109	+	.arg(out)
110	+	.arg(format!("{mib}M"))
111	+	.status()
112	+	.context("run mkfs.ext4 (e2fsprogs 1.43 or later)")?;
113	+	if !status.success() {
114	+	bail!("mkfs.ext4 failed for the workspace image");
115	+	}
116	+	Ok(())
117	+	}

1	+	# OxygenFlow M0: first pipeline on Earth
2	+
3	+	\| \| \|
4	+	\|---\|---\|
5	+	\| Status \| Plan, 2026-09-28 \|
6	+	\| Implements \| a subset of `docs/spec.md` (Draft 0.1) \|
7	+	\| Goal \| A push to a repository on OxygenHub (Earth) runs `.flow/pipeline.kdl` in Firecracker microVMs, and the hub shows the run as a graph of jobs with live status and logs. \|
8	+
9	+	M0 is the smallest slice that exercises the real architecture: a typed
10	+	pipeline compiled to a canonical form, host-side execution in the default
11	+	isolation tier, receipts computed by the host, and results reported to the
12	+	hub as checks. Everything M0 leaves out is listed at the end.
13	+
14	+	## 1. Pipeline subset
15	+
16	+	The file is `.flow/pipeline.kdl` in the state being built (KDL 2.0). M0
17	+	accepts:
18	+
19	+	```kdl
20	+	schema-version 1
21	+
22	+	sandbox "alpine" {
23	+	runtime "microvm" // "microvm" (default) \| "syscall"; "wasm" is refused in M0
24	+	rootfs "cas://blob/<64 hex>" // BLAKE3 of an ext4 image the runner holds
25	+	memory-limit "1GiB" // KiB, MiB, GiB
26	+	cpu-limit 2
27	+	network "isolated" // M0 accepts only "isolated"
28	+	}
29	+
30	+	job "lint" {
31	+	sandbox "alpine"
32	+	scope "src/**" "README.md" // recorded; the job runs on every push in M0
33	+	env { LANG "C.UTF-8" }
34	+	exec "grep" "-rn" "TODO" "src"
35	+	}
36	+
37	+	job "test" {
38	+	sandbox "alpine"
39	+	depends-on "lint"
40	+	exec "sh" "-eu" "test.sh"
41	+	}
42	+	```
43	+
44	+	Rules, all checked by `flow check` and again by the hub before anything
45	+	runs (spec §2.2 I2, §3.2):
46	+
47	+	- Unknown nodes, arguments or properties are errors, with the line.
48	+	- Every string stays a string; `cpu-limit` is an integer from 1 to 16;
49	+	`memory-limit` parses to 64 MiB to 16 GiB.
50	+	- A job names a declared sandbox, has a non-empty `exec` of literal
51	+	strings, and `depends-on` only declared jobs; the graph has no cycle.
52	+	- Names are 1 to 64 characters from `[A-Za-z0-9._-]`.
53	+	- `exec` becomes `argv` unchanged: metacharacters are bytes (spec §10.1).
54	+	A job that wants a shell names one, as `sh` above; nothing adds one.
55	+	- `mount`, `output`, `publish` and `secret` are recognised and refused with
56	+	"not in M0", so a pipeline written for the full spec fails clearly.
57	+
58	+	The pipeline compiles to a canonical CBOR job graph (sorted keys, sandboxes
59	+	and jobs in name order). Its BLAKE3 is the `pipeline_hash` used in receipts
60	+	and shown by the hub.
61	+
62	+	## 2. Tier 2: Firecracker (default)
63	+
64	+	Per job the runner:
65	+
66	+	1. Verifies the rootfs image: `BLAKE3(file)` must equal the pinned hash.
67	+	Images live in `$FLOW_HOME/cas/<hash>.ext4`; an unknown hash fails the
68	+	job ("rootfs not on this runner").
69	+	2. Materializes the state at its heads with `oxy` into a scratch
70	+	directory (a clone at the state), removes `.oxy`, and builds a
71	+	workspace image with `mkfs.ext4 -d` sized to the tree plus 512 MiB.
72	+	3. Boots Firecracker with no network device (so `isolated` holds by
73	+	construction), `vcpu_count` and `mem_size_mib` from the sandbox, the
74	+	pinned guest kernel, the rootfs as a read-only `vda`, the
75	+	workspace as a read-write `vdb`, and an initramfs holding `flow-init`.
76	+	4. `flow-init` (a static binary, PID 1 in the guest) mounts `/proc`,
77	+	`/sys`, `/dev`, the rootfs at `/newroot` read-only, `vdb` at
78	+	`/newroot/src`, and tmpfs at `/newroot/tmp` and `/newroot/run`; it
79	+	switches root, then runs `argv` with exactly the job's `env` (plus
80	+	`PATH`, `HOME=/tmp`) in `/src` as uid 1000, with a job timeout.
81	+	5. `flow-init` frames the child's output on the serial console as
82	+	`\x1eO <line>` (stdout) and `\x1eE <line>` (stderr), then
83	+	`\x1eX <code>`, and powers off. The kernel runs `quiet` so the host sees
84	+	only frames; anything else is kept as `console` records.
85	+	6. The host reads frames, streams them as log records, and kills the VM
86	+	at the timeout (the guest cannot extend it). Scratch files are deleted
87	+	afterwards. Guest memory is released with the process.
88	+
89	+	The guest kernel, `flow-init` and Firecracker itself are pinned by BLAKE3
90	+	in the runner's configuration; their hashes and the rootfs hash make the
91	+	receipt's `environment` together with `argv` and `env`.
92	+
93	+	## 3. Tier 1: syscall sandbox
94	+
95	+	`runtime "syscall"`: a child in new user, mount, PID, network, IPC and
96	+	UTS namespaces (no interfaces but loopback), the rootfs image's tree
97	+	(unpacked once per hash, read-only) as `/`, the workspace directory at
98	+	`/src`, tmpfs at `/tmp`; Landlock limits writes to `/src` and `/tmp`; a
99	+	seccomp filter refuses `socket` for anything but `AF_UNIX`, plus
100	+	`ptrace`, `mount`, `bpf`, `perf_event_open`, `keyctl` and module calls;
101	+	`cpu-limit` and `memory-limit` apply through a cgroup v2 scope when the
102	+	runner has a delegated cgroup, and the job fails to start otherwise
103	+	rather than run unbounded. Same frames and receipts as tier 2. Done after
104	+	tier 2 works on Earth.
105	+
106	+	## 4. Receipts
107	+
108	+	Every finished job gets a receipt (spec §6.1): `job_name`,
109	+	`pipeline_hash`, `scope_tree_hash` (BLAKE3 over the sorted
110	+	`(path, content hash)` of the files `scope` matches; the whole tree when
111	+	there is no scope), `inputs` (empty in M0), `environment`, `outputs`
112	+	(empty), `exit_code`, `duration_ms`. It is encoded as canonical CBOR and
113	+	hashed as `BLAKE3("oxy1 receipt " \|\| len \|\| 0x00 \|\| body)`. M0 stores it on
114	+	the hub with the job; storing receipts as Oxygen objects needs the
115	+	`receipt` object type in the Oxygen spec, and cache reuse (§6.2) follows.
116	+
117	+	## 5. Hub side (OxygenHub)
118	+
119	+	- Configuration: `[flow] runner_tokens = [...]` and `repos = ["owner/*"]`,
120	+	the repositories whose pushes may start runs. No proposals trigger runs
121	+	in M0, so code from people without write access never runs.
122	+	- On a push that moves a bookmark of an allowed repository, the hub reads
123	+	`.flow/pipeline.kdl` from the new state, validates it with `flow-core`,
124	+	and records a run (repository, bookmark, state address, pipeline
125	+	hash, pusher) with one job per pipeline job. An invalid pipeline
126	+	records a failed run with the error and its line.
127	+	- Runner API, bearer runner token:
128	+	- `POST /api/v1/flow/claim` → the oldest queued job whose dependencies
129	+	passed, with its run's state heads, repository clone URL, the job's
130	+	compiled form and its sandbox, and a job token: read access to that
131	+	repository only, valid until the job finishes or 2 hours pass. `204`
132	+	when there is nothing.
133	+	- `POST /api/v1/flow/jobs/{id}/log` appends records `{t, stream, text}`.
134	+	- `POST /api/v1/flow/jobs/{id}/finish` with the exit code, duration and
135	+	receipt; a job whose dependency failed becomes `skipped`.
136	+	- Each job is also a check named `flow/<job>` on the state (hub spec §13),
137	+	so proposals, the status bar in VS Code and required checks see it.
138	+	- Pages: a Flow tab with runs (status, bookmark, state, pusher,
139	+	duration); a run page with the job graph drawn as SVG (columns by
140	+	dependency depth, status by colour and glyph), each job's log folded
141	+	by default and the failed one open, refreshed while the run is live.
142	+
143	+	## 6. Earth
144	+
145	+	Under `/home/claude/flow`: `bin/firecracker` (release binary, pinned),
146	+	`kernel/vmlinux` (pinned), `cas/` (rootfs images), `work/` (scratch), and
147	+	`runner.toml`. `flow runner` runs as the `flow-runner` systemd service
148	+	under the `claude` user (member of `kvm`), polling the hub at
149	+	`https://earth.andymccall.xyz:7343`.
150	+
151	+	## 7. Not in M0
152	+
153	+	WASM tier; FastCDC-backed rootfs and `casfs` mounts (M0 copies the tree
154	+	into an image); `mount`, `output` and artifacts; receipt reuse; provenance
155	+	attestations and composite signatures; policy gates; secrets; proposal
156	+	runs; scope-based skipping; re-run and cancel from the hub.

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

OxygenFlow M0: pipeline parser and canonical job graph, receipts, the runner API types, flow-init, the Firecracker tier, flow check, run, rootfs import and runner e0918b887850

Keyboard