Skip to main content

max / makenotwork

ops-exec: trusted executor (E1+E2), Sando refactor, thin macOS driver Lane A release pipeline. New MIT crate shared/ops-exec: the capability-gated Executor trait with LocalExec/SshExec/AgentRpc transports, typed Step/Action, double-enforced CapabilitySet, and the ops-agent binary (tailscale whois identity, Aqua LaunchAgent for in-session macOS signing). ops-core re-exports it so bento compiles unchanged. Sando deploy refactored onto the executor, behavior-preserving: ssh/rsync now run through the per-node executor map built from topology; Node gains actuate/observe caps defaulting to deploy+restart/health so sando.toml is unchanged. LiveLog implements ops_exec::LogSink. bento/driver (bento-release-macos): the thin agent-driven driver (launchplan A decision (a)) — checkout -> release-macos.sh --keychain -> spctl verify -> pull DMG, transport-agnostic via ops-exec. Tests: ops-exec 25 unit + 4 HTTP e2e, ops-core 8, sando 76, driver 6; zero warnings, clippy clean. Real DMG run / clean-Mac smoke / mbp install remain handoffs H1/M7. Runbook: _private/docs/ops-core/operator-runbook.md.
Co-Authored-By
Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-07 19:03 UTC
Signed with PGP, not checked
Commit: 14fd620d9683016a5a47c8f3fbc8a03b57381e0d
Parent: 9cceb18
35 files changed, +6394 insertions, -144 deletions
@@ -46,6 +46,17 @@
46 46 source = "registry+https://github.com/rust-lang/crates.io-index"
47 47 checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
48 48
49 + [[package]]
50 + name = "async-trait"
51 + version = "0.1.89"
52 + source = "registry+https://github.com/rust-lang/crates.io-index"
53 + checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
54 + dependencies = [
55 + "proc-macro2",
56 + "quote",
57 + "syn",
58 + ]
59 +
49 60 [[package]]
50 61 name = "atoi"
51 62 version = "2.0.0"
@@ -1262,13 +1273,27 @@
1262 1273 version = "0.1.0"
1263 1274 dependencies = [
1264 1275 "anyhow",
1276 + "async-trait",
1265 1277 "chrono",
1278 + "ops-exec",
1266 1279 "serde",
1267 1280 "sqlx",
1268 1281 "tokio",
1269 1282 "tracing",
1270 1283 ]
1271 1284
1285 + [[package]]
1286 + name = "ops-exec"
1287 + version = "0.1.0"
1288 + dependencies = [
1289 + "anyhow",
1290 + "async-trait",
1291 + "serde",
1292 + "thiserror",
1293 + "tokio",
1294 + "tracing",
1295 + ]
1296 +
1272 1297 [[package]]
1273 1298 name = "parking"
1274 1299 version = "2.2.1"
@@ -32,6 +32,17 @@
32 32 source = "registry+https://github.com/rust-lang/crates.io-index"
33 33 checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
34 34
35 + [[package]]
36 + name = "async-trait"
37 + version = "0.1.89"
38 + source = "registry+https://github.com/rust-lang/crates.io-index"
39 + checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
40 + dependencies = [
41 + "proc-macro2",
42 + "quote",
43 + "syn",
44 + ]
45 +
35 46 [[package]]
36 47 name = "atoi"
37 48 version = "2.0.0"
@@ -1152,6 +1163,18 @@
1152 1163 source = "registry+https://github.com/rust-lang/crates.io-index"
1153 1164 checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
1154 1165
1166 + [[package]]
1167 + name = "ops-exec"
1168 + version = "0.1.0"
1169 + dependencies = [
1170 + "anyhow",
1171 + "async-trait",
1172 + "serde",
1173 + "thiserror",
1174 + "tokio",
1175 + "tracing",
1176 + ]
1177 +
1155 1178 [[package]]
1156 1179 name = "parking"
1157 1180 version = "2.2.1"
@@ -1617,11 +1640,13 @@
1617 1640 version = "0.2.0"
1618 1641 dependencies = [
1619 1642 "anyhow",
1643 + "async-trait",
1620 1644 "axum",
1621 1645 "chrono",
1622 1646 "http-body-util",
1623 1647 "metrics",
1624 1648 "metrics-exporter-prometheus",
1649 + "ops-exec",
1625 1650 "reqwest",
1626 1651 "semver",
1627 1652 "serde",
@@ -11,6 +11,8 @@
11 11 [dependencies]
12 12 axum = { version = "0.8.8", features = ["macros", "ws"] }
13 13 tokio = { version = "1.50.0", features = ["macros", "rt-multi-thread", "net", "signal", "fs", "process"] }
14 + ops-exec = { path = "../../shared/ops-exec" }
15 + async-trait = "0.1.83"
14 16 serde = { version = "1.0.228", features = ["derive"] }
15 17 serde_json = "1"
16 18 toml = "0.8"
@@ -23,6 +23,17 @@
23 23 source = "registry+https://github.com/rust-lang/crates.io-index"
24 24 checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
25 25
26 + [[package]]
27 + name = "async-trait"
28 + version = "0.1.89"
29 + source = "registry+https://github.com/rust-lang/crates.io-index"
30 + checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
31 + dependencies = [
32 + "proc-macro2",
33 + "quote",
34 + "syn",
35 + ]
36 +
26 37 [[package]]
27 38 name = "atoi"
28 39 version = "2.0.0"
@@ -805,7 +816,9 @@
805 816 version = "0.1.0"
806 817 dependencies = [
807 818 "anyhow",
819 + "async-trait",
808 820 "chrono",
821 + "ops-exec",
809 822 "serde",
810 823 "serde_json",
811 824 "sqlx",
@@ -814,6 +827,18 @@
814 827 "tracing",
815 828 ]
816 829
830 + [[package]]
831 + name = "ops-exec"
832 + version = "0.1.0"
833 + dependencies = [
834 + "anyhow",
835 + "async-trait",
836 + "serde",
837 + "thiserror",
838 + "tokio",
839 + "tracing",
840 + ]
841 +
817 842 [[package]]
818 843 name = "parking"
819 844 version = "2.2.1"
@@ -6,10 +6,12 @@
6 6 description = "Shared operator-tool infrastructure: streaming remote-exec, a generic event bus, a disk+broadcast live-log sink, and sqlite helpers. Used by sando-daemon and bento-daemon."
7 7
8 8 [dependencies]
9 + ops-exec = { path = "../ops-exec" }
9 10 tokio = { version = "1.50.0", features = ["macros", "rt-multi-thread", "io-util", "fs", "process", "sync"] }
10 11 serde = { version = "1.0.228", features = ["derive"] }
11 12 chrono = { version = "0.4", features = ["serde"] }
12 13 anyhow = "1.0.102"
14 + async-trait = "0.1.83"
13 15 tracing = "0.1.44"
14 16 sqlx = { version = "0.8.6", features = ["runtime-tokio", "sqlite"] }
15 17
@@ -14,28 +14,56 @@
14 14 //! `<release_root>/current/<bin_name>` so reload-or-restart picks up the new
15 15 //! binary without ever pointing at a missing path.
16 16 //!
17 - //! For nodes with `ssh_target` set to anything other than `"local"`, deploy
18 - //! goes via rsync + ssh; the bootstrap (creating release_root, installing the
19 - //! service unit, granting sudo for systemctl) is out of scope here — it
20 - //! happens once per node, not per deploy.
17 + //! The host-side transport — `ssh` for shell steps, `rsync` for the release
18 + //! dir — comes from the shared [`ops_exec::Executor`] (a `LocalExec` for
19 + //! `ssh_target = "local"`, an `SshExec` otherwise), built once per node in
20 + //! [`crate::state`]. This module owns the *deploy choreography* (mkdir, push,
21 + //! atomic swap, restart, gc); the transport is the crate's. SSH push behavior
22 + //! is identical to the pre-extraction code — this is a transport extraction,
23 + //! not a model change.
21 24
22 25 use crate::topology::Node;
23 26 use anyhow::{Context, Result};
27 + use async_trait::async_trait;
28 + use ops_exec::{Action, Executor, LogSink, RunOutput, Step, SyncOpts, sh_quote};
24 29 use std::path::{Path, PathBuf};
25 30 use tokio::process::Command;
26 31
27 - /// SSH options used everywhere we shell out to ssh — fail fast, no prompts.
28 - const SSH_FLAGS: &[&str] = &[
29 - "-o", "BatchMode=yes",
30 - "-o", "ConnectTimeout=10",
31 - "-o", "StrictHostKeyChecking=accept-new",
32 - ];
33 -
34 32 /// Keep this many release dirs per node; older ones get gc'd after a
35 33 /// successful deploy. Fixed for now; promote to config if the constant ever
36 34 /// needs to vary by tier.
37 35 const RELEASES_TO_KEEP: usize = 5;
38 36
37 + /// A sink that drops streamed bytes. Deploy steps don't have a live-log handle
38 + /// (gates do), so output is discarded as it streams; [`RunOutput`] still
39 + /// captures the full stdout/stderr for error reporting, preserving the
40 + /// pre-extraction behavior of surfacing `stderr` in failure messages.
41 + struct DiscardSink;
42 +
43 + #[async_trait]
44 + impl LogSink for DiscardSink {
45 + async fn write_chunk(&mut self, _bytes: &[u8]) {}
46 + }
47 +
48 + /// Run a shell step through `executor`, treating a non-zero exit as an error
49 + /// whose message carries the captured stderr — exactly as the old bespoke
50 + /// `ssh()` helper did (`ssh <target> failed: <stderr>`).
51 + async fn run_checked(executor: &dyn Executor, script: &str, what: &str) -> Result<RunOutput> {
52 + let step = Step::shell(Action::Deploy, script);
53 + let mut sink = DiscardSink;
54 + let out = executor
55 + .run_streaming(&step, &mut sink)
56 + .await
57 + .with_context(|| format!("{what}: spawning command"))?;
58 + anyhow::ensure!(
59 + out.status.success(),
60 + "{what} failed (exit {}): {}",
61 + out.status.code().map(|c| c.to_string()).unwrap_or_else(|| "signal".into()),
62 + String::from_utf8_lossy(&out.stderr),
63 + );
64 + Ok(out)
65 + }
66 +
39 67 pub async fn deploy_local(
40 68 release_root: &Path,
41 69 version: &crate::domain::Version,
@@ -72,12 +100,14 @@
72 100 }
73 101
74 102 /// Deploy `staged_release_dir` (a directory built on the Sando host by
75 - /// `deploy_local`) to `node`. For `ssh_target=local`, this is just symlink
76 - /// swap + restart; for remote nodes, we rsync the whole dir.
103 + /// `deploy_local`) to `node` using `executor` (its transport from the topology
104 + /// executor map). For `ssh_target=local`, this is just a symlink swap; for
105 + /// remote nodes, we rsync the whole dir over the executor.
77 106 ///
78 107 /// `primary_bin` is only used for logging — every file present in the staged
79 108 /// dir gets shipped.
80 109 pub async fn deploy_node(
110 + executor: &dyn Executor,
81 111 node: &Node,
82 112 version: &str,
83 113 staged_release_dir: &Path,
@@ -86,78 +116,63 @@
86 116 if node.ssh_target == "local" || node.ssh_target.is_empty() {
87 117 // Local deploy already happened when we staged on the Sando host.
88 118 // Just re-point `current` at the staged dir.
89 - return reset_local_current(Path::new(&node.release_root), version).await;
119 + return reset_local_current(executor, Path::new(&node.release_root), version).await;
90 120 }
91 - deploy_remote(node, version, staged_release_dir, primary_bin).await
121 + deploy_remote(executor, node, version, staged_release_dir, primary_bin).await
92 122 }
93 123
94 - async fn reset_local_current(release_root: &Path, version: &str) -> Result<PathBuf> {
124 + async fn reset_local_current(
125 + executor: &dyn Executor,
126 + release_root: &Path,
127 + version: &str,
128 + ) -> Result<PathBuf> {
95 129 let current = release_root.join("current");
96 130 let target = format!("releases/{version}");
97 - let out = Command::new("ln")
98 - .args(["-sfn", &target])
99 - .arg(&current)
100 - .output()
101 - .await?;
102 - anyhow::ensure!(
103 - out.status.success(),
104 - "symlink swap failed: {}",
105 - String::from_utf8_lossy(&out.stderr),
106 - );
131 + run_checked(
132 + executor,
133 + &format!("ln -sfn {} {}", sh_quote(&target), sh_quote(&current.to_string_lossy())),
134 + "local symlink swap",
135 + )
136 + .await?;
107 137 Ok(release_root.join("releases").join(version))
108 138 }
109 139
110 140 async fn deploy_remote(
141 + executor: &dyn Executor,
111 142 node: &Node,
112 143 version: &str,
113 144 staged_release_dir: &Path,
114 145 primary_bin: &str,
115 146 ) -> Result<PathBuf> {
116 147 let release_root = &node.release_root;
117 - let ssh_target = &node.ssh_target;
118 148 let service = &node.service_name;
119 149 let release_dir = format!("{release_root}/releases/{version}");
120 150
121 151 tracing::info!(node = %node.name, version, "deploy: mkdir release dir");
122 - ssh(ssh_target, &format!("set -e; mkdir -p {q}", q = sh_quote(&release_dir)))
123 - .await
124 - .context("creating remote release dir")?;
152 + run_checked(
153 + executor,
154 + &format!("set -e; mkdir -p {q}", q = sh_quote(&release_dir)),
155 + "creating remote release dir",
156 + )
157 + .await?;
125 158
126 159 tracing::info!(node = %node.name, version, primary = %primary_bin, "deploy: rsync release dir");
127 160 // Rsync the whole staged dir (binaries + every release_contents entry).
128 - // Trailing slash on source = contents of dir, not the dir itself.
129 - //
130 - // --delete: removed assets across versions don't accumulate on the
131 - // target. Bundle stays self-contained per version.
132 - // --chmod: `F+X` preserves execute bit per-file (binaries land 0755,
133 - // data files 0644) instead of the old blanket-0755 that was wrong for
134 - // static assets + docs.
135 - let rsync_src = format!("{}/", staged_release_dir.display());
136 - let rsync_dest = format!("{ssh_target}:{release_dir}/");
137 - let mut rsync = Command::new("rsync");
138 - rsync
139 - .arg("-az")
140 - .arg("--partial")
141 - .arg("--delete")
142 - .arg("--chmod=Du=rwx,Dgo=rx,Fu=rw,Fgo=r,F+X")
143 - .arg("-e")
144 - .arg(format!(
145 - "ssh {}",
146 - SSH_FLAGS.iter().map(|s| s.to_string()).collect::<Vec<_>>().join(" ")
147 - ))
148 - .arg(&rsync_src)
149 - .arg(&rsync_dest);
150 - let out = rsync.output().await.context("spawning rsync")?;
151 - anyhow::ensure!(
152 - out.status.success(),
153 - "rsync failed (current symlink left intact): {}",
154 - String::from_utf8_lossy(&out.stderr),
155 - );
161 + // `SyncOpts::release_mirror()` is the exact pre-extraction rsync flag set:
162 + // -az --partial --delete --chmod=Du=rwx,Dgo=rx,Fu=rw,Fgo=r,F+X.
163 + // --delete: removed assets across versions don't accumulate on the
164 + // target; the bundle stays self-contained per version.
165 + // --chmod: F+X preserves the execute bit per-file (binaries land 0755,
166 + // data files 0644) instead of a blanket 0755.
167 + executor
168 + .push(staged_release_dir, Path::new(&release_dir), &SyncOpts::release_mirror())
169 + .await
170 + .context("rsync failed (current symlink left intact)")?;
156 171
157 172 tracing::info!(node = %node.name, version, "deploy: symlink swap + service reload");
158 - // Symlink swap is atomic via `mv -T` of a freshly-created symlink over
159 - // the old one (the rename(2) is the atomic step; `ln -sfn` does
160 - // unlink+symlink which has a window).
173 + // Symlink swap is atomic via `mv -T` of a freshly-created symlink over the
174 + // old one (the rename(2) is the atomic step; `ln -sfn` does unlink+symlink
175 + // which has a window).
161 176 let swap_and_restart = format!(
162 177 "set -e; \
163 178 cd {root}; \
@@ -168,29 +183,15 @@
168 183 ver = sh_quote(version),
169 184 svc = sh_quote(service),
170 185 );
171 - ssh(ssh_target, &swap_and_restart)
172 - .await
173 - .context("symlink swap + systemctl reload-or-restart")?;
186 + run_checked(executor, &swap_and_restart, "symlink swap + systemctl reload-or-restart").await?;
174 187
175 - if let Err(e) = gc_remote_releases(ssh_target, release_root).await {
188 + if let Err(e) = gc_remote_releases(executor, release_root).await {
176 189 tracing::warn!(error = %e, "remote release GC failed (non-fatal)");
177 190 }
178 191
179 192 Ok(PathBuf::from(release_root).join("releases").join(version))
180 193 }
181 194
182 - async fn ssh(target: &str, script: &str) -> Result<()> {
183 - let mut cmd = Command::new("ssh");
184 - cmd.args(SSH_FLAGS).arg(target).arg(script);
185 - let out = cmd.output().await.context("spawning ssh")?;
186 - anyhow::ensure!(
187 - out.status.success(),
188 - "ssh {target} failed: {}",
189 - String::from_utf8_lossy(&out.stderr),
190 - );
191 - Ok(())
192 - }
193 -
194 195 async fn gc_local_releases(release_root: &Path) -> Result<()> {
195 196 let releases = release_root.join("releases");
196 197 if !releases.exists() {
@@ -216,7 +217,7 @@
216 217 Ok(())
217 218 }
218 219
219 - async fn gc_remote_releases(ssh_target: &str, release_root: &str) -> Result<()> {
220 + async fn gc_remote_releases(executor: &dyn Executor, release_root: &str) -> Result<()> {
220 221 // `ls -t` orders by mtime desc. Skip the first N, rm the rest. `xargs -r`
221 222 // is a no-op when stdin is empty (avoids `rm` complaining).
222 223 let script = format!(
@@ -225,32 +226,18 @@
225 226 root = sh_quote(release_root),
226 227 keep_plus_one = RELEASES_TO_KEEP + 1,
227 228 );
228 - ssh(ssh_target, &script).await
229 - }
230 -
231 - /// Single-quote a string for safe inclusion in a /bin/sh command, escaping
232 - /// any single quote inside. Not bulletproof for adversarial input, but every
233 - /// path here comes from our own config files.
234 - fn sh_quote(s: &str) -> String {
235 - let escaped = s.replace('\'', r"'\''");
236 - format!("'{escaped}'")
229 + run_checked(executor, &script, "remote release gc").await.map(|_| ())
237 230 }
238 231
239 232 #[cfg(test)]
240 233 mod tests {
241 234 use super::*;
235 + use ops_exec::{CapabilitySet, LocalExec, SshExec};
242 236 use std::time::SystemTime;
243 237
244 - #[test]
245 - fn sh_quote_no_quote() {
246 - assert_eq!(sh_quote("hello"), "'hello'");
247 - assert_eq!(sh_quote("/opt/mnw/releases/0.8.12"), "'/opt/mnw/releases/0.8.12'");
248 - }
249 -
250 - #[test]
251 - fn sh_quote_with_quote() {
252 - // The string `it's` becomes `'it'\''s'` — close, escape, open.
253 - assert_eq!(sh_quote("it's"), r"'it'\''s'");
238 + /// A LocalExec granted the default node capabilities (deploy + restart).
239 + fn local_executor() -> LocalExec {
240 + LocalExec::new(CapabilitySet::from_tokens(["deploy", "restart"], ["health"]))
254 241 }
255 242
256 243 #[tokio::test]
@@ -258,7 +245,6 @@
258 245 let tmp = tempfile::tempdir().unwrap();
259 246 let root = tmp.path();
260 247
261 - // Source binaries (worktree's target/release/)
262 248 let src_dir = root.join("src");
263 249 tokio::fs::create_dir_all(&src_dir).await.unwrap();
264 250 let primary = src_dir.join("makenotwork");
@@ -266,7 +252,6 @@
266 252 tokio::fs::write(&primary, b"PRIMARY").await.unwrap();
267 253 tokio::fs::write(&admin, b"ADMIN").await.unwrap();
268 254
269 - // Release root (where staged versions live)
270 255 let release_root = root.join("releases-root");
271 256 tokio::fs::create_dir_all(&release_root).await.unwrap();
272 257
@@ -282,11 +267,9 @@
282 267 assert_eq!(tokio::fs::read(staged.join("makenotwork")).await.unwrap(), b"PRIMARY");
283 268 assert_eq!(tokio::fs::read(staged.join("mnw-admin")).await.unwrap(), b"ADMIN");
284 269
285 - // current symlink should resolve to staged
286 270 let current = release_root.join("current");
287 271 let target = tokio::fs::read_link(&current).await.unwrap();
288 272 assert_eq!(target.to_string_lossy(), "releases/0.8.12");
289 - // And reading through `current/` should give the new content.
290 273 let via_current = tokio::fs::read(current.join("makenotwork")).await.unwrap();
291 274 assert_eq!(via_current, b"PRIMARY");
292 275 }
@@ -304,14 +287,11 @@
304 287 tokio::fs::create_dir_all(&release_root).await.unwrap();
305 288
306 289 deploy_local(&release_root, &"0.1.0".parse().unwrap(), &[bin.clone()]).await.unwrap();
307 - // Rewrite source then deploy 0.2.0.
308 290 tokio::fs::write(&bin, b"V2").await.unwrap();
309 291 deploy_local(&release_root, &"0.2.0".parse().unwrap(), &[bin.clone()]).await.unwrap();
310 292
311 - // Both versions present on disk.
312 293 assert!(release_root.join("releases/0.1.0/server").exists());
313 294 assert!(release_root.join("releases/0.2.0/server").exists());
314 - // current points at the new one.
315 295 let target = tokio::fs::read_link(release_root.join("current")).await.unwrap();
316 296 assert_eq!(target.to_string_lossy(), "releases/0.2.0");
317 297 let via_current = tokio::fs::read(release_root.join("current/server")).await.unwrap();
@@ -320,8 +300,6 @@
320 300
321 301 #[tokio::test]
322 302 async fn gc_local_releases_keeps_last_n_by_mtime() {
323 - // Build > RELEASES_TO_KEEP fake release dirs with distinct mtimes,
324 - // then run gc and check which survived.
325 303 let tmp = tempfile::tempdir().unwrap();
326 304 let root = tmp.path();
327 305 let releases = root.join("releases");
@@ -333,10 +311,6 @@
333 311 let name = format!("v{i:02}");
334 312 let dir = releases.join(&name);
335 313 tokio::fs::create_dir(&dir).await.unwrap();
336 - // Stagger mtimes deterministically. tokio's File doesn't expose
337 - // set_times, so reach for std::fs::File + std::fs::FileTimes
338 - // (stable since 1.75). Synchronous is fine here — this is test
339 - // setup, not the hot path.
340 314 let f = std::fs::File::open(&dir).unwrap();
341 315 let when = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000 + i as u64);
342 316 let times = std::fs::FileTimes::new().set_modified(when);
@@ -346,23 +320,16 @@
346 320
347 321 gc_local_releases(root).await.unwrap();
348 322
349 - // The last RELEASES_TO_KEEP by mtime (i.e. highest i) survive.
350 323 let surviving_expected: Vec<_> = names
351 324 .iter()
352 325 .skip(total - RELEASES_TO_KEEP)
353 326 .cloned()
354 327 .collect();
355 328 for name in &surviving_expected {
356 - assert!(
357 - releases.join(name).exists(),
358 - "expected to survive: {name}"
359 - );
329 + assert!(releases.join(name).exists(), "expected to survive: {name}");
360 330 }
361 331 for name in names.iter().take(total - RELEASES_TO_KEEP) {
362 - assert!(
363 - !releases.join(name).exists(),
364 - "expected to be pruned: {name}"
365 - );
332 + assert!(!releases.join(name).exists(), "expected to be pruned: {name}");
366 333 }
367 334 }
368 335
@@ -401,24 +368,32 @@
401 368 ssh_target: "deploy@192.0.2.1".into(),
402 369 release_root: "/opt/never".into(),
403 370 service_name: "makenotwork.service".into(),
371 + actuate: crate::topology::default_actuate(),
372 + observe: crate::topology::default_observe(),
404 373 };
374 + let executor = SshExec::new(
375 + node.ssh_target.clone(),
376 + CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
377 + );
405 378
406 - let result = deploy_node(&node, "0.0.1", &staged, "server").await;
379 + let result = deploy_node(&executor, &node, "0.0.1", &staged, "server").await;
407 380 let err = result.expect_err("deploy to unreachable host should fail");
408 381 let msg = format!("{err:#}");
409 - // The ssh helper returns `ssh <target> failed: ...`. Don't pin the
410 - // exact wording, just that the failure is attributed and that no
411 - // panic / hang happened.
382 + // Don't pin exact wording, just that the failure is attributed (ssh /
383 + // rsync / connection) and that no panic / hang happened.
412 384 assert!(
413 - msg.contains("ssh") || msg.contains("rsync") || msg.contains("connection"),
385 + msg.contains("ssh")
386 + || msg.contains("rsync")
387 + || msg.contains("connection")
388 + || msg.contains("Connection"),
414 389 "unexpected error: {msg}"
415 390 );
416 391 }
417 392
418 393 #[tokio::test]
419 394 async fn deploy_node_with_local_ssh_target_swaps_symlink() {
420 - // ssh_target="local" should route to the local fast-path: just a
421 - // symlink swap, no remote calls. Helpful for dev loops.
395 + // ssh_target="local" routes to the local fast-path: just a symlink
396 + // swap, no remote calls.
422 397 let tmp = tempfile::tempdir().unwrap();
423 398 let release_root = tmp.path().to_path_buf();
424 399 let staged = release_root.join("releases").join("0.0.1");
@@ -430,11 +405,36 @@
430 405 ssh_target: "local".into(),
431 406 release_root: release_root.to_string_lossy().into_owned(),
432 407 service_name: "makenotwork.service".into(),
408 + actuate: crate::topology::default_actuate(),
409 + observe: crate::topology::default_observe(),
433 410 };
411 + let executor = local_executor();
434 412
435 - let out = deploy_node(&node, "0.0.1", &staged, "server").await.unwrap();
413 + let out = deploy_node(&executor, &node, "0.0.1", &staged, "server").await.unwrap();
436 414 assert_eq!(out, staged);
437 415 let target = tokio::fs::read_link(release_root.join("current")).await.unwrap();
438 416 assert_eq!(target.to_string_lossy(), "releases/0.0.1");
439 417 }
418 +
419 + #[tokio::test]
420 + async fn deploy_node_denied_when_executor_lacks_deploy_grant() {
421 + // Defense in depth: an executor without the deploy grant refuses the
422 + // step before any filesystem / ssh action.
423 + let tmp = tempfile::tempdir().unwrap();
424 + let release_root = tmp.path().to_path_buf();
425 + let staged = release_root.join("releases").join("0.0.1");
426 + tokio::fs::create_dir_all(&staged).await.unwrap();
427 +
428 + let node = crate::topology::Node {
429 + name: "local-dev".into(),
430 + ssh_target: "local".into(),
431 + release_root: release_root.to_string_lossy().into_owned(),
432 + service_name: "makenotwork.service".into(),
433 + actuate: vec!["restart".into()], // no deploy
434 + observe: vec![],
435 + };
436 + let executor = LocalExec::new(CapabilitySet::from_tokens(["restart"], Vec::<&str>::new()));
437 + let err = deploy_node(&executor, &node, "0.0.1", &staged, "server").await.unwrap_err();
438 + assert!(format!("{err:#}").contains("capability denied"), "expected capability denial");
439 + }
440 440 }
@@ -73,6 +73,18 @@
73 73 pub fn chunks_emitted(&self) -> u32 { self.seq }
74 74 }
75 75
76 + /// Lets a Sando `LiveLog` be used directly as an [`ops_exec::LogSink`], so a
77 + /// gate or deploy run through `executor.run_streaming` streams into the same
78 + /// on-disk file + `GateLogChunk` broadcast the bespoke runner used. Delegates
79 + /// to the inherent [`LiveLog::write_chunk`] (inherent methods take precedence,
80 + /// so this is not recursive).
81 + #[async_trait::async_trait]
82 + impl ops_exec::LogSink for LiveLog {
83 + async fn write_chunk(&mut self, bytes: &[u8]) {
84 + self.write_chunk(bytes).await;
85 + }
86 + }
87 +
76 88 async fn open_for_append(path: &Path) -> Option<File> {
77 89 if let Some(parent) = path.parent() {
78 90 if let Err(e) = tokio::fs::create_dir_all(parent).await {
@@ -32,6 +32,7 @@
32 32
33 33 let prom = metrics::init();
34 34 let addr: SocketAddr = cfg.listen.parse()?;
35 + let executors = Arc::new(state::build_executors(&topo));
35 36 let app_state = state::AppState {
36 37 pool,
37 38 topo,
@@ -39,6 +40,7 @@
39 40 prom,
40 41 active_build: Arc::new(tokio::sync::Mutex::new(None)),
41 42 events: events::channel(),
43 + executors,
42 44 };
43 45 let app = routes::router(app_state);
44 46 tracing::info!(%addr, "sando daemon listening");
@@ -219,7 +219,9 @@
219 219 crate::events::emit(&s.events, crate::events::Event::DeployStart {
220 220 tier: target.name.clone(), node: node.name.clone(), version: version.clone(),
221 221 });
222 - let result = crate::deploy::deploy_node(node, &version_str, &staged_dir, s.cfg.primary_bin()).await;
222 + let executor = s.executors.get(&node.name).cloned()
223 + .unwrap_or_else(|| crate::state::build_executor(node));
224 + let result = crate::deploy::deploy_node(executor.as_ref(), node, &version_str, &staged_dir, s.cfg.primary_bin()).await;
223 225 let finished = chrono::Utc::now().to_rfc3339();
224 226 let (outcome_obj, err_for_propagation) = match result {
225 227 Ok(_) => (crate::outcome::DeployOutcome::ok(), None),
@@ -374,7 +376,9 @@
374 376 .to_path_buf();
375 377
376 378 for node in &target.nodes {
377 - crate::deploy::deploy_node(node, &previous_str, &staged_dir, s.cfg.primary_bin())
379 + let executor = s.executors.get(&node.name).cloned()
380 + .unwrap_or_else(|| crate::state::build_executor(node));
381 + crate::deploy::deploy_node(executor.as_ref(), node, &previous_str, &staged_dir, s.cfg.primary_bin())
378 382 .await
379 383 .map_err(crate::error::Error::Other)?;
380 384 }
@@ -630,6 +634,8 @@
630 634 ssh_target: "local".into(),
631 635 release_root: "/tmp/a-node".into(),
632 636 service_name: "makenotwork.service".into(),
637 + actuate: crate::topology::default_actuate(),
638 + observe: crate::topology::default_observe(),
633 639 }],
634 640 },
635 641 ],
@@ -664,13 +670,16 @@
664 670 // Don't call install_recorder in tests — it touches a process-global
665 671 // and conflicts when tests run in parallel.
666 672 let prom = PrometheusBuilder::new().build_recorder().handle();
673 + let topo = test_topo();
674 + let executors = Arc::new(crate::state::build_executors(&topo));
667 675 AppState {
668 676 pool,
669 - topo: Arc::new(test_topo()),
677 + topo: Arc::new(topo),
670 678 cfg: Arc::new(test_cfg()),
671 679 prom,
672 680 active_build: Arc::new(tokio::sync::Mutex::new(None)),
673 681 events: crate::events::channel(),
682 + executors,
674 683 }
675 684 }
676 685
@@ -1,12 +1,20 @@
1 1 use crate::config::Config;
2 + use crate::domain::NodeId;
2 3 use crate::events::EventTx;
3 - use crate::topology::Topology;
4 + use crate::topology::{Node, Topology};
4 5 use metrics_exporter_prometheus::PrometheusHandle;
6 + use ops_exec::{CapabilitySet, Executor, LocalExec, SshExec};
5 7 use sqlx::SqlitePool;
8 + use std::collections::HashMap;
6 9 use std::sync::Arc;
7 10 use tokio::sync::Mutex;
8 11 use tokio::task::AbortHandle;
9 12
13 + /// Per-node executors keyed by node id, built once from the topology at
14 + /// startup. The deploy path looks a node's executor up here instead of
15 + /// constructing ssh/rsync invocations inline.
16 + pub type ExecutorMap = HashMap<NodeId, Arc<dyn Executor>>;
17 +
10 18 #[derive(Clone)]
11 19 pub struct AppState {
12 20 pub pool: SqlitePool,
@@ -19,4 +27,27 @@
19 27 /// Broadcast bus for live operator events. WS /events subscribes; all
20 28 /// build/gate/deploy code sites emit on this.
21 29 pub events: EventTx,
30 + /// One capability-scoped [`Executor`] per node, built from the topology.
31 + pub executors: Arc<ExecutorMap>,
32 + }
33 +
34 + /// Build the executor for one node: a `LocalExec` for the `local` fast-path, an
35 + /// `SshExec` otherwise, each granted the node's declared capabilities (which
36 + /// default to deploy+restart / observe health — the historical behavior).
37 + pub fn build_executor(node: &Node) -> Arc<dyn Executor> {
38 + let caps = CapabilitySet::from_tokens(&node.actuate, &node.observe);
39 + if node.ssh_target == "local" || node.ssh_target.is_empty() {
40 + Arc::new(LocalExec::new(caps))
41 + } else {
42 + Arc::new(SshExec::new(node.ssh_target.clone(), caps))
43 + }
44 + }
45 +
46 + /// Build the full node → executor map from every tier's nodes.
47 + pub fn build_executors(topo: &Topology) -> ExecutorMap {
48 + topo.tiers
49 + .iter()
50 + .flat_map(|t| t.nodes.iter())
51 + .map(|node| (node.name.clone(), build_executor(node)))
52 + .collect()
22 53 }
@@ -152,6 +152,8 @@
152 152 ssh_target: format!("deploy@{name}"),
153 153 release_root: "/opt/mnw".into(),
154 154 service_name: "makenotwork.service".into(),
155 + actuate: crate::topology::default_actuate(),
156 + observe: crate::topology::default_observe(),
155 157 }
156 158 }
157 159
@@ -44,10 +44,23 @@
44 44 /// Defaults to "makenotwork.service" because that's MNW's prod unit.
45 45 #[serde(default = "default_service_name")]
46 46 pub service_name: String,
47 + /// Capability grant for this node's executor (see `ops_exec`). Defaults to
48 + /// the current behavior of every Sando node — actuate deploy+restart,
49 + /// observe health — so an existing `sando.toml` keeps working unedited.
50 + #[serde(default = "default_actuate")]
51 + pub actuate: Vec<String>,
52 + #[serde(default = "default_observe")]
53 + pub observe: Vec<String>,
47 54 }
48 55
49 56 fn default_service_name() -> String { "makenotwork.service".into() }
50 57
58 + /// The capability set every pre-existing Sando node implicitly had: it deploys
59 + /// and restarts, and is health-observed. Keeping these as the defaults is what
60 + /// lets `sando.toml` stay unchanged through the executor refactor.
61 + pub fn default_actuate() -> Vec<String> { vec!["deploy".into(), "restart".into()] }
62 + pub fn default_observe() -> Vec<String> { vec!["health".into()] }
63 +
51 64 #[derive(Debug, Clone, Copy, Serialize, Deserialize, Default)]
52 65 #[serde(rename_all = "snake_case")]
53 66 pub enum CanaryPolicy {