max / makenotwork

135.4 KB · 3574 lines History Blame Raw

1	use crate::error::Result;
2	use crate::state::AppState;
3	use axum::extract::{Path, Query, State, WebSocketUpgrade};
4	use axum::response::IntoResponse;
5	use axum::routing::{get, post};
6	use axum::{Json, Router};
7	use serde::{Deserialize, Serialize};
8	use sqlx::Row;
9
10	mod promotion;
11	use promotion::{clear_partial, promote_inner, set_partial};
12
13	pub fn router(state: AppState) -> Router {
14	let token = state.api_token.clone();
15
16	// Every route requires the bearer token (when one is configured): the
17	// mutators carry prod-deploy authority (CF2), and the reads expose prod
18	// state — deployed versions, build SHAs, full gate logs, the live event
19	// stream — that a tailnet peer should not get unauthenticated. There are no
20	// unauthenticated routes: /metrics used to be the one opt-out, but the
21	// Prometheus exporter was removed rather than leave an auth posture on it
22	// that no scraper actually used. The bearer-success path logs the caller
23	// identity for mutating (POST) requests so a prod ship is attributable.
24	let protected = Router::new()
25	// mutators — prod-deploy authority
26	.route("/promote/{tier}", post(promote))
27	.route("/rollback/{tier}", post(rollback))
28	.route("/rebuild", post(rebuild))
29	.route("/intake", post(intake))
30	.route("/self-update", post(self_update))
31	.route("/confirm/{tier}", post(confirm))
32	.route("/backup/fetch", post(backup_fetch))
33	// reads — prod state, now also gated
34	.route("/state", get(get_state))
35	.route("/status.json", get(get_status_json))
36	.route("/runs/{id}", get(get_run))
37	.route("/runs/{id}/wait", get(get_run_wait))
38	.route("/logs/{version}/{gate}", get(get_gate_log))
39	.route("/events", get(events_ws))
40	.route_layer(axum::middleware::from_fn(move \|req, next\| {
41	require_bearer(token.clone(), req, next)
42	}));
43
44	Router::new().merge(protected).with_state(state)
45	}
46
47	/// The daemon's whole HTTP surface: one product's routes at the root, and every
48	/// product's under `/apps/<id>`.
49	///
50	/// A product is addressed by mounting a whole router per product rather than by
51	/// a `{app}` path parameter every handler has to read. [`AppState`] already
52	/// carries one product's view — its config, topology and node executors — so
53	/// giving each mount its own state means no handler can address the wrong
54	/// product by forgetting to look at a parameter. The mistake is unavailable
55	/// rather than guarded against.
56	///
57	/// The unprefixed paths stay, meaning the default product. `/promote/b` is what
58	/// the deploy runbook says, what the TUI calls, and what an operator types under
59	/// pressure; making all of that conditional on a rename would be a cost paid by
60	/// the wrong people.
61	pub fn router_for_apps(state: AppState) -> Router {
62	// The root mount. `state` already holds the default product's view.
63	let mut r = router(state.clone());
64	for (id, app) in state.apps.iter() {
65	let scoped = AppState {
66	cfg: app.cfg.clone(),
67	topo: app.topo.clone(),
68	executors: app.executors.clone(),
69	..state.clone()
70	};
71	r = r.nest(&format!("/apps/{id}"), router(scoped));
72	}
73	r.merge(apps_index(state))
74	}
75
76	/// `GET /apps` — which products this daemon ships, and which one the unprefixed
77	/// routes address. Behind the same bearer gate as everything else: it describes
78	/// the deploy surface.
79	fn apps_index(state: AppState) -> Router {
80	let token = state.api_token.clone();
81	let protected =
82	Router::new()
83	.route("/apps", get(list_apps))
84	.route_layer(axum::middleware::from_fn(move \|req, next\| {
85	require_bearer(token.clone(), req, next)
86	}));
87	Router::new().merge(protected).with_state(state)
88	}
89
90	#[derive(serde::Serialize)]
91	pub struct AppsView {
92	/// Every product, in configured order.
93	pub apps: Vec<String>,
94	/// The one the unprefixed routes act on.
95	pub default_app: String,
96	}
97
98	async fn list_apps(State(s): State<AppState>) -> Json<AppsView> {
99	Json(AppsView {
100	apps: s.app_ids().iter().map(ToString::to_string).collect(),
101	default_app: s.default_app.to_string(),
102	})
103	}
104
105	/// Bearer-token gate for the deploy mutators. When no token is configured the
106	/// request passes (main() only allows that on a loopback bind). Comparison is
107	/// constant-time to avoid leaking the token via timing.
108	async fn require_bearer(
109	token: Option<std::sync::Arc<str>>,
110	req: axum::extract::Request,
111	next: axum::middleware::Next,
112	) -> axum::response::Response {
113	let Some(expected) = token.as_deref() else {
114	return next.run(req).await;
115	};
116	let path = req.uri().path().to_string();
117	let method = req.method().clone();
118	let ok = req
119	.headers()
120	.get(axum::http::header::AUTHORIZATION)
121	.and_then(\|h\| h.to_str().ok())
122	.and_then(\|h\| h.strip_prefix("Bearer "))
123	.is_some_and(\|t\| ops_core::daemon::ct_eq(t, expected));
124	if ok {
125	// Attribute mutations (POST) to a caller. Reads are GET and poll every
126	// few seconds, so logging them would drown the journal; the audit value
127	// is in who shipped/rolled back prod, which is always a POST. The peer
128	// address is the closest identity we have without per-operator tokens
129	// (`tailscale whois` is a future refinement); a shared token at least
130	// gets a source host into the trail.
131	if method == axum::http::Method::POST {
132	let peer = req
133	.extensions()
134	.get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
135	.map_or_else(\|\| "unknown".into(), \|ci\| ci.0.to_string());
136	tracing::info!(path = %path, peer = %peer, "authenticated deploy mutation");
137	}
138	next.run(req).await
139	} else {
140	tracing::warn!(path = %path, "rejected unauthenticated request");
141	(
142	axum::http::StatusCode::UNAUTHORIZED,
143	"missing or invalid bearer token\n",
144	)
145	.into_response()
146	}
147	}
148
149	#[derive(Serialize)]
150	pub(crate) struct StateView {
151	/// The running sandod's own package version. Lets a self-update caller
152	/// confirm the new binary is live after the restart (the tier versions are
153	/// the deployed product, not the controller).
154	pub(crate) sandod_version: &'static str,
155	pub(crate) tiers: Vec<TierView>,
156	/// The most recent build run (the resource `GET /runs/{id}` exposes in
157	/// full). Surfaced here so a `/state` poller sees an in-flight or failed
158	/// build — the tier versions only ever reflect the last success, so
159	/// without this `/state` looks frozen for the whole build. `null` until
160	/// the first `/rebuild`.
161	pub(crate) build: Option<crate::runs::BuildSummary>,
162	}
163
164	#[derive(Serialize)]
165	pub(crate) struct TierView {
166	pub(crate) name: String,
167	pub(crate) ord: i64,
168	pub(crate) provisioned: bool,
169	pub(crate) canary: String,
170	pub(crate) current_version: Option<String>,
171	pub(crate) previous_version: Option<String>,
172	pub(crate) burn_in_started_at: Option<String>,
173	/// Non-null when the tier was left in a partial / mixed-version state by a
174	/// failed promote or rollback whose compensation could not fully restore
175	/// consistency. NULL when the tier is consistent. The TUI flags it red.
176	pub(crate) partial_reason: Option<String>,
177	pub(crate) nodes: Vec<String>,
178	pub(crate) gates: Vec<GateView>,
179	}
180
181	#[derive(Serialize)]
182	pub(crate) struct GateView {
183	pub(crate) kind: String,
184	pub(crate) finished_at: Option<String>,
185	/// `'passed' \| 'failed' \| 'blocked'` or NULL while in-flight. The TUI
186	/// uses this to choose green/red/yellow rendering.
187	pub(crate) status: Option<String>,
188	/// Full typed `GateOutcome` as a JSON object, when present.
189	/// Deserialized lazily by the consumer; sandod doesn't re-parse it.
190	pub(crate) outcome: Option<serde_json::Value>,
191	/// Relative path under `cfg.logs_root` to the persisted stdout/stderr.
192	pub(crate) log_ref: Option<String>,
193	}
194
195	async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
196	Ok(Json(state_view(&s).await?))
197	}
198
199	/// `GET /status.json` — the same state, in the shared cross-service payload
200	/// every operator surface renders. See `crate::status`.
201	async fn get_status_json(State(s): State<AppState>) -> Result<Json<ops_status::Payload>> {
202	let view = state_view(&s).await?;
203	Ok(Json(crate::status::payload(&view, chrono::Utc::now())))
204	}
205
206	/// The shared read behind `/state` and `/status.json`.
207	pub(crate) async fn state_view(s: &AppState) -> Result<StateView> {
208	let rows = sqlx::query(
209	"SELECT t.name, t.ord, t.provisioned, t.canary,
210	ts.current_version, ts.previous_version, ts.burn_in_started_at,
211	ts.partial_reason
212	FROM tiers t
213	LEFT JOIN tier_state ts ON ts.app = t.app AND ts.tier = t.name
214	WHERE t.app = ?
215	ORDER BY t.ord",
216	)
217	.bind(&s.cfg.id)
218	.fetch_all(&s.pool)
219	.await?;
220
221	let mut tiers: Vec<TierView> = Vec::with_capacity(rows.len());
222	for r in rows {
223	let name: String = r.get("name");
224	let current_version: Option<String> = r.get("current_version");
225
226	let nodes: Vec<String> =
227	sqlx::query_scalar("SELECT name FROM nodes WHERE app = ? AND tier = ? ORDER BY name")
228	.bind(&s.cfg.id)
229	.bind(&name)
230	.fetch_all(&s.pool)
231	.await?;
232
233	// Surface gates for current_version when set, otherwise for the most
234	// recently attempted version on this tier. Without the fallback, a
235	// tier that has never gone green (MM after a build failure, B before
236	// first deploy) exposes no gate detail via /state — debugging required
237	// SSH and direct SQLite access. See sando todo: gate observability.
238	let gate_version: Option<String> = if current_version.is_some() {
239	current_version.clone()
240	} else {
241	sqlx::query_scalar(
242	"SELECT version FROM gate_runs WHERE app = ? AND tier = ?
243	ORDER BY id DESC LIMIT 1",
244	)
245	.bind(&s.cfg.id)
246	.bind(&name)
247	.fetch_optional(&s.pool)
248	.await?
249	};
250
251	let gates: Vec<GateView> = if let Some(ver) = gate_version.as_ref() {
252	// Most recent gate_runs row per gate_kind for (tier, ver).
253	sqlx::query(
254	"SELECT gate_kind, finished_at, status, outcome_json, log_ref
255	FROM gate_runs g
256	WHERE app = ?1 AND tier = ?2 AND version = ?3
257	AND id = (SELECT MAX(id) FROM gate_runs
258	WHERE app = ?1 AND tier = ?2 AND version = ?3
259	AND gate_kind = g.gate_kind)
260	ORDER BY gate_kind",
261	)
262	.bind(&s.cfg.id)
263	.bind(&name)
264	.bind(ver)
265	.fetch_all(&s.pool)
266	.await?
267	.into_iter()
268	.map(\|gr\| GateView {
269	kind: gr.get("gate_kind"),
270	finished_at: gr.get("finished_at"),
271	status: gr.get("status"),
272	outcome: gr
273	.get::<Option<String>, _>("outcome_json")
274	.and_then(\|s\| serde_json::from_str(&s).ok()),
275	log_ref: gr.get("log_ref"),
276	})
277	.collect()
278	} else {
279	Vec::new()
280	};
281
282	tiers.push(TierView {
283	name,
284	ord: r.get("ord"),
285	provisioned: r.get::<i64, _>("provisioned") != 0,
286	canary: r.get("canary"),
287	current_version,
288	previous_version: r.get("previous_version"),
289	burn_in_started_at: r.get("burn_in_started_at"),
290	partial_reason: r.get("partial_reason"),
291	nodes,
292	gates,
293	});
294	}
295
296	let build = crate::runs::latest_summary(&s.pool, &s.cfg.id).await?;
297	Ok(StateView {
298	sandod_version: env!("CARGO_PKG_VERSION"),
299	tiers,
300	build,
301	})
302	}
303
304	#[derive(Deserialize, Default)]
305	struct PromoteBody {
306	/// Optional. If absent, defaults to the predecessor tier's `current_version`
307	/// (i.e. promote whatever just finished baking on the previous tier).
308	#[serde(default)]
309	version: Option<String>,
310	#[serde(default)]
311	hotfix: bool,
312	#[serde(default)]
313	reset_burn_in: bool,
314	/// Set when the release carries a database migration. Rollback restores the
315	/// binary and release_contents only, never the schema (MNW migrates forward
316	/// on boot and has no down path), so a migration-bearing promote is one-way.
317	/// This forces a fresh `manual_confirm` on the predecessor tier even if the
318	/// tier does not configure one, so the operator consciously acknowledges the
319	/// rollback contract. `hotfix` does not suppress it. See
320	/// `deploy/README.md` "Rollback contract".
321	#[serde(default)]
322	bears_migration: bool,
323	}
324
325	async fn promote(
326	State(s): State<AppState>,
327	Path(tier): Path<String>,
328	body: Option<Json<PromoteBody>>,
329	) -> Result<Json<serde_json::Value>> {
330	let body = body.map(\|Json(b)\| b).unwrap_or_default();
331	// Run the deploy in a detached task, not inline in the request future.
332	// A promote's rsync + restart can outlast a client's socket timeout; if the
333	// client disconnects, axum drops the handler future — and if the deploy ran
334	// inline that would abandon the rollout mid-flight (symlink half-swapped,
335	// tier_state not advanced, and — before the kill_on_drop fix — an orphaned
336	// rsync). Spawning detaches the work from the connection: the task owns the
337	// deploy lock and runs to completion regardless, while a still-connected
338	// client still receives the result. Mirrors how /rebuild already runs its
339	// build off-request.
340	tokio::spawn(promote_inner(s, tier, body))
341	.await
342	.map_err(\|e\| {
343	crate::error::Error::Other(anyhow::anyhow!("promote task failed to join: {e}"))
344	})?
345	}
346
347	async fn rollback(
348	State(s): State<AppState>,
349	Path(tier): Path<String>,
350	) -> Result<Json<serde_json::Value>> {
351	// Serialize against any concurrent promote/rollback (CF3).
352	let _deploy_guard = s.deploy_lock.lock().await;
353	let tier = crate::domain::TierId::new(tier);
354	let target = s
355	.topo
356	.tiers
357	.iter()
358	.find(\|t\| t.name == tier)
359	.ok_or(crate::error::Error::NotFound)?;
360
361	let row: Option<(Option<String>, Option<String>)> = sqlx::query_as(
362	"SELECT current_version, previous_version FROM tier_state WHERE app = ? AND tier = ?",
363	)
364	.bind(&s.cfg.id)
365	.bind(&tier)
366	.fetch_optional(&s.pool)
367	.await
368	.map_err(crate::error::Error::Db)?;
369	let (Some(current_str), Some(previous_str)) = row.unwrap_or((None, None)) else {
370	return Err(crate::error::Error::GateBlocked(
371	"no previous_version to roll back to".into(),
372	));
373	};
374	let current = crate::domain::Version::parse(&current_str)
375	.map_err(\|e\| crate::error::Error::Other(anyhow::anyhow!(e)))?;
376	let previous = crate::domain::Version::parse(&previous_str)
377	.map_err(\|e\| crate::error::Error::Other(anyhow::anyhow!(e)))?;
378
379	let bin: Option<(String,)> =
380	sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?")
381	.bind(&s.cfg.id)
382	.bind(&previous)
383	.fetch_optional(&s.pool)
384	.await
385	.map_err(crate::error::Error::Db)?;
386	let Some((bin,)) = bin else {
387	return Err(crate::error::Error::GateBlocked(format!(
388	"previous version {previous} has no artifact_path; rollback impossible"
389	)));
390	};
391	let bin_path = std::path::PathBuf::from(bin);
392	let staged_dir = bin_path
393	.parent()
394	.ok_or_else(\|\| crate::error::Error::Other(anyhow::anyhow!("artifact_path has no parent")))?
395	.to_path_buf();
396
397	// Deploy `previous` to every node. A failure partway through leaves the fleet
398	// split-brain (some on `previous`, some still on `current`); record exactly how
399	// far we got so /state surfaces it rather than reporting a clean rollback.
400	let total = target.nodes.len();
401	// Same per-node resolution as a promote: on a two-architecture product the
402	// version being rolled back to is also two bundles, and each node takes its
403	// own. Resolved before any node is touched, so a version whose other half
404	// was never accepted fails the rollback whole rather than partway.
405	let target_nodes: Vec<&crate::topology::Node> = target.nodes.iter().collect();
406	// No fallback build id: a rollback re-runs no gates, so there is no evidence
407	// to key on. These are bytes the tier already ran.
408	let bundles =
409	promotion::bundles_for_nodes(&s, &previous_str, &target_nodes, &staged_dir, None, None)
410	.await?;
411	for (i, (node, node_bundle, node_bundle_platform, _)) in bundles.iter().enumerate() {
412	let executor = s
413	.executors
414	.get(&node.name)
415	.cloned()
416	.unwrap_or_else(\|\| crate::state::build_executor(node));
417	let placement =
418	crate::deploy::Placement::check(node, node_bundle, node_bundle_platform.as_ref())
419	.map_err(\|e\| crate::error::Error::GateBlocked(e.to_string()))?;
420	if let Err(e) = crate::deploy::deploy_node(
421	executor.as_ref(),
422	placement,
423	&previous_str,
424	s.cfg.primary_bin(),
425	)
426	.await
427	{
428	set_partial(
429	&s,
430	&tier,
431	&format!(
432	"rollback to {previous} incomplete: {i}/{total} nodes rolled back, \
433	node {} failed; remaining nodes still on {current} — manual check needed",
434	node.name,
435	),
436	)
437	.await;
438	tracing::error!(tier = %tier, node = %node.name, done = i, of = total,
439	to = %previous, "rollback failed mid-fleet; tier left partial");
440	return Err(crate::error::Error::Other(e));
441	}
442	}
443
444	// `previous_version` becomes NULL, not `current`. Sando tracks exactly one
445	// step of history, so writing the version we just rolled off into the
446	// rollback slot makes a second /rollback roll FORWARD onto the build the
447	// operator was escaping. NULL makes the second call fail loudly with "no
448	// previous_version to roll back to" — honest about the depth we keep.
449	// Stepping back further is the deploys-table walk we deliberately don't do.
450	//
451	// `advanced_at = now` even though burn_in_started_at is nulled: a rollback IS
452	// a change to the deployed identity, so the startup reconcile must treat the
453	// roll-off `deploys` row as predating this state — otherwise a cleanly
454	// rolled-back tier would look like an unrecorded deploy (migration 009 /
455	// [`crate::reconcile`]).
456	let now = chrono::Utc::now().to_rfc3339();
457	sqlx::query(
458	"UPDATE tier_state
459	SET current_version = ?, previous_version = NULL,
460	burn_in_started_at = NULL, advanced_at = ?
461	WHERE app = ? AND tier = ?",
462	)
463	.bind(&previous)
464	.bind(&now)
465	.bind(&s.cfg.id)
466	.bind(&tier)
467	.execute(&s.pool)
468	.await
469	.map_err(crate::error::Error::Db)?;
470
471	// Full rollback succeeded on every node: the tier is consistent again.
472	clear_partial(&s, &tier).await;
473
474	tracing::warn!(tier = %tier, from = %current, to = %previous, "rollback complete");
475	crate::events::emit(
476	&s.events,
477	crate::events::Event::Rollback {
478	tier: tier.clone(),
479	from: current.clone(),
480	to: previous.clone(),
481	},
482	);
483
484	Ok(Json(serde_json::json!({
485	"tier": tier,
486	"rolled_back_from": current,
487	"now_running": previous,
488	})))
489	}
490
491	#[derive(Deserialize, Default)]
492	struct RebuildBody {
493	/// Specific sha to build. If absent, resolve `topo.repo.branch` from the bare repo.
494	#[serde(default)]
495	sha: Option<String>,
496	}
497
498	#[derive(Deserialize)]
499	struct IntakeBody {
500	/// Directory holding the assembled bundle, which must already be under this
501	/// app's `release_root/staging/`. Publishing is an atomic same-filesystem
502	/// rename, so a path anywhere else would silently become a copy. Getting the
503	/// bytes there is the builder's job; this endpoint decides whether they may
504	/// stay.
505	staged: String,
506	/// The builder's `ArtifactRecord`, verbatim. Bento writes one per target at
507	/// `dist_root/<app>/<version>/<target-slug>/record.json` and sends it here
508	/// rather than inside the bundle: the record names the digest of the bundle,
509	/// and the digest covers every file in it, so a record shipped among the
510	/// artifacts would change the digest it names.
511	record: String,
512	}
513
514	/// `POST /intake` — accept an artifact this daemon did not build, and take it
515	/// through the same host-tier gating a locally-built one gets.
516	///
517	/// This is the Sando half of the Bento boundary (wiki [[sando-bento-boundary]]):
518	/// Bento builds and packages, Sando decides whether a thing advances a stage.
519	/// The bundle arrives already assembled, with no worktree and no compiler run,
520	/// and joins the pipeline at the point where a Sando build has just published
521	/// its own bundle.
522	///
523	/// The bytes are proved against the record before anything else happens, so a
524	/// bundle that drifted in transit is refused with the offending file named
525	/// rather than gated and shipped.
526	async fn intake(
527	State(s): State<AppState>,
528	Json(body): Json<IntakeBody>,
529	) -> Result<Json<serde_json::Value>> {
530	let staged = std::path::PathBuf::from(&body.staged);
531	// The staging dir has to be under this app's release root, and this is the
532	// one place that can be enforced: `intake::accept` renames out of it, and a
533	// path elsewhere would either cross a filesystem or reach somewhere the
534	// caller should not be able to name.
535	let staging_root = s.cfg.release_root.join("staging");
536	let staged_abs = staged.canonicalize().map_err(\|e\| {
537	crate::error::Error::Other(anyhow::anyhow!("staged path {}: {e}", body.staged))
538	})?;
539	let staging_root_abs = staging_root.canonicalize().map_err(\|e\| {
540	crate::error::Error::Other(anyhow::anyhow!(
541	"staging root {}: {e}",
542	staging_root.display()
543	))
544	})?;
545	if !staged_abs.starts_with(&staging_root_abs) {
546	return Err(crate::error::Error::GateBlocked(format!(
547	"staged bundle {} is not under {}; publishing is an atomic rename \
548	within the release root",
549	staged_abs.display(),
550	staging_root_abs.display(),
551	)));
552	}
553
554	// The run row is created before the record is parsed so a refused intake is
555	// still visible as an attempt. Its sha comes from the record's provenance,
556	// which is the artifact's own claim about what it was built from; `accept`
557	// is what decides whether to believe the document at all.
558	let sha = serde_json::from_str::<serde_json::Value>(&body.record)
559	.ok()
560	.and_then(\|v\| {
561	v.get("provenance")
562	.and_then(\|p\| p.get("git_sha"))
563	.and_then(\|g\| g.as_str())
564	.map(str::to_owned)
565	})
566	.unwrap_or_default();
567	let run_id = crate::runs::create(&s.pool, &s.cfg.id, &sha)
568	.await
569	.map_err(crate::error::Error::Other)?;
570
571	tracing::info!(staged = %staged_abs.display(), run_id = %run_id, "artifact intake requested");
572
573	// Prove the bytes BEFORE answering. This has to be synchronous, and the
574	// reason is the whole contract: the producer treats a non-success here as a
575	// failed handoff and fails its own build on it. Proving in a spawned task
576	// and answering `accepted: true` first would tell Bento the artifact landed
577	// and only then discover it had not — a corrupt bundle would go green
578	// upstream and be visible solely in this daemon's run row, which is exactly
579	// the "evidence vouches for one thing, the deploy ships another" class the
580	// boundary exists to close.
581	//
582	// It is affordable: hashing the bundle is the work the transfer just did.
583	let accepted =
584	match crate::build::accept_intake(&s.pool, &s.cfg, &staged_abs, &body.record, run_id).await
585	{
586	Ok(p) => p,
587	Err(e) => {
588	let msg = format!("{e:#}");
589	tracing::error!(error = %msg, "refused an intake");
590	crate::runs::mark_failed(&s.pool, run_id, &msg).await.ok();
591	return Err(crate::error::Error::GateBlocked(msg));
592	}
593	};
594
595	// Gating is not. It runs the host tier's gates, which can take an hour, and
596	// whether the artifact advances is this daemon's business rather than the
597	// builder's. The producer's job ended when the bytes were believed.
598	let pool = s.pool.clone();
599	let pool_for_task = s.pool.clone();
600	let cfg = s.cfg.clone();
601	let topo = s.topo.clone();
602	let events = s.events.clone();
603	let deploy_lock = s.deploy_lock.clone();
604	tokio::spawn(async move {
605	if let Err(e) =
606	crate::build::gate_intake(pool, cfg, topo, accepted, events, run_id, deploy_lock).await
607	{
608	tracing::error!(error = %e, "gating an accepted artifact failed");
609	crate::runs::mark_failed(&pool_for_task, run_id, &format!("{e:#}"))
610	.await
611	.ok();
612	}
613	});
614
615	Ok(Json(
616	serde_json::json!({ "accepted": true, "run_id": run_id.0 }),
617	))
618	}
619
620	async fn rebuild(
621	State(s): State<AppState>,
622	body: Option<Json<RebuildBody>>,
623	) -> Result<Json<serde_json::Value>> {
624	let body = body.map(\|Json(b)\| b).unwrap_or_default();
625	let sha = match body.sha {
626	Some(sha) => sha,
627	None => {
628	// Omitted sha = "build the deploy branch's tip". Fetch upstream
629	// first so we resolve the upstream HEAD, not a possibly-stale
630	// local branch ref — the build task fetches too, but only after the
631	// sha is already chosen, so without this `/rebuild {}` could build
632	// an old commit. A fetch failure is non-fatal: fall back to the
633	// current bare-repo tip (same policy as the build task).
634	let repo = s.topo.repo.as_ref().ok_or_else(\|\| {
635	crate::error::Error::GateBlocked(format!(
636	"{} is intake-only: it declares no [repo], so there is nothing to \
637	build from. Ship it with POST /intake.",
638	s.cfg.id
639	))
640	})?;
641	let bare = std::path::Path::new(&repo.bare_path);
642	if let Some(upstream) = repo.upstream.as_deref()
643	&& let Err(e) = crate::git::fetch_upstream(bare, upstream, &repo.branch).await
644	{
645	tracing::warn!(error = %e, "pre-resolve upstream fetch failed; resolving current bare-repo branch tip");
646	}
647	crate::git::resolve_ref(bare, &repo.branch)
648	.await
649	.map_err(crate::error::Error::Other)?
650	}
651	};
652
653	// Boundary parse: a sha entering Sando must be hex of plausible length.
654	// The build pipeline downstream only ever sees `GitSha`.
655	let sha = crate::domain::GitSha::parse(&sha)
656	.map_err(\|e\| crate::error::Error::Other(anyhow::anyhow!(e)))?;
657
658	tracing::info!(sha = %sha, "rebuild requested");
659	crate::events::emit(
660	&s.events,
661	crate::events::Event::RebuildRequested { sha: sha.clone() },
662	);
663
664	// One pollable resource per triggered build. Created before the spawn so
665	// the run id is in the response even if the task is aborted milliseconds
666	// later by a still-newer /rebuild.
667	let run_id = crate::runs::create(&s.pool, &s.cfg.id, sha.as_str())
668	.await
669	.map_err(crate::error::Error::Other)?;
670
671	// Latest /rebuild wins: abort any in-flight build before spawning a new
672	// one. Aborting drops the spawned task's future, which drops any
673	// tokio::process::Child it owns; with `kill_on_drop(true)` set on the
674	// cargo Command, SIGKILL propagates to cargo + its rustc children.
675	let mut slot = s.active_build.lock().await;
676	if let Some(prev) = slot.take()
677	&& !prev.handle.is_finished()
678	{
679	tracing::warn!("aborting in-flight build for newer /rebuild request");
680	crate::events::emit(
681	&s.events,
682	crate::events::Event::BuildAborted {
683	sha_aborted: sha.clone(),
684	},
685	);
686	prev.handle.abort();
687	// Aborting drops the task before it can settle its own row, so
688	// record the supersession here.
689	crate::runs::mark_aborted(&s.pool, prev.run_id).await.ok();
690	}
691
692	let pool = s.pool.clone();
693	let cfg = s.cfg.clone();
694	let topo = s.topo.clone();
695	let events_for_task = s.events.clone();
696	let sha_for_task = sha.clone();
697	let sha_response = sha.to_string();
698	let pool_for_task = s.pool.clone();
699	let deploy_lock = s.deploy_lock.clone();
700	let handle = tokio::spawn(async move {
701	if let Err(e) = crate::build::build_and_run_host(
702	pool,
703	cfg,
704	topo,
705	sha_for_task.clone(),
706	events_for_task,
707	run_id,
708	deploy_lock,
709	)
710	.await
711	{
712	tracing::error!(sha = %sha_for_task, error = %e, "rebuild pipeline failed");
713	// Pre-gate bails (fetch/checkout/version/scratch) don't settle the
714	// run themselves; the build-step compile error already did. First
715	// terminal write wins, so this is a safety net for the rest.
716	crate::runs::mark_failed(&pool_for_task, run_id, &format!("{e:#}"))
717	.await
718	.ok();
719	}
720	});
721	*slot = Some(crate::state::ActiveBuild {
722	handle: handle.abort_handle(),
723	run_id,
724	});
725
726	Ok(Json(
727	serde_json::json!({ "accepted": true, "sha": sha_response, "run_id": run_id.0 }),
728	))
729	}
730
731	/// `GET /runs/{id}` — the build-status resource a non-TUI driver polls after
732	/// `/rebuild`. Open (read-only) like `/state` and `/logs`.
733	async fn get_run(
734	State(s): State<AppState>,
735	Path(id): Path<i64>,
736	) -> Result<Json<crate::runs::RunView>> {
737	crate::runs::get(&s.pool, crate::domain::RunId(id))
738	.await
739	.map_err(crate::error::Error::Other)?
740	.map(Json)
741	.ok_or(crate::error::Error::NotFound)
742	}
743
744	#[derive(Deserialize)]
745	struct WaitParams {
746	/// How long to hold the request open before returning a still-building
747	/// run. Default 30s, capped at 120s.
748	#[serde(default)]
749	timeout_ms: Option<u64>,
750	}
751
752	/// `GET /runs/{id}/wait` — long-poll: hold the request open until the run
753	/// settles (`result != building`) or the timeout elapses, then return the
754	/// current `RunView`. Removes polling-cadence guessing for a headless driver
755	/// (fire `/rebuild`, block on `/wait`). On timeout the run is returned
756	/// still-building (200) and the caller re-issues `/wait`. 404 if unknown.
757	async fn get_run_wait(
758	State(s): State<AppState>,
759	Path(id): Path<i64>,
760	Query(p): Query<WaitParams>,
761	) -> Result<Json<crate::runs::RunView>> {
762	let run_id = crate::domain::RunId(id);
763	let timeout = std::time::Duration::from_millis(p.timeout_ms.unwrap_or(30_000).min(120_000));
764	let deadline = tokio::time::Instant::now() + timeout;
765	// Poll the row rather than wiring a per-run notifier: a build settles on
766	// the minute scale, so a sub-second tick is plenty responsive and the
767	// query is a single indexed read. The request releases its pool handle
768	// between ticks.
769	let tick = std::time::Duration::from_millis(750);
770	loop {
771	let view = crate::runs::get(&s.pool, run_id)
772	.await
773	.map_err(crate::error::Error::Other)?
774	.ok_or(crate::error::Error::NotFound)?;
775	let now = tokio::time::Instant::now();
776	if view.result != "building" \|\| now >= deadline {
777	return Ok(Json(view));
778	}
779	tokio::time::sleep((deadline - now).min(tick)).await;
780	}
781	}
782
783	#[derive(Deserialize)]
784	struct SelfUpdateBody {
785	/// The commit to rebuild sandod from. Must already be on the canonical
786	/// remote (the updater `git fetch`es it).
787	sha: String,
788	}
789
790	/// The privileged updater unit instance for `sha`. A git sha is hex-only, so it
791	/// is a safe systemd instance name with no escaping needed.
792	fn self_update_unit(sha: &crate::domain::GitSha) -> String {
793	format!("sando-update@{sha}.service")
794	}
795
796	/// Trigger a rebuild + restart of sandod itself to `sha`. sandod runs
797	/// unprivileged (User=sando, NoNewPrivileges, ProtectSystem=strict) and cannot
798	/// write `/usr/local/bin/sandod` or restart its own service — so it only
799	/// triggers the root oneshot `sando-update@<sha>.service` (which the sando
800	/// user is authorized to start by a scoped polkit rule). That unit builds
801	/// `sando/daemon` as the sando user in a dedicated checkout, installs the new
802	/// binary, and restarts sandod. Bearer-gated like the other mutators; the new
803	/// version shows up in `/state`'s `sandod_version` once the restart lands.
804	async fn self_update(
805	State(s): State<AppState>,
806	crate::error::TypedBody(body): crate::error::TypedBody<SelfUpdateBody>,
807	) -> Result<Json<serde_json::Value>> {
808	let sha = crate::domain::GitSha::parse(&body.sha)
809	.map_err(\|e\| crate::error::Error::BadRequest(format!("invalid sha: {e}")))?;
810
811	let unit = self_update_unit(&sha);
812
813	// Don't restart the controller out from under an in-flight server build — the
814	// restart would SIGKILL it mid-deploy. Hold the active_build slot across the
815	// trigger, not just the check: releasing it before triggering let a /rebuild
816	// claim the slot in the gap and start a build the updater then killed (the
817	// Run 2 TOCTOU). With the guard held, no build can start until the updater is
818	// enqueued.
819	{
820	let slot = s.active_build.lock().await;
821	if slot.as_ref().is_some_and(\|b\| !b.handle.is_finished()) {
822	return Err(crate::error::Error::GateBlocked(
823	"a server build is in flight; retry /self-update once it settles".into(),
824	));
825	}
826	// Also refuse during an in-flight promote/rollback: the updater's restart
827	// would SIGKILL the daemon mid-deploy. try_lock (not lock) so we reject
828	// rather than queue behind a long deploy; the guard is dropped immediately
829	// — it only needs to observe that no deploy holds it right now.
830	if s.deploy_lock.try_lock().is_err() {
831	return Err(crate::error::Error::GateBlocked(
832	"a promote/rollback is in flight; retry /self-update once it settles".into(),
833	));
834	}
835
836	tracing::warn!(sha = %sha, unit = %unit, "self-update requested; triggering privileged updater");
837	// `--no-block`: return as soon as the job is enqueued. The build+restart
838	// outcome lands in `journalctl -u <unit>`; sandod is restarted out from
839	// under this request, so there is nothing more to await here.
840	let status = tokio::process::Command::new("systemctl")
841	.args(["start", "--no-block", &unit])
842	.status()
843	.await
844	.map_err(\|e\| crate::error::Error::Other(anyhow::anyhow!("spawning systemctl: {e}")))?;
845	if !status.success() {
846	return Err(crate::error::Error::Other(anyhow::anyhow!(
847	"systemctl start {unit} exited {status}; is sando-update@.service installed and the sando-user polkit rule in place?"
848	)));
849	}
850	}
851	Ok(Json(
852	serde_json::json!({ "accepted": true, "sha": sha.to_string(), "unit": unit }),
853	))
854	}
855
856	async fn confirm(
857	State(s): State<AppState>,
858	Path(tier): Path<String>,
859	) -> Result<Json<serde_json::Value>> {
860	// Operator-driven satisfaction of a `manual_confirm` gate. Looks up the
861	// pending version (current MM version, or the tier's own if non-mm) and
862	// inserts a passing gate_runs row so /promote can advance.
863	//
864	// Hold deploy_lock so the version we read and the row we insert reference the
865	// same tier state — a concurrent rollback can't change current_version between
866	// the read and the insert (ultra-fuzz Run 2, F3).
867	let _deploy_guard = s.deploy_lock.lock().await;
868	let tier = crate::domain::TierId::new(tier);
869	let target = s
870	.topo
871	.tiers
872	.iter()
873	.find(\|t\| t.name == tier)
874	.ok_or(crate::error::Error::NotFound)?;
875
876	let version_str: Option<String> =
877	sqlx::query_scalar("SELECT current_version FROM tier_state WHERE app = ? AND tier = ?")
878	.bind(&s.cfg.id)
879	.bind(&target.name)
880	.fetch_optional(&s.pool)
881	.await
882	.map_err(crate::error::Error::Db)?
883	.flatten();
884	let version_str = version_str.ok_or_else(\|\| {
885	crate::error::Error::GateBlocked(format!(
886	"tier {tier} has no current_version; nothing to confirm"
887	))
888	})?;
889	let version = crate::domain::Version::parse(&version_str)
890	.map_err(\|e\| crate::error::Error::Other(anyhow::anyhow!(e)))?;
891
892	let now = chrono::Utc::now().to_rfc3339();
893	let outcome =
894	crate::outcome::GateOutcome::passed(crate::outcome::PassNote::OperatorConfirmed {
895	at: chrono::Utc::now(),
896	});
897	let outcome_json = serde_json::to_string(&outcome)
898	.map_err(\|e\| crate::error::Error::Other(anyhow::anyhow!(e)))?;
899	sqlx::query(
900	"INSERT INTO gate_runs (app, version, tier, gate_kind, started_at, finished_at, status, outcome_json)
901	VALUES (?, ?, ?, 'manual_confirm', ?, ?, 'passed', ?)",
902	)
903	.bind(&s.cfg.id)
904	.bind(&version).bind(&target.name).bind(&now).bind(&now).bind(&outcome_json)
905	.execute(&s.pool).await.map_err(crate::error::Error::Db)?;
906
907	tracing::info!(tier = %tier, version = %version, "manual_confirm recorded");
908	crate::events::emit(
909	&s.events,
910	crate::events::Event::ManualConfirm {
911	tier: tier.clone(),
912	version: version.clone(),
913	},
914	);
915
916	Ok(Json(
917	serde_json::json!({ "tier": tier, "version": version }),
918	))
919	}
920
921	#[derive(Deserialize, Default)]
922	struct BackupFetchBody {
923	/// Accept a dump that falls below the plausibility floor, re-baselining it to
924	/// this fetch's size. For the case where the source legitimately shrank and
925	/// the floor has wedged itself; the `gzip -t` integrity check still applies.
926	/// Operator-only — never set this from a timer.
927	#[serde(default)]
928	force: bool,
929	/// Fetch only this configured dump (`[[backup]]` name). Omitted, every
930	/// configured dump is fetched — which is what the daily timer wants, since
931	/// each one gates a different database.
932	#[serde(default)]
933	name: Option<String>,
934	}
935
936	async fn backup_fetch(
937	State(s): State<AppState>,
938	body: Option<Json<BackupFetchBody>>,
939	) -> Result<Json<serde_json::Value>> {
940	let body = body.map(\|Json(b)\| b).unwrap_or_default();
941	let fetched = crate::backup::fetch(&s.pool, &s.cfg, &s.topo, body.force, body.name.as_deref())
942	.await
943	.map_err(crate::error::Error::Other)?;
944	for fb in &fetched {
945	crate::events::emit(
946	&s.events,
947	crate::events::Event::BackupFetched {
948	source: fb.source.clone(),
949	byte_size: fb.byte_size.unwrap_or(0),
950	},
951	);
952	}
953	// `backups` stays an array even for a one-entry topology: a caller that
954	// parses the response should not start failing the day a second database
955	// gets a gate. The top-level keys mirror the first entry for the operator
956	// reading `curl` output, and for the pre-list callers that read them.
957	let first = fetched.first();
958	Ok(Json(serde_json::json!({
959	"source": first.map(\|f\| f.source.clone()),
960	"local_path": first.map(\|f\| f.local_path.clone()),
961	"byte_size": first.and_then(\|f\| f.byte_size),
962	"backups": fetched.iter().map(\|f\| serde_json::json!({
963	"name": f.name,
964	"source": f.source,
965	"local_path": f.local_path,
966	"byte_size": f.byte_size,
967	})).collect::<Vec<_>>(),
968	})))
969	}
970
971	async fn get_gate_log(
972	State(s): State<AppState>,
973	Path((version, gate)): Path<(String, String)>,
974	) -> Result<axum::response::Response> {
975	// Guard against `..` / absolute paths — the version segment must be a single
976	// safe component. Without this, `GET /logs/..%2Fetc/passwd` would escape
977	// logs_root.
978	fn safe(seg: &str) -> bool {
979	!seg.is_empty() && !seg.contains('/') && !seg.contains('\\') && seg != "." && seg != ".."
980	}
981	if !safe(&version) {
982	return Err(crate::error::Error::NotFound);
983	}
984	// The gate segment is an allowlisted kind, not a free-form filename: parse it
985	// to `GateKind` so only the five known `<kind>.log` files are reachable. This
986	// closes `*.log` filename probing within a version dir (traversal was already
987	// blocked; this bounds the basename to the known set).
988	let kind: crate::domain::GateKind = gate.parse().map_err(\|_\| crate::error::Error::NotFound)?;
989	let path = s
990	.cfg
991	.logs_root
992	.join(&version)
993	.join(format!("{}.log", kind.as_str()));
994	// Bound how much a single request can pull into the daemon's memory: a
995	// runaway gate log shouldn't be read whole. Past the cap we return the tail
996	// (the recent, relevant output) behind a truncation marker.
997	const MAX_LOG_BYTES: u64 = 8 * 1024 * 1024;
998	use tokio::io::{AsyncReadExt, AsyncSeekExt};
999	let mut file = match tokio::fs::File::open(&path).await {
1000	Ok(f) => f,
1001	Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
1002	return Err(crate::error::Error::NotFound);
1003	}
1004	Err(e) => return Err(crate::error::Error::Other(e.into())),
1005	};
1006	let len = file
1007	.metadata()
1008	.await
1009	.map_err(\|e\| crate::error::Error::Other(e.into()))?
1010	.len();
1011	let body: Vec<u8> = if len <= MAX_LOG_BYTES {
1012	let mut buf = Vec::with_capacity(len as usize);
1013	file.read_to_end(&mut buf)
1014	.await
1015	.map_err(\|e\| crate::error::Error::Other(e.into()))?;
1016	buf
1017	} else {
1018	file.seek(std::io::SeekFrom::End(-(MAX_LOG_BYTES as i64)))
1019	.await
1020	.map_err(\|e\| crate::error::Error::Other(e.into()))?;
1021	let mut buf = format!(
1022	"[log truncated: showing the last {} MiB of {len} bytes]\n",
1023	MAX_LOG_BYTES / 1024 / 1024,
1024	)
1025	.into_bytes();
1026	file.read_to_end(&mut buf)
1027	.await
1028	.map_err(\|e\| crate::error::Error::Other(e.into()))?;
1029	buf
1030	};
1031	Ok((
1032	[(
1033	axum::http::header::CONTENT_TYPE,
1034	"text/plain; charset=utf-8",
1035	)],
1036	body,
1037	)
1038	.into_response())
1039	}
1040
1041	async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
1042	use axum::extract::ws::Message;
1043	use tokio::sync::broadcast::error::RecvError;
1044
1045	ws.on_upgrade(move \|mut socket\| async move {
1046	// Subscribe to both channels and merge them: a lag on the high-rate log
1047	// stream emits its own `lagged` frame without dropping anything on the
1048	// status stream (and vice versa), so a busy gate's chunk firehose can't
1049	// evict a PromoteComplete/GateDone the operator needs to see.
1050	let mut status_rx = s.events.subscribe_status();
1051	let mut logs_rx = s.events.subscribe_logs();
1052	loop {
1053	let recv = tokio::select! {
1054	r = status_rx.recv() => r,
1055	r = logs_rx.recv() => r,
1056	};
1057	match recv {
1058	Ok(env) => {
1059	let json = match serde_json::to_string(&env) {
1060	Ok(s) => s,
1061	Err(e) => {
1062	tracing::warn!(error = %e, "events ws: serialize failed");
1063	continue;
1064	}
1065	};
1066	if socket.send(Message::Text(json.into())).await.is_err() {
1067	break;
1068	}
1069	}
1070	Err(RecvError::Lagged(n)) => {
1071	let _ = socket
1072	.send(Message::Text(
1073	format!(r#"{{"kind":"lagged","skipped":{n}}}"#).into(),
1074	))
1075	.await;
1076	}
1077	Err(RecvError::Closed) => break,
1078	}
1079	}
1080	})
1081	}
1082
1083	#[cfg(test)]
1084	mod tests {
1085	use super::promotion::{
1086	PromotedBuild, RollbackReport, rollback_deployed_nodes, unsatisfied_gates,
1087	};
1088	use super::*;
1089	use crate::config::AppConfig;
1090	use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology};
1091	use async_trait::async_trait;
1092	use axum::body::Body;
1093	use axum::http::{Request, StatusCode};
1094	use http_body_util::BodyExt;
1095	use ops_exec::{CapabilitySet, Executor, LogSink, RunOutput, Step, SyncOpts};
1096	use sqlx::SqlitePool;
1097	use sqlx::sqlite::SqlitePoolOptions;
1098	use std::os::unix::process::ExitStatusExt;
1099	use std::path::PathBuf;
1100	use std::sync::{Arc, Mutex as StdMutex};
1101	use tower::ServiceExt;
1102
1103	async fn fresh_pool() -> SqlitePool {
1104	let pool = SqlitePoolOptions::new()
1105	.max_connections(1)
1106	.connect("sqlite::memory:")
1107	.await
1108	.unwrap();
1109	sqlx::migrate!("./migrations").run(&pool).await.unwrap();
1110	pool
1111	}
1112
1113	/// Two-tier topology used by the route tests: mm (provisioned, no nodes)
1114	/// → a (provisioned, one local node). Mirrors the production shape
1115	/// without involving real ssh / postgres.
1116	fn test_topo() -> Topology {
1117	Topology {
1118	repo: Some(RepoConfig {
1119	bare_path: "/tmp/test.git".into(),
1120	branch: "main".into(),
1121	upstream: None,
1122	}),
1123	backup: vec![BackupConfig {
1124	name: "server".into(),
1125	source: "file:///tmp/test-backup.sql".into(),
1126	local_path: "/tmp/local-backup.sql".into(),
1127	}],
1128	tiers: vec![
1129	Tier {
1130	public_url: None,
1131	name: "host".into(),
1132	provisioned: true,
1133	gates: vec![],
1134	canary: CanaryPolicy::Sequential,
1135	nodes: vec![],
1136	},
1137	Tier {
1138	public_url: None,
1139	name: "a".into(),
1140	provisioned: true,
1141	gates: vec![Gate::BootSmoke],
1142	canary: CanaryPolicy::Sequential,
1143	nodes: vec![Node {
1144	platform: None,
1145	name: "a-local".into(),
1146	ssh_target: "local".into(),
1147	release_root: "/tmp/a-node".into(),
1148	service_name: "makenotwork.service".into(),
1149	health_url: None,
1150	config_check_env_file: None,
1151	actuate: crate::topology::default_actuate(),
1152	observe: crate::topology::default_observe(),
1153	companions: Vec::new(),
1154	}],
1155	},
1156	],
1157	aux_repos: Vec::new(),
1158	}
1159	}
1160
1161	fn test_cfg() -> AppConfig {
1162	AppConfig {
1163	page_smoke_cmd: None,
1164	platform: None,
1165	id: crate::domain::AppId::default(),
1166	topology_path: PathBuf::from("/tmp/test-sando.toml"),
1167	build_host: Some("test-host".into()),
1168	workdir: PathBuf::from("/tmp/sando-work"),
1169	release_root: PathBuf::from("/tmp/sando-releases"),
1170	scratch_db_url: None,
1171	scratch_owner_role: "makenotwork".into(),
1172	boot_smoke_port: 18181,
1173	code_smoke_port: 18182,
1174	bin_names: vec!["makenotwork".into()],
1175	logs_root: PathBuf::from("/tmp/sando-logs"),
1176	release_contents: vec![],
1177	cargo_target_dir: None,
1178	gate_timeout_secs: 2400,
1179	companions: Vec::new(),
1180	test_targets: vec![crate::config::TestTarget {
1181	dir: PathBuf::from("server"),
1182	aux_repo: None,
1183	features: vec!["fast-tests".into()],
1184	all_features: false,
1185	scratch_db: true,
1186	}],
1187	migration_checks: vec![],
1188	frontend_builds: vec![],
1189	backup_max_age_hours: 48,
1190	}
1191	}
1192
1193	/// Two products mounted on one daemon address different state.
1194	///
1195	/// The mount-per-product shape is what makes this true: each router carries
1196	/// its own product's config, so `/apps/pom/state` cannot answer from MNW's
1197	/// tiers even if a handler forgets the product exists. The root mount keeps
1198	/// meaning the default product, which is what the runbook and the TUI call.
1199	#[tokio::test]
1200	async fn each_app_is_addressable_and_the_root_stays_the_default() {
1201	let pool = fresh_pool().await;
1202	// MNW ships host + a; pom ships one tier of its own, named differently
1203	// so the response says which product answered.
1204	for (app, tiers) in [("mnw", vec!["host", "a"]), ("pom", vec!["pom-host"])] {
1205	for (i, name) in tiers.iter().enumerate() {
1206	sqlx::query("INSERT INTO tiers (app, name, ord, provisioned) VALUES (?, ?, ?, 1)")
1207	.bind(app)
1208	.bind(name)
1209	.bind(i as i64)
1210	.execute(&pool)
1211	.await
1212	.unwrap();
1213	sqlx::query("INSERT INTO tier_state (app, tier) VALUES (?, ?)")
1214	.bind(app)
1215	.bind(name)
1216	.execute(&pool)
1217	.await
1218	.unwrap();
1219	}
1220	}
1221
1222	let mnw_topo = Arc::new(test_topo());
1223	let mut pom_topo = test_topo();
1224	pom_topo.tiers = vec![crate::topology::Tier {
1225	public_url: None,
1226	name: "pom-host".into(),
1227	provisioned: true,
1228	gates: vec![],
1229	canary: crate::topology::CanaryPolicy::Sequential,
1230	nodes: vec![],
1231	}];
1232	let pom_topo = Arc::new(pom_topo);
1233
1234	let mnw_cfg = Arc::new(test_cfg());
1235	let mut pom = test_cfg();
1236	pom.id = crate::domain::AppId::new("pom");
1237	let pom_cfg = Arc::new(pom);
1238
1239	let mut apps = crate::state::AppMap::new();
1240	for (id, cfg, topo) in [
1241	(mnw_cfg.id.clone(), mnw_cfg.clone(), mnw_topo.clone()),
1242	(pom_cfg.id.clone(), pom_cfg.clone(), pom_topo.clone()),
1243	] {
1244	let executors = Arc::new(crate::state::build_executors(&topo));
1245	apps.insert(
1246	id,
1247	Arc::new(crate::state::App {
1248	cfg,
1249	topo,
1250	executors,
1251	}),
1252	);
1253	}
1254	let state = AppState {
1255	pool,
1256	apps: Arc::new(apps),
1257	default_app: mnw_cfg.id.clone(),
1258	topo: mnw_topo,
1259	cfg: mnw_cfg,
1260	active_build: Arc::new(tokio::sync::Mutex::new(None)),
1261	deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1262	events: crate::events::channel(),
1263	executors: Arc::new(std::collections::HashMap::new()),
1264	api_token: None,
1265	};
1266
1267	let get = async \|uri: &str\| -> String {
1268	let resp = router_for_apps(state.clone())
1269	.oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
1270	.await
1271	.unwrap();
1272	assert_eq!(resp.status(), StatusCode::OK, "GET {uri}");
1273	body_string(resp).await
1274	};
1275
1276	// The root is the default product.
1277	let root = get("/state").await;
1278	assert!(root.contains("\"host\""), "root /state: {root}");
1279	assert!(!root.contains("pom-host"), "root must not show pom: {root}");
1280
1281	// Each product answers under its own mount.
1282	let mnw = get("/apps/mnw/state").await;
1283	assert_eq!(mnw, root, "the default mount and the root are one product");
1284	let pom = get("/apps/pom/state").await;
1285	assert!(pom.contains("pom-host"), "/apps/pom/state: {pom}");
1286	assert!(
1287	!pom.contains("\"host\""),
1288	"pom must not see mnw's tiers: {pom}"
1289	);
1290
1291	// And the index says what is mounted.
1292	let index = get("/apps").await;
1293	assert!(
1294	index.contains("\"mnw\"") && index.contains("\"pom\""),
1295	"{index}"
1296	);
1297	assert!(index.contains("\"default_app\":\"mnw\""), "{index}");
1298
1299	// An unconfigured product is not a route.
1300	let resp = router_for_apps(state.clone())
1301	.oneshot(
1302	Request::builder()
1303	.uri("/apps/nope/state")
1304	.body(Body::empty())
1305	.unwrap(),
1306	)
1307	.await
1308	.unwrap();
1309	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1310	}
1311
1312	async fn test_state() -> AppState {
1313	let pool = fresh_pool().await;
1314	// Seed tier rows so FKs on tier_state / gate_runs are satisfied.
1315	for (i, name) in ["host", "a"].iter().enumerate() {
1316	sqlx::query(
1317	"INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
1318	)
1319	.bind(name)
1320	.bind(i as i64)
1321	.execute(&pool)
1322	.await
1323	.unwrap();
1324	sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
1325	.bind(name)
1326	.execute(&pool)
1327	.await
1328	.unwrap();
1329	}
1330	// Don't call install_recorder in tests — it touches a process-global
1331	// and conflicts when tests run in parallel.
1332	let topo = test_topo();
1333	let executors = Arc::new(crate::state::build_executors(&topo));
1334	let topo = Arc::new(topo);
1335	let cfg = Arc::new(test_cfg());
1336	let (apps, default_app) =
1337	crate::state::one_app(cfg.clone(), topo.clone(), executors.clone());
1338	AppState {
1339	pool,
1340	apps,
1341	default_app,
1342	topo,
1343	cfg,
1344	active_build: Arc::new(tokio::sync::Mutex::new(None)),
1345	deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1346	events: crate::events::channel(),
1347	executors,
1348	api_token: None,
1349	}
1350	}
1351
1352	async fn body_string(resp: axum::response::Response) -> String {
1353	let bytes = resp.into_body().collect().await.unwrap().to_bytes();
1354	String::from_utf8(bytes.to_vec()).unwrap()
1355	}
1356
1357	/// Insert the FK prerequisites for inserting gate_runs/tier_state rows.
1358	async fn seed(pool: &SqlitePool, tier: &str, version: &str) {
1359	sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, 0, 1, 'sequential') ON CONFLICT DO NOTHING")
1360	.bind(tier).execute(pool).await.unwrap();
1361	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
1362	.bind(version).execute(pool).await.unwrap();
1363	sqlx::query("INSERT INTO tier_state (tier, current_version) VALUES (?, NULL) ON CONFLICT DO NOTHING")
1364	.bind(tier).execute(pool).await.unwrap();
1365	}
1366
1367	async fn insert_gate(pool: &SqlitePool, tier: &str, version: &str, kind: &str, passed: i64) {
1368	let status = if passed == 1 { "passed" } else { "failed" };
1369	sqlx::query(
1370	"INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
1371	VALUES (?, ?, ?, datetime('now'), datetime('now'), ?)",
1372	)
1373	.bind(version)
1374	.bind(tier)
1375	.bind(kind)
1376	.bind(status)
1377	.execute(pool)
1378	.await
1379	.unwrap();
1380	}
1381
1382	// ---- unsatisfied_gates ----
1383
1384	fn tid(s: &str) -> crate::domain::TierId {
1385	crate::domain::TierId::new(s)
1386	}
1387
1388	#[tokio::test]
1389	async fn unsatisfied_gates_empty_when_no_configured_gates() {
1390	// A tier that configures no gates has nothing to satisfy.
1391	let pool = fresh_pool().await;
1392	seed(&pool, "host", "0.8.12").await;
1393	let pending = unsatisfied_gates(
1394	&pool,
1395	&crate::domain::AppId::default(),
1396	&tid("host"),
1397	&[],
1398	"0.8.12",
1399	&[],
1400	false,
1401	)
1402	.await
1403	.unwrap();
1404	assert_eq!(pending, Vec::<String>::new());
1405	}
1406
1407	#[tokio::test]
1408	async fn unsatisfied_gates_flags_configured_gate_that_never_ran() {
1409	// THE CF1 FIX: a configured gate with no gate_runs row is unsatisfied
1410	// (fail closed), NOT silently treated as green. Before this, an A tier
1411	// whose boot_smoke never executed exposed zero rows and waved promotion
1412	// straight through to prod.
1413	let pool = fresh_pool().await;
1414	seed(&pool, "a", "0.8.12").await;
1415	let pending = unsatisfied_gates(
1416	&pool,
1417	&crate::domain::AppId::default(),
1418	&tid("a"),
1419	&[Gate::BootSmoke],
1420	"0.8.12",
1421	&[],
1422	false,
1423	)
1424	.await
1425	.unwrap();
1426	assert_eq!(pending, vec!["boot_smoke".to_string()]);
1427	}
1428
1429	#[tokio::test]
1430	async fn unsatisfied_gates_flags_failed_kind() {
1431	let pool = fresh_pool().await;
1432	seed(&pool, "host", "0.8.12").await;
1433	insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1434	insert_gate(&pool, "host", "0.8.12", "boot_smoke", 1).await;
1435	let pending = unsatisfied_gates(
1436	&pool,
1437	&crate::domain::AppId::default(),
1438	&tid("host"),
1439	&[Gate::CargoTest, Gate::BootSmoke],
1440	"0.8.12",
1441	&[],
1442	false,
1443	)
1444	.await
1445	.unwrap();
1446	assert_eq!(pending, vec!["cargo_test".to_string()]);
1447	}
1448
1449	#[tokio::test]
1450	async fn unsatisfied_gates_latest_row_wins() {
1451	// Two runs of the same gate; only the latest counts. A flap from
1452	// red to green should clear the pending entry.
1453	let pool = fresh_pool().await;
1454	seed(&pool, "host", "0.8.12").await;
1455	insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1456	insert_gate(&pool, "host", "0.8.12", "cargo_test", 1).await;
1457	let pending = unsatisfied_gates(
1458	&pool,
1459	&crate::domain::AppId::default(),
1460	&tid("host"),
1461	&[Gate::CargoTest],
1462	"0.8.12",
1463	&[],
1464	false,
1465	)
1466	.await
1467	.unwrap();
1468	assert!(pending.is_empty());
1469	}
1470
1471	async fn insert_confirm(
1472	pool: &SqlitePool,
1473	tier: &str,
1474	version: &str,
1475	at: chrono::DateTime<chrono::Utc>,
1476	) {
1477	sqlx::query(
1478	"INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status) \
1479	VALUES (?, ?, 'manual_confirm', ?, ?, 'passed')",
1480	)
1481	.bind(version)
1482	.bind(tier)
1483	.bind(at.to_rfc3339())
1484	.bind(at.to_rfc3339())
1485	.execute(pool)
1486	.await
1487	.unwrap();
1488	}
1489
1490	#[tokio::test]
1491	async fn unsatisfied_gates_manual_confirm_requires_fresh_confirmation() {
1492	// A confirmation only satisfies the gate if it post-dates the version's
1493	// current landing on the tier (burn_in_started_at). A stale confirm left
1494	// over from before a rollback + rollback-forward must NOT wave it through.
1495	let pool = fresh_pool().await;
1496	seed(&pool, "a", "0.8.12").await;
1497	let landed = chrono::Utc::now();
1498	sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
1499	.bind(landed.to_rfc3339())
1500	.execute(&pool)
1501	.await
1502	.unwrap();
1503
1504	// Stale confirmation (recorded before this landing) -> unsatisfied.
1505	insert_confirm(&pool, "a", "0.8.12", landed - chrono::Duration::hours(1)).await;
1506	let pending = unsatisfied_gates(
1507	&pool,
1508	&crate::domain::AppId::default(),
1509	&tid("a"),
1510	&[Gate::ManualConfirm],
1511	"0.8.12",
1512	&[],
1513	false,
1514	)
1515	.await
1516	.unwrap();
1517	assert_eq!(
1518	pending,
1519	vec!["manual_confirm".to_string()],
1520	"stale confirm must not satisfy"
1521	);
1522
1523	// Fresh confirmation (after this landing) -> satisfied.
1524	insert_confirm(&pool, "a", "0.8.12", landed + chrono::Duration::minutes(5)).await;
1525	let pending = unsatisfied_gates(
1526	&pool,
1527	&crate::domain::AppId::default(),
1528	&tid("a"),
1529	&[Gate::ManualConfirm],
1530	"0.8.12",
1531	&[],
1532	false,
1533	)
1534	.await
1535	.unwrap();
1536	assert!(pending.is_empty(), "fresh confirm satisfies");
1537	}
1538
1539	#[tokio::test]
1540	async fn unsatisfied_gates_manual_confirm_fails_closed_without_baseline() {
1541	// No landing clock (burn_in_started_at NULL) -> a passed confirm row is
1542	// not provably fresh, so fail closed and require a new confirmation.
1543	let pool = fresh_pool().await;
1544	seed(&pool, "a", "0.8.12").await; // leaves burn_in_started_at NULL
1545	insert_confirm(&pool, "a", "0.8.12", chrono::Utc::now()).await;
1546	let pending = unsatisfied_gates(
1547	&pool,
1548	&crate::domain::AppId::default(),
1549	&tid("a"),
1550	&[Gate::ManualConfirm],
1551	"0.8.12",
1552	&[],
1553	false,
1554	)
1555	.await
1556	.unwrap();
1557	assert_eq!(
1558	pending,
1559	vec!["manual_confirm".to_string()],
1560	"no baseline -> fail closed"
1561	);
1562	}
1563
1564	#[tokio::test]
1565	async fn unsatisfied_gates_hotfix_skips_only_burn_in() {
1566	// burn_in is evaluated live (no clock started -> not elapsed); cargo_test
1567	// has a failing row. Normal: both unsatisfied, in configured order.
1568	// hotfix: burn_in suppressed, cargo_test still flagged. Lock the semantic
1569	// so a future change doesn't widen the hotfix bypass.
1570	let pool = fresh_pool().await;
1571	seed(&pool, "a", "0.8.12").await;
1572	insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
1573	let gates = [Gate::BurnIn { hours: 48 }, Gate::CargoTest];
1574
1575	let normal = unsatisfied_gates(
1576	&pool,
1577	&crate::domain::AppId::default(),
1578	&tid("a"),
1579	&gates,
1580	"0.8.12",
1581	&[],
1582	false,
1583	)
1584	.await
1585	.unwrap();
1586	assert_eq!(
1587	normal,
1588	vec!["burn_in".to_string(), "cargo_test".to_string()]
1589	);
1590
1591	let with_hotfix = unsatisfied_gates(
1592	&pool,
1593	&crate::domain::AppId::default(),
1594	&tid("a"),
1595	&gates,
1596	"0.8.12",
1597	&[],
1598	true,
1599	)
1600	.await
1601	.unwrap();
1602	assert_eq!(with_hotfix, vec!["cargo_test".to_string()]);
1603	}
1604
1605	#[tokio::test]
1606	async fn unsatisfied_gates_burn_in_passes_when_window_elapsed() {
1607	// A burn-in clock started far enough in the past satisfies the gate
1608	// live — no gate_runs row needed.
1609	let pool = fresh_pool().await;
1610	seed(&pool, "a", "0.8.12").await;
1611	sqlx::query("UPDATE tier_state SET burn_in_started_at = ? WHERE tier = 'a'")
1612	.bind((chrono::Utc::now() - chrono::Duration::hours(50)).to_rfc3339())
1613	.execute(&pool)
1614	.await
1615	.unwrap();
1616	let pending = unsatisfied_gates(
1617	&pool,
1618	&crate::domain::AppId::default(),
1619	&tid("a"),
1620	&[Gate::BurnIn { hours: 48 }],
1621	"0.8.12",
1622	&[],
1623	false,
1624	)
1625	.await
1626	.unwrap();
1627	assert!(pending.is_empty(), "50h elapsed satisfies a 48h burn-in");
1628	}
1629
1630	#[tokio::test]
1631	async fn unsatisfied_gates_ignores_other_tiers_and_versions() {
1632	let pool = fresh_pool().await;
1633	seed(&pool, "host", "0.8.12").await;
1634	seed(&pool, "host", "0.8.11").await;
1635	seed(&pool, "a", "0.8.12").await;
1636	// Mark host/0.8.12 cargo_test failing, but unrelated tiers/versions
1637	// shouldn't pollute the query.
1638	insert_gate(&pool, "host", "0.8.12", "cargo_test", 0).await;
1639	insert_gate(&pool, "a", "0.8.12", "cargo_test", 0).await;
1640	insert_gate(&pool, "host", "0.8.11", "cargo_test", 0).await;
1641
1642	let pending = unsatisfied_gates(
1643	&pool,
1644	&crate::domain::AppId::default(),
1645	&tid("host"),
1646	&[Gate::CargoTest],
1647	"0.8.12",
1648	&[],
1649	false,
1650	)
1651	.await
1652	.unwrap();
1653	assert_eq!(pending, vec!["cargo_test".to_string()]);
1654	}
1655
1656	#[tokio::test]
1657	async fn unsatisfied_gates_null_status_is_treated_as_failing() {
1658	// An in-flight gate (started_at set, finished_at + status NULL)
1659	// should NOT be treated as green. Otherwise a race could promote
1660	// before the gate concludes.
1661	let pool = fresh_pool().await;
1662	seed(&pool, "host", "0.8.12").await;
1663	sqlx::query(
1664	"INSERT INTO gate_runs (version, tier, gate_kind, started_at) \
1665	VALUES ('0.8.12', 'host', 'cargo_test', datetime('now'))",
1666	)
1667	.execute(&pool)
1668	.await
1669	.unwrap();
1670
1671	let pending = unsatisfied_gates(
1672	&pool,
1673	&crate::domain::AppId::default(),
1674	&tid("host"),
1675	&[Gate::CargoTest],
1676	"0.8.12",
1677	&[],
1678	false,
1679	)
1680	.await
1681	.unwrap();
1682	assert_eq!(pending, vec!["cargo_test".to_string()]);
1683	}
1684
1685	// ---- /confirm/{tier} ----
1686
1687	#[tokio::test]
1688	async fn confirm_rejects_when_tier_has_no_current_version() {
1689	// tier_state.a.current_version is NULL by default. /confirm has
1690	// nothing to confirm against → GateBlocked (400).
1691	let state = test_state().await;
1692	let app = router(state.clone());
1693	let resp = app
1694	.oneshot(
1695	Request::builder()
1696	.method("POST")
1697	.uri("/confirm/a")
1698	.body(Body::empty())
1699	.unwrap(),
1700	)
1701	.await
1702	.unwrap();
1703	assert_eq!(resp.status(), StatusCode::CONFLICT);
1704	let body = body_string(resp).await;
1705	assert!(body.contains("no current_version"), "got: {body}");
1706	}
1707
1708	#[tokio::test]
1709	async fn confirm_accepts_when_current_version_set_and_inserts_row() {
1710	let state = test_state().await;
1711	// Seed a version + advance tier a's state to it.
1712	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
1713	.execute(&state.pool).await.unwrap();
1714	sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'a'")
1715	.execute(&state.pool)
1716	.await
1717	.unwrap();
1718
1719	let app = router(state.clone());
1720	let resp = app
1721	.oneshot(
1722	Request::builder()
1723	.method("POST")
1724	.uri("/confirm/a")
1725	.body(Body::empty())
1726	.unwrap(),
1727	)
1728	.await
1729	.unwrap();
1730	assert_eq!(resp.status(), StatusCode::OK);
1731	let body = body_string(resp).await;
1732	assert!(body.contains("\"tier\":\"a\""));
1733	assert!(body.contains("\"version\":\"0.8.12\""));
1734
1735	// A passing gate_runs row was inserted.
1736	let count: (i64,) = sqlx::query_as(
1737	"SELECT COUNT(*) FROM gate_runs WHERE tier='a' AND gate_kind='manual_confirm' AND status='passed'",
1738	)
1739	.fetch_one(&state.pool)
1740	.await
1741	.unwrap();
1742	assert_eq!(count.0, 1);
1743	}
1744
1745	#[tokio::test]
1746	async fn confirm_404s_for_unknown_tier() {
1747	let state = test_state().await;
1748	let app = router(state);
1749	let resp = app
1750	.oneshot(
1751	Request::builder()
1752	.method("POST")
1753	.uri("/confirm/zzzz")
1754	.body(Body::empty())
1755	.unwrap(),
1756	)
1757	.await
1758	.unwrap();
1759	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1760	}
1761
1762	#[tokio::test]
1763	async fn get_run_404s_for_unknown_id() {
1764	let state = test_state().await;
1765	let app = router(state);
1766	let resp = app
1767	.oneshot(
1768	Request::builder()
1769	.uri("/runs/999")
1770	.body(Body::empty())
1771	.unwrap(),
1772	)
1773	.await
1774	.unwrap();
1775	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1776	}
1777
1778	#[tokio::test]
1779	async fn get_run_returns_view_with_gates() {
1780	let state = test_state().await;
1781	// A run that reached version 0.10.2 and ran two host gates (one red).
1782	let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def")
1783	.await
1784	.unwrap();
1785	let ver: crate::domain::Version = "0.10.2".parse().unwrap();
1786	seed(&state.pool, "host", "0.10.2").await;
1787	crate::runs::set_version(&state.pool, run_id, &ver)
1788	.await
1789	.unwrap();
1790	insert_gate(&state.pool, "host", "0.10.2", "cargo_test", 0).await;
1791	insert_gate(&state.pool, "host", "0.10.2", "boot_smoke", 1).await;
1792
1793	let app = router(state);
1794	let resp = app
1795	.oneshot(
1796	Request::builder()
1797	.uri(format!("/runs/{}", run_id.0))
1798	.body(Body::empty())
1799	.unwrap(),
1800	)
1801	.await
1802	.unwrap();
1803	assert_eq!(resp.status(), StatusCode::OK);
1804	let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1805	assert_eq!(v["run_id"], run_id.0);
1806	assert_eq!(v["sha"], "abc1234def");
1807	assert_eq!(v["version"], "0.10.2");
1808	assert_eq!(v["result"], "building");
1809	// Both host gates surface, latest-per-kind, alphabetized by kind.
1810	assert_eq!(v["gates"].as_array().unwrap().len(), 2);
1811	assert_eq!(v["gates"][0]["kind"], "boot_smoke");
1812	assert_eq!(v["gates"][0]["status"], "passed");
1813	assert_eq!(v["gates"][1]["kind"], "cargo_test");
1814	assert_eq!(v["gates"][1]["status"], "failed");
1815	}
1816
1817	#[tokio::test]
1818	async fn get_run_wait_returns_immediately_when_settled() {
1819	let state = test_state().await;
1820	let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def")
1821	.await
1822	.unwrap();
1823	crate::runs::mark_passed(&state.pool, run_id).await.unwrap();
1824
1825	let app = router(state);
1826	// Generous timeout, but an already-settled run must not wait for it.
1827	let resp = app
1828	.oneshot(
1829	Request::builder()
1830	.uri(format!("/runs/{}/wait?timeout_ms=60000", run_id.0))
1831	.body(Body::empty())
1832	.unwrap(),
1833	)
1834	.await
1835	.unwrap();
1836	assert_eq!(resp.status(), StatusCode::OK);
1837	let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1838	assert_eq!(v["result"], "passed");
1839	}
1840
1841	#[tokio::test]
1842	async fn get_run_wait_returns_building_at_timeout() {
1843	let state = test_state().await;
1844	let run_id = crate::runs::create(&state.pool, &state.cfg.id, "abc1234def")
1845	.await
1846	.unwrap();
1847
1848	let app = router(state);
1849	// timeout_ms=0 → deadline is now → the first poll returns the
1850	// still-building run rather than blocking.
1851	let resp = app
1852	.oneshot(
1853	Request::builder()
1854	.uri(format!("/runs/{}/wait?timeout_ms=0", run_id.0))
1855	.body(Body::empty())
1856	.unwrap(),
1857	)
1858	.await
1859	.unwrap();
1860	assert_eq!(resp.status(), StatusCode::OK);
1861	let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
1862	assert_eq!(v["result"], "building");
1863	}
1864
1865	#[tokio::test]
1866	async fn get_run_wait_404s_for_unknown_id() {
1867	let state = test_state().await;
1868	let app = router(state);
1869	let resp = app
1870	.oneshot(
1871	Request::builder()
1872	.uri("/runs/999/wait?timeout_ms=0")
1873	.body(Body::empty())
1874	.unwrap(),
1875	)
1876	.await
1877	.unwrap();
1878	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
1879	}
1880
1881	#[test]
1882	fn self_update_unit_maps_sha_to_instance() {
1883	let sha = crate::domain::GitSha::parse("abc1234def5678").unwrap();
1884	assert_eq!(
1885	self_update_unit(&sha),
1886	"sando-update@abc1234def5678.service"
1887	);
1888	}
1889
1890	#[tokio::test]
1891	async fn self_update_rejects_bad_sha_with_400() {
1892	// A malformed sha is a client error and must be rejected before any
1893	// privileged unit is triggered (so this test never shells out).
1894	let state = test_state().await;
1895	let app = router(state);
1896	let resp = app
1897	.oneshot(
1898	Request::builder()
1899	.method("POST")
1900	.uri("/self-update")
1901	.header("Content-Type", "application/json")
1902	.body(Body::from(r#"{"sha":"not-a-sha!"}"#))
1903	.unwrap(),
1904	)
1905	.await
1906	.unwrap();
1907	assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1908	}
1909
1910	// ---- /promote/{tier} default-version resolution ----
1911
1912	#[tokio::test]
1913	async fn promote_to_first_tier_is_rejected() {
1914	// tier 0 is host — you /rebuild, not /promote.
1915	let state = test_state().await;
1916	let app = router(state);
1917	let resp = app
1918	.oneshot(
1919	Request::builder()
1920	.method("POST")
1921	.uri("/promote/host")
1922	.body(Body::empty())
1923	.unwrap(),
1924	)
1925	.await
1926	.unwrap();
1927	assert_eq!(resp.status(), StatusCode::CONFLICT);
1928	let body = body_string(resp).await;
1929	assert!(
1930	body.contains("cannot /promote to the first tier"),
1931	"got: {body}"
1932	);
1933	}
1934
1935	#[tokio::test]
1936	async fn promote_without_body_and_no_predecessor_version_errors() {
1937	// tier a has no body version supplied AND its predecessor mm has
1938	// current_version=NULL. Should fail before any deploy.
1939	let state = test_state().await;
1940	let app = router(state);
1941	let resp = app
1942	.oneshot(
1943	Request::builder()
1944	.method("POST")
1945	.uri("/promote/a")
1946	.body(Body::empty())
1947	.unwrap(),
1948	)
1949	.await
1950	.unwrap();
1951	assert_eq!(resp.status(), StatusCode::CONFLICT);
1952	let body = body_string(resp).await;
1953	assert!(
1954	body.contains("no version specified") \|\| body.contains("no current_version"),
1955	"got: {body}"
1956	);
1957	}
1958
1959	#[tokio::test]
1960	async fn promote_blocked_when_predecessor_gate_never_ran() {
1961	// End-to-end CF1: the host tier configures boot_smoke but it never ran
1962	// (no gate_runs row). Promoting host -> a must be GateBlocked, citing the
1963	// unsatisfied gate, instead of waving through on zero evidence. A real
1964	// `versions` row is present so the ONLY thing that can block is the gate.
1965	let pool = fresh_pool().await;
1966	for (i, name) in ["host", "a"].iter().enumerate() {
1967	sqlx::query(
1968	"INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
1969	)
1970	.bind(name)
1971	.bind(i as i64)
1972	.execute(&pool)
1973	.await
1974	.unwrap();
1975	sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
1976	.bind(name)
1977	.execute(&pool)
1978	.await
1979	.unwrap();
1980	}
1981	let mut topo = test_topo();
1982	topo.tiers[0].gates = vec![Gate::BootSmoke]; // host configures a gate...
1983	let executors = Arc::new(crate::state::build_executors(&topo));
1984	let topo = Arc::new(topo);
1985	let cfg = Arc::new(test_cfg());
1986	let (apps, default_app) =
1987	crate::state::one_app(cfg.clone(), topo.clone(), executors.clone());
1988	let state = AppState {
1989	pool,
1990	apps,
1991	default_app,
1992	topo,
1993	cfg,
1994	active_build: Arc::new(tokio::sync::Mutex::new(None)),
1995	deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
1996	events: crate::events::channel(),
1997	executors,
1998	api_token: None,
1999	};
2000	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
2001	.execute(&state.pool).await.unwrap();
2002	sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
2003	.execute(&state.pool)
2004	.await
2005	.unwrap();
2006
2007	let app = router(state);
2008	let resp = app
2009	.oneshot(
2010	Request::builder()
2011	.method("POST")
2012	.uri("/promote/a")
2013	.body(Body::empty())
2014	.unwrap(),
2015	)
2016	.await
2017	.unwrap();
2018	assert_eq!(resp.status(), StatusCode::CONFLICT);
2019	let body = body_string(resp).await;
2020	assert!(
2021	body.contains("boot_smoke"),
2022	"expected boot_smoke to block; got: {body}"
2023	);
2024	}
2025
2026	#[tokio::test]
2027	async fn migration_bearing_promote_requires_fresh_confirm() {
2028	// The predecessor (host) configures NO gates, so nothing would normally
2029	// block host -> a. A `bears_migration` promote must still be blocked on a
2030	// fresh `manual_confirm` it does not have: rollback restores the binary
2031	// only, so the one-way advance needs a conscious operator sign-off.
2032	let pool = fresh_pool().await;
2033	for (i, name) in ["host", "a"].iter().enumerate() {
2034	sqlx::query(
2035	"INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, ?, 1, 'sequential')",
2036	)
2037	.bind(name)
2038	.bind(i as i64)
2039	.execute(&pool)
2040	.await
2041	.unwrap();
2042	sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
2043	.bind(name)
2044	.execute(&pool)
2045	.await
2046	.unwrap();
2047	}
2048	let mut topo = test_topo();
2049	topo.tiers[0].gates = vec![]; // host configures no gates at all
2050	let executors = Arc::new(crate::state::build_executors(&topo));
2051	let topo = Arc::new(topo);
2052	let cfg = Arc::new(test_cfg());
2053	let (apps, default_app) =
2054	crate::state::one_app(cfg.clone(), topo.clone(), executors.clone());
2055	let state = AppState {
2056	pool,
2057	apps,
2058	default_app,
2059	topo,
2060	cfg,
2061	active_build: Arc::new(tokio::sync::Mutex::new(None)),
2062	deploy_lock: Arc::new(tokio::sync::Mutex::new(())),
2063	events: crate::events::channel(),
2064	executors,
2065	api_token: None,
2066	};
2067	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.8.12','sha',datetime('now'),'/tmp/x')")
2068	.execute(&state.pool).await.unwrap();
2069	sqlx::query("UPDATE tier_state SET current_version = '0.8.12' WHERE tier = 'host'")
2070	.execute(&state.pool)
2071	.await
2072	.unwrap();
2073
2074	let app = router(state);
2075	let resp = app
2076	.oneshot(
2077	Request::builder()
2078	.method("POST")
2079	.uri("/promote/a")
2080	.header("content-type", "application/json")
2081	.body(Body::from(r#"{"bears_migration": true}"#))
2082	.unwrap(),
2083	)
2084	.await
2085	.unwrap();
2086	assert_eq!(resp.status(), StatusCode::CONFLICT);
2087	let body = body_string(resp).await;
2088	assert!(
2089	body.contains("manual_confirm"),
2090	"expected the migration promote to block on manual_confirm; got: {body}"
2091	);
2092	}
2093
2094	#[tokio::test]
2095	async fn tier_state_advance_is_atomic_previous_from_old_current() {
2096	// CF3: the promote advance is a single UPDATE where previous_version is
2097	// set from the row's old current_version (SQLite evaluates RHS against
2098	// the original row). No read-modify-write to lose under concurrency.
2099	let pool = fresh_pool().await;
2100	seed(&pool, "a", "1.0.0").await;
2101	// current_version FKs into versions, so the target must exist too.
2102	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('2.0.0','sha',datetime('now'),'/tmp/x')")
2103	.execute(&pool).await.unwrap();
2104	sqlx::query("UPDATE tier_state SET current_version = '1.0.0' WHERE tier = 'a'")
2105	.execute(&pool)
2106	.await
2107	.unwrap();
2108
2109	// Exercise the sealed forward-advance primitive itself — the same op
2110	// /promote and the host build path both call (S1), not a copy of its SQL.
2111	let v = crate::domain::Version::parse("2.0.0").unwrap();
2112	crate::runs::advance_tier(&pool, &crate::domain::AppId::default(), "a", &v, None)
2113	.await
2114	.unwrap();
2115
2116	let (cur, prev): (Option<String>, Option<String>) = sqlx::query_as(
2117	"SELECT current_version, previous_version FROM tier_state WHERE tier = 'a'",
2118	)
2119	.fetch_one(&pool)
2120	.await
2121	.unwrap();
2122	assert_eq!(cur.as_deref(), Some("2.0.0"));
2123	assert_eq!(
2124	prev.as_deref(),
2125	Some("1.0.0"),
2126	"previous = the pre-update current, atomically"
2127	);
2128	}
2129
2130	#[tokio::test]
2131	async fn canary_rollback_restores_deployed_nodes_to_previous_version() {
2132	use crate::topology::{Node, default_actuate, default_observe};
2133	let tmp = tempfile::tempdir().unwrap();
2134
2135	// Two local nodes, each pre-seeded as if a promote had flipped them to
2136	// 2.0.0 (current -> releases/2.0.0), with the prior 1.0.0 still on disk.
2137	let mut nodes = Vec::new();
2138	for name in ["n1", "n2"] {
2139	let rr = tmp.path().join(name);
2140	for v in ["1.0.0", "2.0.0"] {
2141	tokio::fs::create_dir_all(rr.join("releases").join(v))
2142	.await
2143	.unwrap();
2144	}
2145	tokio::fs::symlink("releases/2.0.0", rr.join("current"))
2146	.await
2147	.unwrap();
2148	nodes.push(Node {
2149	platform: None,
2150	name: name.into(),
2151	ssh_target: "local".into(),
2152	release_root: rr.to_string_lossy().into_owned(),
2153	service_name: "x.service".into(),
2154	health_url: None,
2155	config_check_env_file: None,
2156	actuate: default_actuate(),
2157	observe: default_observe(),
2158	companions: Vec::new(),
2159	});
2160	}
2161
2162	let mut state = test_state().await;
2163	// The rollback target needs a versions row. The release dir name comes
2164	// from the artifact_path's parent (legacy layout: releases/<version>).
2165	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('1.0.0','sha',datetime('now'),'/tmp/staged/releases/1.0.0/makenotwork')")
2166	.execute(&state.pool).await.unwrap();
2167	let execs: crate::state::ExecutorMap = nodes
2168	.iter()
2169	.map(\|n\| (n.name.clone(), crate::state::build_executor(n)))
2170	.collect();
2171	state.executors = std::sync::Arc::new(execs);
2172
2173	let refs: Vec<&Node> = nodes.iter().collect();
2174	let report = rollback_deployed_nodes(&state, &tid("a"), &refs, "1.0.0").await;
2175	assert_eq!(report.restored, 2, "both deployed nodes should be restored");
2176	assert!(
2177	report.is_consistent(),
2178	"nothing should be indeterminate: {report:?}"
2179	);
2180
2181	for n in &nodes {
2182	let cur = tokio::fs::read_link(std::path::Path::new(&n.release_root).join("current"))
2183	.await
2184	.unwrap();
2185	assert_eq!(
2186	cur.to_string_lossy(),
2187	"releases/1.0.0",
2188	"node {} rolled back",
2189	n.name
2190	);
2191	}
2192	}
2193
2194	#[tokio::test]
2195	async fn canary_rollback_is_noop_without_a_previous_artifact() {
2196	use crate::topology::{Node, default_actuate, default_observe};
2197	let tmp = tempfile::tempdir().unwrap();
2198	let rr = tmp.path().join("n1");
2199	tokio::fs::create_dir_all(&rr).await.unwrap();
2200	let node = Node {
2201	platform: None,
2202	name: "n1".into(),
2203	ssh_target: "local".into(),
2204	release_root: rr.to_string_lossy().into_owned(),
2205	service_name: "x.service".into(),
2206	health_url: None,
2207	config_check_env_file: None,
2208	actuate: default_actuate(),
2209	observe: default_observe(),
2210	companions: Vec::new(),
2211	};
2212	let state = test_state().await; // no versions row for "9.9.9"
2213	let report = rollback_deployed_nodes(&state, &tid("a"), &[&node], "9.9.9").await;
2214	assert_eq!(
2215	report.restored, 0,
2216	"no artifact to roll back to -> nothing restored, no panic"
2217	);
2218	assert_eq!(
2219	report.touched(),
2220	1,
2221	"the node must be accounted for somewhere"
2222	);
2223	// No rollback could even be attempted, so the node's version is not
2224	// knowable here. That must read as indeterminate, not as safe.
2225	assert_eq!(report.indeterminate, 1);
2226	assert!(!report.is_consistent());
2227	}
2228
2229	/// The 2026-08-01 prod incident, as a unit test on the reporting shape.
2230	///
2231	/// A rollback that fails before the symlink swap leaves the node on the
2232	/// version it was already running — the one being rolled back to. The old
2233	/// code counted that as "not restored" and reported it as `restored=0 of=1`
2234	/// plus "it remains on the new version — manual intervention needed", which
2235	/// sent an operator to inspect a production box that was entirely fine.
2236	#[test]
2237	fn a_rollback_that_failed_before_the_swap_is_not_an_incident() {
2238	let report = RollbackReport {
2239	restored: 0,
2240	already_on_previous: 1,
2241	indeterminate: 0,
2242	};
2243	assert_eq!(report.touched(), 1);
2244	assert!(
2245	report.is_consistent(),
2246	"a node that never left the previous version is not split-brain"
2247	);
2248
2249	// Contrast: the same zero restored, but the swap had run. This one does
2250	// warrant a human, and the two must not report the same way.
2251	let real = RollbackReport {
2252	restored: 0,
2253	already_on_previous: 0,
2254	indeterminate: 1,
2255	};
2256	assert_eq!(real.touched(), 1);
2257	assert!(!real.is_consistent());
2258	}
2259
2260	/// A genuine split-brain still reports as one: some nodes back on the old
2261	/// version, one stranded.
2262	#[test]
2263	fn a_mixed_outcome_is_inconsistent_if_any_node_is_unknown() {
2264	let report = RollbackReport {
2265	restored: 2,
2266	already_on_previous: 1,
2267	indeterminate: 1,
2268	};
2269	assert_eq!(report.touched(), 4);
2270	assert!(!report.is_consistent());
2271	}
2272
2273	// ---- FleetFake: a multi-node promote across recorded fake executors ----
2274	//
2275	// The route-level promote tests until now ran a single local node against a
2276	// real LocalExec, so the sequential-canary fan-out, the cross-node deploy
2277	// ordering, and the mid-canary rollback of already-flipped nodes had no
2278	// coverage. FleetFake records every deploy op (tagged by node) into one
2279	// shared log so ordering is visible across the fleet, and can fail any op
2280	// whose shell script or rsync target contains a marker — used to fail a
2281	// node's forward deploy of the new version while its rollback to the prior
2282	// version (a different `releases/<v>` path) still succeeds.
2283
2284	struct FleetFake {
2285	tag: String,
2286	caps: CapabilitySet,
2287	log: Arc<StdMutex<Vec<String>>>,
2288	fail_if_contains: Option<String>,
2289	}
2290
2291	impl FleetFake {
2292	fn fails(&self, text: &str) -> bool {
2293	self.fail_if_contains
2294	.as_deref()
2295	.is_some_and(\|m\| text.contains(m))
2296	}
2297	}
2298
2299	#[async_trait]
2300	impl Executor for FleetFake {
2301	async fn run_streaming(
2302	&self,
2303	step: &Step,
2304	_sink: &mut dyn LogSink,
2305	) -> anyhow::Result<RunOutput> {
2306	let script = step.argv.last().cloned().unwrap_or_default();
2307	self.log
2308	.lock()
2309	.unwrap()
2310	.push(format!("{}:run:{script}", self.tag));
2311	Ok(RunOutput {
2312	status: std::process::ExitStatus::from_raw(if self.fails(&script) {
2313	1 << 8
2314	} else {
2315	0
2316	}),
2317	stdout: Vec::new(),
2318	stderr: Vec::new(),
2319	})
2320	}
2321	async fn pull_file(
2322	&self,
2323	_r: &std::path::Path,
2324	_l: &std::path::Path,
2325	_o: &SyncOpts,
2326	) -> anyhow::Result<()> {
2327	Ok(())
2328	}
2329	async fn pull_dir(
2330	&self,
2331	_r: &std::path::Path,
2332	_l: &std::path::Path,
2333	_o: &SyncOpts,
2334	) -> anyhow::Result<()> {
2335	Ok(())
2336	}
2337	async fn pull_glob(
2338	&self,
2339	_g: &str,
2340	_l: &std::path::Path,
2341	_o: &SyncOpts,
2342	) -> anyhow::Result<()> {
2343	Ok(())
2344	}
2345	async fn push_dir(
2346	&self,
2347	_local: &std::path::Path,
2348	remote: &std::path::Path,
2349	_o: &SyncOpts,
2350	) -> anyhow::Result<()> {
2351	let dst = remote.display().to_string();
2352	self.log
2353	.lock()
2354	.unwrap()
2355	.push(format!("{}:push:{dst}", self.tag));
2356	if self.fails(&dst) {
2357	anyhow::bail!("fake rsync failure on {}", self.tag);
2358	}
2359	Ok(())
2360	}
2361	fn capabilities(&self) -> &CapabilitySet {
2362	&self.caps
2363	}
2364	}
2365
2366	/// Rebuild tier "a" with `names` as remote fake nodes sharing one op log,
2367	/// seed the version/tier_state prerequisites for a promote of 3.0.0 up from
2368	/// `host` (tier "a" starts on 2.0.0 with 1.0.0 behind it), and optionally
2369	/// make `fail_node` fail any op containing `fail_marker`. Returns the state
2370	/// and the shared log.
2371	async fn fleet_fixture(
2372	names: &[&str],
2373	fail_node: Option<&str>,
2374	fail_marker: &str,
2375	) -> (AppState, Arc<StdMutex<Vec<String>>>) {
2376	use crate::topology::{default_actuate, default_observe};
2377	let mut state = test_state().await;
2378	let log = Arc::new(StdMutex::new(Vec::<String>::new()));
2379
2380	let nodes: Vec<Node> = names
2381	.iter()
2382	.map(\|name\| Node {
2383	platform: None,
2384	name: (*name).into(),
2385	ssh_target: format!("deploy@{name}"),
2386	release_root: format!("/tmp/fleet/{name}"),
2387	service_name: "makenotwork.service".into(),
2388	health_url: None,
2389	config_check_env_file: None,
2390	actuate: default_actuate(),
2391	observe: default_observe(),
2392	companions: Vec::new(),
2393	})
2394	.collect();
2395
2396	let mut topo = (*state.topo).clone();
2397	topo.tiers[1].nodes = nodes.clone();
2398	// Drop the tier's post-deploy gate: the deploy fan-out is the subject
2399	// here, and node_health would need its own probe wiring.
2400	topo.tiers[1].gates = vec![];
2401	state.topo = Arc::new(topo);
2402
2403	let execs: crate::state::ExecutorMap = nodes
2404	.iter()
2405	.map(\|n\| {
2406	let nm = n.name.to_string();
2407	let fail = fail_node == Some(nm.as_str());
2408	let exec: Arc<dyn Executor> = Arc::new(FleetFake {
2409	tag: nm,
2410	caps: CapabilitySet::from_tokens(["deploy", "restart"], ["health"]),
2411	log: log.clone(),
2412	fail_if_contains: fail.then(\|\| fail_marker.to_string()),
2413	});
2414	(n.name.clone(), exec)
2415	})
2416	.collect();
2417	state.executors = Arc::new(execs);
2418
2419	for n in &nodes {
2420	// deploys.node FKs into `nodes`.
2421	sqlx::query(
2422	"INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES (?, 'a', ?, ?)",
2423	)
2424	.bind(&n.name)
2425	.bind(&n.ssh_target)
2426	.bind(&n.release_root)
2427	.execute(&state.pool)
2428	.await
2429	.unwrap();
2430	}
2431	for v in ["1.0.0", "2.0.0", "3.0.0"] {
2432	// Legacy (pre-identity) artifact_path is `releases/<version>/<bin>`,
2433	// so the release dir the node mirrors is named for the version. The
2434	// fixture reflects that real layout (parent basename == version).
2435	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
2436	.bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
2437	}
2438	sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
2439	.execute(&state.pool)
2440	.await
2441	.unwrap();
2442	sqlx::query(
2443	"UPDATE tier_state SET current_version = '2.0.0', previous_version = '1.0.0' WHERE tier = 'a'",
2444	)
2445	.execute(&state.pool)
2446	.await
2447	.unwrap();
2448	(state, log)
2449	}
2450
2451	/// Index of the first logged op belonging to `node` (panics if the node was
2452	/// never touched — the message names it).
2453	fn first_touch(log: &[String], node: &str) -> usize {
2454	let prefix = format!("{node}:");
2455	log.iter()
2456	.position(\|e\| e.starts_with(&prefix))
2457	.unwrap_or_else(\|\| panic!("node {node:?} was never deployed to; log: {log:#?}"))
2458	}
2459
2460	#[tokio::test]
2461	async fn promote_deploys_every_node_in_tier_order_and_advances() {
2462	let (state, log) = fleet_fixture(&["a1", "a2", "a3"], None, "").await;
2463	let pool = state.pool.clone();
2464
2465	let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
2466	.await
2467	.expect("all nodes deploy, so the promote succeeds");
2468	assert_eq!(
2469	body["nodes_deployed"],
2470	serde_json::json!(["a1", "a2", "a3"]),
2471	"the response names every node the promote reached",
2472	);
2473
2474	let (cur, prev) = tier_versions(&pool, "a").await;
2475	assert_eq!(cur.as_deref(), Some("3.0.0"));
2476	assert_eq!(prev.as_deref(), Some("2.0.0"));
2477
2478	// Sequential canary: a1 is fully touched before a2, a2 before a3.
2479	let log = log.lock().unwrap().clone();
2480	assert!(
2481	first_touch(&log, "a1") < first_touch(&log, "a2")
2482	&& first_touch(&log, "a2") < first_touch(&log, "a3"),
2483	"nodes must deploy in tier order: {log:#?}",
2484	);
2485
2486	// Every node has a green deploy row for the promoted version.
2487	let ok: i64 = sqlx::query_scalar(
2488	"SELECT COUNT(*) FROM deploys WHERE version = '3.0.0' AND outcome = 'ok'",
2489	)
2490	.fetch_one(&pool)
2491	.await
2492	.unwrap();
2493	assert_eq!(ok, 3, "one ok deploy row per node");
2494	}
2495
2496	#[tokio::test]
2497	async fn a_mid_canary_deploy_failure_rolls_touched_nodes_back_and_does_not_advance() {
2498	// a2 fails its forward deploy of 3.0.0; a1 was already flipped, a3 is
2499	// never reached. The touched nodes (a1, a2) roll back to 2.0.0 — their
2500	// rollback ops target `releases/2.0.0`, which the marker does not match —
2501	// and tier_state must NOT advance.
2502	let (state, log) = fleet_fixture(&["a1", "a2", "a3"], Some("a2"), "releases/3.0.0").await;
2503	let pool = state.pool.clone();
2504
2505	let err = promote_inner(state, "a".into(), PromoteBody::default())
2506	.await
2507	.expect_err("a mid-canary deploy failure must fail the promote");
2508	assert!(
2509	matches!(err, crate::error::Error::Other(_)),
2510	"a deploy failure propagates as Other, got: {err:?}",
2511	);
2512
2513	// tier_state untouched: the failure returns before advance_tier.
2514	let (cur, prev) = tier_versions(&pool, "a").await;
2515	assert_eq!(
2516	cur.as_deref(),
2517	Some("2.0.0"),
2518	"a failed rollout must not advance"
2519	);
2520	assert_eq!(prev.as_deref(), Some("1.0.0"));
2521
2522	let log = log.lock().unwrap().clone();
2523	// a3 sits after the failed a2 in the sequence and is never touched.
2524	assert!(
2525	!log.iter().any(\|e\| e.starts_with("a3:")),
2526	"nodes after the failure must not be deployed to: {log:#?}",
2527	);
2528	// Both touched nodes were rolled back to the prior version.
2529	for n in ["a1", "a2"] {
2530	assert!(
2531	log.iter()
2532	.any(\|e\| e.starts_with(&format!("{n}:")) && e.contains("releases/2.0.0")),
2533	"touched node {n} must be restored to 2.0.0: {log:#?}",
2534	);
2535	}
2536
2537	// The forward attempt is on the record: a1 ok, a2 failed.
2538	let a1: String = sqlx::query_scalar(
2539	"SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a1'",
2540	)
2541	.fetch_one(&pool)
2542	.await
2543	.unwrap();
2544	assert_eq!(a1, "ok");
2545	let a2: String = sqlx::query_scalar(
2546	"SELECT outcome FROM deploys WHERE version = '3.0.0' AND node = 'a2'",
2547	)
2548	.fetch_one(&pool)
2549	.await
2550	.unwrap();
2551	assert_eq!(a2, "failed");
2552
2553	// Both touched nodes restored => the tier is consistent on 2.0.0, so the
2554	// partial flag is cleared, not set.
2555	let reason: Option<String> =
2556	sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2557	.fetch_one(&pool)
2558	.await
2559	.unwrap();
2560	assert_eq!(
2561	reason, None,
2562	"a fully-restored canary leaves the tier consistent, not partial",
2563	);
2564	}
2565
2566	/// Build a one-node tier "a" on a tempdir release root, pre-seeded as if a
2567	/// promote had flipped it to `current` with `prev` still staged on disk.
2568	/// Returns the state (topology rewired to the tempdir node) and the tempdir,
2569	/// which the caller must keep alive.
2570	async fn rollback_fixture(prev: &str, current: &str) -> (AppState, tempfile::TempDir) {
2571	use crate::topology::{Node, default_actuate, default_observe};
2572	let tmp = tempfile::tempdir().unwrap();
2573	let rr = tmp.path().join("a-local");
2574	for v in [prev, current] {
2575	tokio::fs::create_dir_all(rr.join("releases").join(v))
2576	.await
2577	.unwrap();
2578	}
2579	tokio::fs::symlink(format!("releases/{current}"), rr.join("current"))
2580	.await
2581	.unwrap();
2582	let node = Node {
2583	platform: None,
2584	name: "a-local".into(),
2585	ssh_target: "local".into(),
2586	release_root: rr.to_string_lossy().into_owned(),
2587	service_name: "x.service".into(),
2588	health_url: None,
2589	config_check_env_file: None,
2590	actuate: default_actuate(),
2591	observe: default_observe(),
2592	companions: Vec::new(),
2593	};
2594
2595	let mut state = test_state().await;
2596	let mut topo = (*state.topo).clone();
2597	topo.tiers[1].nodes = vec![node];
2598	state.executors = Arc::new(crate::state::build_executors(&topo));
2599	state.topo = Arc::new(topo);
2600
2601	// deploys.node FKs into `nodes`, so the promote path needs the row.
2602	sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
2603	.bind(rr.to_string_lossy().into_owned())
2604	.execute(&state.pool).await.unwrap();
2605	for v in [prev, current] {
2606	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), ?)")
2607	.bind(v).bind(format!("/tmp/staged/releases/{v}/makenotwork")).execute(&state.pool).await.unwrap();
2608	}
2609	sqlx::query(
2610	"UPDATE tier_state SET current_version = ?, previous_version = ? WHERE tier = 'a'",
2611	)
2612	.bind(current)
2613	.bind(prev)
2614	.execute(&state.pool)
2615	.await
2616	.unwrap();
2617	(state, tmp)
2618	}
2619
2620	async fn tier_versions(pool: &SqlitePool, tier: &str) -> (Option<String>, Option<String>) {
2621	sqlx::query_as("SELECT current_version, previous_version FROM tier_state WHERE tier = ?")
2622	.bind(tier)
2623	.fetch_one(pool)
2624	.await
2625	.unwrap()
2626	}
2627
2628	#[tokio::test]
2629	async fn rollback_clears_previous_version_rather_than_swapping_it() {
2630	// The swap bug: writing the version we just rolled OFF into
2631	// previous_version made a second /rollback roll FORWARD onto the broken
2632	// build the operator was escaping. previous_version must go NULL.
2633	let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2634	let pool = state.pool.clone();
2635
2636	let _ = rollback(State(state), Path("a".to_string())).await.unwrap();
2637
2638	let (cur, prev) = tier_versions(&pool, "a").await;
2639	assert_eq!(
2640	cur.as_deref(),
2641	Some("1.0.0"),
2642	"rolled back to the previous version"
2643	);
2644	assert_eq!(
2645	prev, None,
2646	"the version we rolled off must NOT become the rollback target"
2647	);
2648	}
2649
2650	#[tokio::test]
2651	async fn second_rollback_refuses_instead_of_rolling_forward() {
2652	let (state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2653	let pool = state.pool.clone();
2654	let release_root = state.topo.tiers[1].nodes[0].release_root.clone();
2655
2656	let _ = rollback(State(state.clone()), Path("a".to_string()))
2657	.await
2658	.unwrap();
2659	let err = rollback(State(state), Path("a".to_string()))
2660	.await
2661	.expect_err("only one step of history is tracked; a second rollback must refuse");
2662	assert!(
2663	matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("no previous_version")),
2664	"expected a loud refusal, got: {err:?}",
2665	);
2666
2667	// The refusal is total: neither the DB nor the node moved back to 2.0.0.
2668	let (cur, _) = tier_versions(&pool, "a").await;
2669	assert_eq!(cur.as_deref(), Some("1.0.0"));
2670	let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
2671	.await
2672	.unwrap();
2673	assert_eq!(
2674	link.to_string_lossy(),
2675	"releases/1.0.0",
2676	"node stays on the rolled-back version"
2677	);
2678	}
2679
2680	#[tokio::test]
2681	async fn promote_refuses_an_unprovisioned_tier() {
2682	// Every step of a promote to a node-less tier is a silent no-op that
2683	// still reports success: the deploy loop iterates nothing and
2684	// advance_tier records a current_version the tier never received.
2685	let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2686	let mut topo = (*state.topo).clone();
2687	topo.tiers[1].provisioned = false;
2688	topo.tiers[1].nodes.clear();
2689	state.topo = Arc::new(topo);
2690	let pool = state.pool.clone();
2691	sqlx::query("UPDATE tier_state SET current_version = '2.0.0' WHERE tier = 'host'")
2692	.execute(&pool)
2693	.await
2694	.unwrap();
2695
2696	let err = promote_inner(state, "a".into(), PromoteBody::default())
2697	.await
2698	.expect_err("promoting to an unprovisioned tier must be refused");
2699	assert!(
2700	matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("not provisioned")),
2701	"got: {err:?}",
2702	);
2703
2704	// And nothing was recorded: the tier keeps whatever it had before.
2705	let (cur, _) = tier_versions(&pool, "a").await;
2706	assert_eq!(
2707	cur.as_deref(),
2708	Some("2.0.0"),
2709	"a refused promote must not advance tier_state",
2710	);
2711	}
2712
2713	#[tokio::test]
2714	async fn promote_advances_the_tier_and_flips_the_symlink_when_gates_are_green() {
2715	// The happy path. Every other promote test asserts a refusal or a red
2716	// outcome, so nothing pinned what a successful promote actually does:
2717	// deploy reaches the node, the `current` symlink flips, tier_state
2718	// advances with previous_version = the version we came off, any stale
2719	// partial flag clears, and the handler reports the nodes it touched.
2720	use crate::topology::Gate;
2721	let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2722
2723	// Source tier `host` gates the promote on cargo_test. Satisfying it is
2724	// the point of the test: the sibling case asserts an unsatisfied gate
2725	// blocks, this one asserts a satisfied gate lets the promote through.
2726	let mut topo = (*state.topo).clone();
2727	topo.tiers[0].gates = vec![Gate::CargoTest];
2728	state.topo = Arc::new(topo);
2729	let pool = state.pool.clone();
2730	let release_root = state.topo.tiers[1].nodes[0].release_root.clone();
2731
2732	// 3.0.0 is staged on the build host and green on `host`.
2733	tokio::fs::create_dir_all(
2734	std::path::Path::new(&release_root)
2735	.join("releases")
2736	.join("3.0.0"),
2737	)
2738	.await
2739	.unwrap();
2740	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('3.0.0','sha',datetime('now'),'/tmp/staged/releases/3.0.0/makenotwork')")
2741	.execute(&pool).await.unwrap();
2742	sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
2743	.execute(&pool)
2744	.await
2745	.unwrap();
2746	insert_gate(&pool, "host", "3.0.0", "cargo_test", 1).await;
2747
2748	// A stale flag from an earlier incident, which a clean rollout clears.
2749	set_partial(&state, &tid("a"), "left over from a previous canary").await;
2750
2751	let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
2752	.await
2753	.expect("gates are green and the node deploys, so the promote succeeds");
2754
2755	assert_eq!(body["tier"], "a");
2756	assert_eq!(body["version"], "3.0.0");
2757	assert_eq!(
2758	body["nodes_deployed"],
2759	serde_json::json!(["a-local"]),
2760	"the response names every node the promote reached",
2761	);
2762
2763	let (cur, prev) = tier_versions(&pool, "a").await;
2764	assert_eq!(cur.as_deref(), Some("3.0.0"));
2765	assert_eq!(
2766	prev.as_deref(),
2767	Some("2.0.0"),
2768	"previous_version is the version we came off, so a rollback aims at it",
2769	);
2770
2771	// The node genuinely moved: the promote is not just bookkeeping.
2772	let link = tokio::fs::read_link(std::path::Path::new(&release_root).join("current"))
2773	.await
2774	.unwrap();
2775	assert_eq!(link.to_string_lossy(), "releases/3.0.0");
2776
2777	let reason: Option<String> =
2778	sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2779	.fetch_one(&pool)
2780	.await
2781	.unwrap();
2782	assert_eq!(
2783	reason, None,
2784	"a clean full rollout clears a stale partial flag",
2785	);
2786
2787	// The deploy is on the record as having succeeded, which is what the
2788	// next promote's gate check and /state both read.
2789	let (node, outcome): (String, String) =
2790	sqlx::query_as("SELECT node, outcome FROM deploys WHERE version = '3.0.0'")
2791	.fetch_one(&pool)
2792	.await
2793	.unwrap();
2794	assert_eq!(node, "a-local");
2795	assert_eq!(outcome, "ok");
2796	}
2797
2798	#[tokio::test]
2799	async fn promote_fails_and_flags_the_tier_when_post_deploy_gates_are_red() {
2800	// The deploy reached every node, so tier_state advances (a stale
2801	// current_version would aim a later rollback at the wrong artifact), but
2802	// the promote must NOT report success: the tier is flagged partial and
2803	// the handler returns the gate failure.
2804	use crate::topology::Gate;
2805	let (mut state, _tmp) = rollback_fixture("1.0.0", "2.0.0").await;
2806	// node_health is the tier's only gate, and it fails closed with no
2807	// probes — an empty executor map gives it nothing to probe while
2808	// deploy_node still falls back to a built executor and succeeds.
2809	let mut topo = (*state.topo).clone();
2810	topo.tiers[1].gates = vec![Gate::NodeHealth];
2811	state.topo = Arc::new(topo);
2812	state.executors = Arc::new(crate::state::ExecutorMap::new());
2813	let pool = state.pool.clone();
2814	// Promote 3.0.0 up from host, which configures no gates of its own.
2815	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('3.0.0','sha',datetime('now'),'/tmp/staged/releases/3.0.0/makenotwork')")
2816	.execute(&pool).await.unwrap();
2817	sqlx::query("UPDATE tier_state SET current_version = '3.0.0' WHERE tier = 'host'")
2818	.execute(&pool)
2819	.await
2820	.unwrap();
2821
2822	let err = promote_inner(state, "a".into(), PromoteBody::default())
2823	.await
2824	.expect_err("red post-deploy gates must fail the promote");
2825	assert!(
2826	matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("node_health")),
2827	"the failure must name the red gate, got: {err:?}",
2828	);
2829
2830	let (cur, _) = tier_versions(&pool, "a").await;
2831	assert_eq!(
2832	cur.as_deref(),
2833	Some("3.0.0"),
2834	"tier_state tracks what the nodes actually run"
2835	);
2836	let reason: Option<String> =
2837	sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = 'a'")
2838	.fetch_one(&pool)
2839	.await
2840	.unwrap();
2841	assert!(
2842	reason.as_deref().is_some_and(\|r\| r.contains("node_health")),
2843	"the tier must be flagged for /state and the TUI, got: {reason:?}",
2844	);
2845	}
2846
2847	#[tokio::test]
2848	async fn set_partial_then_clear_roundtrips() {
2849	let state = test_state().await;
2850	let read = \|\| async {
2851	sqlx::query_scalar::<_, Option<String>>(
2852	"SELECT partial_reason FROM tier_state WHERE tier = 'a'",
2853	)
2854	.fetch_one(&state.pool)
2855	.await
2856	.unwrap()
2857	};
2858	assert_eq!(read().await, None, "consistent tier starts clean");
2859	set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
2860	assert_eq!(
2861	read().await.as_deref(),
2862	Some("canary rollback incomplete: 1/2")
2863	);
2864	clear_partial(&state, &tid("a")).await;
2865	assert_eq!(read().await, None, "clear nulls it back out");
2866	}
2867
2868	#[tokio::test]
2869	async fn state_surfaces_partial_reason() {
2870	use axum::extract::State;
2871	let state = test_state().await;
2872	set_partial(
2873	&state,
2874	&tid("a"),
2875	"first-deploy canary failed: 1 node(s) on 2.0.0",
2876	)
2877	.await;
2878	let Json(view) = get_state(State(state)).await.unwrap();
2879	let a = view.tiers.iter().find(\|t\| t.name == "a").unwrap();
2880	assert_eq!(
2881	a.partial_reason.as_deref(),
2882	Some("first-deploy canary failed: 1 node(s) on 2.0.0"),
2883	);
2884	let host = view.tiers.iter().find(\|t\| t.name == "host").unwrap();
2885	assert_eq!(
2886	host.partial_reason, None,
2887	"untouched tier stays clean in /state"
2888	);
2889	}
2890
2891	#[tokio::test]
2892	async fn state_build_is_null_until_first_rebuild_then_surfaces_latest() {
2893	use axum::extract::State;
2894	let state = test_state().await;
2895	// No build runs yet → build is null, so /state doesn't pretend a build
2896	// is happening.
2897	let Json(view) = get_state(State(state.clone())).await.unwrap();
2898	assert!(view.build.is_none());
2899
2900	// A failed run must surface its cause in /state, not just in /runs.
2901	let run_id = crate::runs::create(&state.pool, &state.cfg.id, "deadbeef")
2902	.await
2903	.unwrap();
2904	crate::runs::mark_failed(&state.pool, run_id, "cargo_test: 3 test(s) failed")
2905	.await
2906	.unwrap();
2907	let Json(view) = get_state(State(state)).await.unwrap();
2908	let b = view.build.expect("build surfaced");
2909	assert_eq!(b.run_id, run_id.0);
2910	assert_eq!(b.result, "failed");
2911	assert_eq!(
2912	b.failure_summary.as_deref(),
2913	Some("cargo_test: 3 test(s) failed")
2914	);
2915	}
2916
2917	#[tokio::test]
2918	async fn status_json_serves_the_shared_payload_over_the_real_router() {
2919	// The mapping itself is tested in `crate::status`. This asserts the
2920	// route is wired, serves valid JSON, and stays internally consistent
2921	// (no dangling child or action references) against a real topology
2922	// rather than a hand-built fixture.
2923	let state = test_state().await;
2924	set_partial(&state, &tid("a"), "canary rollback incomplete: 1/2").await;
2925
2926	let resp = router(state)
2927	.oneshot(
2928	Request::builder()
2929	.uri("/status.json")
2930	.body(Body::empty())
2931	.unwrap(),
2932	)
2933	.await
2934	.unwrap();
2935	assert_eq!(resp.status(), StatusCode::OK);
2936
2937	let body = http_body_util::BodyExt::collect(resp.into_body())
2938	.await
2939	.unwrap()
2940	.to_bytes();
2941	let payload: ops_status::Payload = serde_json::from_slice(&body).unwrap();
2942
2943	assert_eq!(payload.source, crate::status::SOURCE);
2944	assert_eq!(payload.schema, ops_status::SCHEMA_VERSION);
2945	assert_eq!(payload.validate(), Ok(()));
2946	assert_eq!(
2947	payload.node("tier:a").unwrap().status,
2948	ops_status::Status::Failed,
2949	"a partial tier must surface as failed over the wire"
2950	);
2951	assert_eq!(payload.worst_status(), ops_status::Status::Failed);
2952	}
2953
2954	#[tokio::test]
2955	async fn promote_with_explicit_version_but_missing_artifact_404s() {
2956	// Explicit version supplied, gates trivially pass (mm has none in
2957	// test_topo), but `versions` table has no row → 404.
2958	let state = test_state().await;
2959	let app = router(state);
2960	let resp = app
2961	.oneshot(
2962	Request::builder()
2963	.method("POST")
2964	.uri("/promote/a")
2965	.header("content-type", "application/json")
2966	.body(Body::from(r#"{"version":"9.9.9"}"#))
2967	.unwrap(),
2968	)
2969	.await
2970	.unwrap();
2971	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2972	}
2973
2974	// ---- GET /logs/{version}/{gate} ----
2975
2976	async fn state_with_logs_root(logs_root: PathBuf) -> AppState {
2977	let mut s = test_state().await;
2978	let mut cfg = (*s.cfg).clone();
2979	cfg.logs_root = logs_root;
2980	s.cfg = Arc::new(cfg);
2981	s
2982	}
2983
2984	#[tokio::test]
2985	async fn get_gate_log_returns_file_contents() {
2986	let tmp = tempfile::tempdir().unwrap();
2987	let dir = tmp.path().join("0.9.5");
2988	tokio::fs::create_dir_all(&dir).await.unwrap();
2989	tokio::fs::write(dir.join("cargo_test.log"), b"hello sandod\n")
2990	.await
2991	.unwrap();
2992
2993	let state = state_with_logs_root(tmp.path().to_path_buf()).await;
2994	let app = router(state);
2995	let resp = app
2996	.oneshot(
2997	Request::builder()
2998	.uri("/logs/0.9.5/cargo_test")
2999	.body(Body::empty())
3000	.unwrap(),
3001	)
3002	.await
3003	.unwrap();
3004	assert_eq!(resp.status(), StatusCode::OK);
3005	assert_eq!(body_string(resp).await, "hello sandod\n");
3006	}
3007
3008	#[tokio::test]
3009	async fn get_gate_log_404s_when_missing() {
3010	let tmp = tempfile::tempdir().unwrap();
3011	let state = state_with_logs_root(tmp.path().to_path_buf()).await;
3012	let app = router(state);
3013	let resp = app
3014	.oneshot(
3015	Request::builder()
3016	.uri("/logs/0.9.5/cargo_test")
3017	.body(Body::empty())
3018	.unwrap(),
3019	)
3020	.await
3021	.unwrap();
3022	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3023	}
3024
3025	// ---- CF2: bearer-token auth on deploy mutators ----
3026
3027	#[tokio::test]
3028	async fn mutating_route_requires_bearer_when_token_set() {
3029	let mut state = test_state().await;
3030	state.api_token = Some(std::sync::Arc::from("s3cr3t"));
3031	let app = router(state);
3032
3033	// No Authorization header -> 401, before any deploy logic runs.
3034	let resp = app
3035	.clone()
3036	.oneshot(
3037	Request::builder()
3038	.method("POST")
3039	.uri("/promote/a")
3040	.body(Body::empty())
3041	.unwrap(),
3042	)
3043	.await
3044	.unwrap();
3045	assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
3046
3047	// Wrong token -> 401.
3048	let resp = app
3049	.clone()
3050	.oneshot(
3051	Request::builder()
3052	.method("POST")
3053	.uri("/promote/a")
3054	.header("authorization", "Bearer nope")
3055	.body(Body::empty())
3056	.unwrap(),
3057	)
3058	.await
3059	.unwrap();
3060	assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
3061
3062	// Correct token -> passes auth (then blocked downstream by gates /
3063	// missing predecessor version, but specifically NOT 401).
3064	let resp = app
3065	.clone()
3066	.oneshot(
3067	Request::builder()
3068	.method("POST")
3069	.uri("/promote/a")
3070	.header("authorization", "Bearer s3cr3t")
3071	.body(Body::empty())
3072	.unwrap(),
3073	)
3074	.await
3075	.unwrap();
3076	assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
3077	}
3078
3079	#[tokio::test]
3080	async fn read_routes_require_token_when_set() {
3081	// Reads expose prod state (versions, SHAs, gate logs, the event stream),
3082	// so they are bearer-gated too — not just the mutators. A tailnet peer
3083	// without the token gets 401; the TUI presents the token and gets 200.
3084	let mut state = test_state().await;
3085	state.api_token = Some(std::sync::Arc::from("s3cr3t"));
3086	let app = router(state);
3087
3088	// No token -> 401 on a read.
3089	let resp = app
3090	.clone()
3091	.oneshot(
3092	Request::builder()
3093	.uri("/state")
3094	.body(Body::empty())
3095	.unwrap(),
3096	)
3097	.await
3098	.unwrap();
3099	assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
3100
3101	// Correct token -> 200.
3102	let resp = app
3103	.oneshot(
3104	Request::builder()
3105	.uri("/state")
3106	.header("authorization", "Bearer s3cr3t")
3107	.body(Body::empty())
3108	.unwrap(),
3109	)
3110	.await
3111	.unwrap();
3112	assert_eq!(resp.status(), StatusCode::OK);
3113	}
3114
3115	#[tokio::test]
3116	async fn read_routes_open_without_a_token() {
3117	// The loopback/dev posture: no token configured, reads pass through so a
3118	// local TUI needs no credential.
3119	let state = test_state().await;
3120	let app = router(state);
3121	let resp = app
3122	.oneshot(
3123	Request::builder()
3124	.uri("/state")
3125	.body(Body::empty())
3126	.unwrap(),
3127	)
3128	.await
3129	.unwrap();
3130	assert_eq!(resp.status(), StatusCode::OK);
3131	}
3132
3133	#[tokio::test]
3134	async fn self_update_malformed_body_is_400_not_422() {
3135	// TypedBody funnels a JSON deserialize failure through the Error envelope
3136	// (400), not axum's raw 422 — keeping every mutator on one error contract.
3137	let state = test_state().await; // no token -> auth passes, body is the gate
3138	let app = router(state);
3139	let resp = app
3140	.oneshot(
3141	Request::builder()
3142	.method("POST")
3143	.uri("/self-update")
3144	.header("content-type", "application/json")
3145	.body(Body::from("{ not valid json"))
3146	.unwrap(),
3147	)
3148	.await
3149	.unwrap();
3150	assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
3151	}
3152
3153	#[tokio::test]
3154	async fn gate_log_rejects_unknown_gate_kind() {
3155	// The gate segment is an allowlisted GateKind, not a free-form filename:
3156	// an unknown kind is a 404, so `*.log` basenames can't be probed.
3157	let state = test_state().await;
3158	let app = router(state);
3159	let resp = app
3160	.oneshot(
3161	Request::builder()
3162	.uri("/logs/0.9.6/passwd")
3163	.body(Body::empty())
3164	.unwrap(),
3165	)
3166	.await
3167	.unwrap();
3168	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3169	}
3170
3171	/// Path-traversal guard: a `..` segment must not escape logs_root.
3172	/// axum's `{name}` param already rejects a literal `/` in the value, but
3173	/// `..` as a whole segment is structurally valid and must be blocked at
3174	/// the handler.
3175	#[tokio::test]
3176	async fn get_gate_log_rejects_dotdot_segments() {
3177	let tmp = tempfile::tempdir().unwrap();
3178	let state = state_with_logs_root(tmp.path().to_path_buf()).await;
3179	let app = router(state);
3180	let resp = app
3181	.oneshot(
3182	Request::builder()
3183	.uri("/logs/../etc/passwd")
3184	.body(Body::empty())
3185	.unwrap(),
3186	)
3187	.await
3188	.unwrap();
3189	assert_eq!(resp.status(), StatusCode::NOT_FOUND);
3190	}
3191
3192	// ---- build-identity path (wiki release-artifact-identity) ----
3193
3194	/// Insert a settled (passed) build_runs row with content identity and return
3195	/// its id. `staged_path` is the content-addressed release dir the bundle was
3196	/// published to (`releases/<digest16>`).
3197	async fn seed_build(pool: &SqlitePool, sha: &str, version: &str, staged_path: &str) -> i64 {
3198	sqlx::query_scalar(
3199	"INSERT INTO build_runs (sha, version, phase, result, started_at, bundle_digest, staged_path)
3200	VALUES (?, ?, 'done', 'passed', datetime('now'), ?, ?) RETURNING id",
3201	)
3202	.bind(sha)
3203	.bind(version)
3204	.bind(format!("{sha}-digest"))
3205	.bind(staged_path)
3206	.fetch_one(pool)
3207	.await
3208	.unwrap()
3209	}
3210
3211	async fn insert_gate_build(
3212	pool: &SqlitePool,
3213	tier: &str,
3214	version: &str,
3215	kind: &str,
3216	passed: i64,
3217	build_id: i64,
3218	) {
3219	let status = if passed == 1 { "passed" } else { "failed" };
3220	sqlx::query(
3221	"INSERT INTO gate_runs (version, tier, gate_kind, started_at, finished_at, status, build_id) \
3222	VALUES (?, ?, ?, datetime('now'), datetime('now'), ?, ?)",
3223	)
3224	.bind(version)
3225	.bind(tier)
3226	.bind(kind)
3227	.bind(status)
3228	.bind(build_id)
3229	.execute(pool)
3230	.await
3231	.unwrap();
3232	}
3233
3234	#[tokio::test]
3235	async fn unsatisfied_gates_keys_build_evidence_on_build_id() {
3236	let pool = fresh_pool().await;
3237	seed(&pool, "a", "3.0.0").await;
3238	// Two builds of the SAME version string. Only b1's gate ran.
3239	let b1 = seed_build(&pool, "sha1", "3.0.0", "/rel/1111111111111111").await;
3240	let b2 = seed_build(&pool, "sha2", "3.0.0", "/rel/2222222222222222").await;
3241	insert_gate_build(&pool, "a", "3.0.0", "cargo_test", 1, b1).await;
3242
3243	// b1's own evidence satisfies.
3244	let ok = unsatisfied_gates(
3245	&pool,
3246	&crate::domain::AppId::default(),
3247	&tid("a"),
3248	&[Gate::CargoTest],
3249	"3.0.0",
3250	&[PromotedBuild {
3251	platform: None,
3252	build_id: Some(b1),
3253	}],
3254	false,
3255	)
3256	.await
3257	.unwrap();
3258	assert!(ok.is_empty(), "the build that passed its gate is satisfied");
3259
3260	// b2 shares the version but has no gate row of its own: fail closed. This
3261	// is the hole — a rebuild reusing the version must not ride b1's evidence.
3262	let bad = unsatisfied_gates(
3263	&pool,
3264	&crate::domain::AppId::default(),
3265	&tid("a"),
3266	&[Gate::CargoTest],
3267	"3.0.0",
3268	&[PromotedBuild {
3269	platform: None,
3270	build_id: Some(b2),
3271	}],
3272	false,
3273	)
3274	.await
3275	.unwrap();
3276	assert_eq!(
3277	bad,
3278	vec!["cargo_test".to_string()],
3279	"a different build of the same version does not inherit the evidence"
3280	);
3281
3282	// Legacy (pre-identity) callers still resolve by version string.
3283	let legacy = unsatisfied_gates(
3284	&pool,
3285	&crate::domain::AppId::default(),
3286	&tid("a"),
3287	&[Gate::CargoTest],
3288	"3.0.0",
3289	&[],
3290	false,
3291	)
3292	.await
3293	.unwrap();
3294	assert!(legacy.is_empty(), "version-keyed legacy path unchanged");
3295	}
3296
3297	fn plat(s: &str) -> crate::domain::Platform {
3298	crate::domain::Platform::parse(s).unwrap()
3299	}
3300
3301	/// The looseness this closes: one pom version is two bundles with two
3302	/// digests, each accepted through its own intake. Checking only the build the
3303	/// source tier points at let the sibling ship on gate rows nobody read.
3304	/// Every build the promote will ship must show its own passed row.
3305	#[tokio::test]
3306	async fn every_shipped_build_must_show_its_own_gate_evidence() {
3307	let pool = fresh_pool().await;
3308	seed(&pool, "a", "4.0.0").await;
3309	let arm = seed_build(&pool, "sha-arm", "4.0.0", "/rel/aaaaaaaaaaaaaaaa").await;
3310	let x86 = seed_build(&pool, "sha-x86", "4.0.0", "/rel/bbbbbbbbbbbbbbbb").await;
3311	// Only the aarch64 half was gated. This is exactly the state a
3312	// two-architecture release passes through while the second build runs.
3313	insert_gate_build(&pool, "a", "4.0.0", "cargo_test", 1, arm).await;
3314
3315	let both = [
3316	PromotedBuild {
3317	platform: Some(plat("linux/aarch64")),
3318	build_id: Some(arm),
3319	},
3320	PromotedBuild {
3321	platform: Some(plat("linux/x86_64")),
3322	build_id: Some(x86),
3323	},
3324	];
3325	let pending = unsatisfied_gates(
3326	&pool,
3327	&crate::domain::AppId::default(),
3328	&tid("a"),
3329	&[Gate::CargoTest],
3330	"4.0.0",
3331	&both,
3332	false,
3333	)
3334	.await
3335	.unwrap();
3336	assert_eq!(
3337	pending,
3338	vec!["cargo_test (linux/x86_64)".to_string()],
3339	"the ungated half blocks the promote, and the message says which half"
3340	);
3341
3342	// Gate the sibling and the promote clears. Each architecture stands on
3343	// its own evidence; neither inherits the other's.
3344	insert_gate_build(&pool, "a", "4.0.0", "cargo_test", 1, x86).await;
3345	let pending = unsatisfied_gates(
3346	&pool,
3347	&crate::domain::AppId::default(),
3348	&tid("a"),
3349	&[Gate::CargoTest],
3350	"4.0.0",
3351	&both,
3352	false,
3353	)
3354	.await
3355	.unwrap();
3356	assert!(pending.is_empty(), "both halves gated: {pending:?}");
3357	}
3358
3359	/// A single-platform product's error message is exactly what it always was.
3360	/// The platform qualifier is for telling two halves apart, so adding it to a
3361	/// product that has one half would be noise in the one message an operator
3362	/// reads under pressure.
3363	#[tokio::test]
3364	async fn one_shipped_build_reports_an_unqualified_gate_name() {
3365	let pool = fresh_pool().await;
3366	seed(&pool, "a", "5.0.0").await;
3367	let only = seed_build(&pool, "sha-one", "5.0.0", "/rel/cccccccccccccccc").await;
3368
3369	let pending = unsatisfied_gates(
3370	&pool,
3371	&crate::domain::AppId::default(),
3372	&tid("a"),
3373	&[Gate::CargoTest],
3374	"5.0.0",
3375	&[PromotedBuild {
3376	platform: Some(plat("linux/x86_64")),
3377	build_id: Some(only),
3378	}],
3379	false,
3380	)
3381	.await
3382	.unwrap();
3383	assert_eq!(pending, vec!["cargo_test".to_string()]);
3384	}
3385
3386	/// `burn_in` is keyed on the tier's clock, not on a build, so a
3387	/// two-architecture promote must ask about it once rather than name it twice
3388	/// in the failure.
3389	#[tokio::test]
3390	async fn a_tier_scoped_gate_is_reported_once_across_several_builds() {
3391	let pool = fresh_pool().await;
3392	seed(&pool, "a", "6.0.0").await;
3393	let arm = seed_build(&pool, "sha-arm6", "6.0.0", "/rel/dddddddddddddddd").await;
3394	let x86 = seed_build(&pool, "sha-x866", "6.0.0", "/rel/eeeeeeeeeeeeeeee").await;
3395
3396	let pending = unsatisfied_gates(
3397	&pool,
3398	&crate::domain::AppId::default(),
3399	&tid("a"),
3400	&[Gate::BurnIn { hours: 48 }],
3401	"6.0.0",
3402	&[
3403	PromotedBuild {
3404	platform: Some(plat("linux/aarch64")),
3405	build_id: Some(arm),
3406	},
3407	PromotedBuild {
3408	platform: Some(plat("linux/x86_64")),
3409	build_id: Some(x86),
3410	},
3411	],
3412	false,
3413	)
3414	.await
3415	.unwrap();
3416	assert_eq!(
3417	pending,
3418	vec!["burn_in".to_string()],
3419	"a tier-scoped gate belongs to the tier, not to each build"
3420	);
3421	}
3422
3423	/// A tier is usually several nodes on one architecture. Deduplicating means
3424	/// three x86_64 nodes ask about one build once, rather than repeating the
3425	/// same gate name three times in the error.
3426	#[test]
3427	fn distinct_builds_collapses_nodes_that_share_a_build() {
3428	use super::promotion::distinct_builds;
3429	let node = Node {
3430	platform: None,
3431	name: "n1".into(),
3432	ssh_target: "local".into(),
3433	release_root: "/tmp/n1".into(),
3434	service_name: "makenotwork.service".into(),
3435	health_url: None,
3436	config_check_env_file: None,
3437	actuate: crate::topology::default_actuate(),
3438	observe: crate::topology::default_observe(),
3439	companions: Vec::new(),
3440	};
3441	let bundles = vec![
3442	(
3443	&node,
3444	std::path::PathBuf::from("/rel/a"),
3445	Some(plat("linux/x86_64")),
3446	Some(7),
3447	),
3448	(
3449	&node,
3450	std::path::PathBuf::from("/rel/a"),
3451	Some(plat("linux/x86_64")),
3452	Some(7),
3453	),
3454	(
3455	&node,
3456	std::path::PathBuf::from("/rel/b"),
3457	Some(plat("linux/aarch64")),
3458	Some(8),
3459	),
3460	];
3461	let builds = distinct_builds(&bundles);
3462	assert_eq!(builds.len(), 2);
3463	assert_eq!(builds[0].build_id, Some(7));
3464	assert_eq!(builds[1].build_id, Some(8));
3465	}
3466
3467	#[tokio::test]
3468	async fn promote_rejects_an_explicit_version_that_is_not_the_source_build() {
3469	// The burn-in hole: `promote --version Y` used to check the SOURCE tier's
3470	// clock/evidence (which belong to whatever is current there), letting Y
3471	// inherit another build's 48h. Now promote resolves the source's current
3472	// build and refuses an explicit version that isn't it.
3473	let state = test_state().await;
3474	seed_version(&state.pool, "3.0.0").await;
3475	let b = seed_build(&state.pool, "shaB", "3.0.0", "/rel/deadbeefdeadbeef").await;
3476	sqlx::query(
3477	"UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
3478	)
3479	.bind(b)
3480	.execute(&state.pool)
3481	.await
3482	.unwrap();
3483
3484	let err = promote_inner(
3485	state,
3486	"a".into(),
3487	PromoteBody {
3488	version: Some("2.0.0".into()),
3489	..Default::default()
3490	},
3491	)
3492	.await
3493	.expect_err("promoting a version other than the source build must be refused");
3494	assert!(
3495	matches!(&err, crate::error::Error::GateBlocked(m) if m.contains("vouched for")),
3496	"the refusal must name the mismatch, got: {err:?}",
3497	);
3498	}
3499
3500	#[tokio::test]
3501	async fn promote_identity_path_deploys_the_source_build_and_advances_build_id() {
3502	// Full promote through the identity path: source tier points at a build,
3503	// promote resolves the artifact through it, deploys build_runs.staged_path
3504	// (content-addressed), and advances the target's current_build_id.
3505	let mut state = test_state().await;
3506	let node_root = tempfile::tempdir().unwrap();
3507	// Point the a-local node at a real tempdir so the symlink swap is checkable.
3508	let mut topo = (*state.topo).clone();
3509	topo.tiers[1].nodes[0].release_root = node_root.path().to_string_lossy().into_owned();
3510	topo.tiers[1].gates = vec![]; // isolate the deploy fan-out
3511	state.topo = Arc::new(topo);
3512	state.executors = Arc::new(crate::state::build_executors(&state.topo));
3513
3514	// deploys.node FKs into `nodes`, so the promote path needs the row.
3515	sqlx::query("INSERT INTO nodes (name, tier, ssh_target, release_root) VALUES ('a-local', 'a', 'local', ?)")
3516	.bind(node_root.path().to_string_lossy().into_owned())
3517	.execute(&state.pool)
3518	.await
3519	.unwrap();
3520
3521	seed_version(&state.pool, "3.0.0").await;
3522	let staged = format!(
3523	"{}/releases/abc123abc123abc1",
3524	node_root.path().to_string_lossy()
3525	);
3526	let b = seed_build(&state.pool, "shaB", "3.0.0", &staged).await;
3527	sqlx::query(
3528	"UPDATE tier_state SET current_version='3.0.0', current_build_id=? WHERE tier='host'",
3529	)
3530	.bind(b)
3531	.execute(&state.pool)
3532	.await
3533	.unwrap();
3534
3535	let pool = state.pool.clone();
3536	let Json(body) = promote_inner(state, "a".into(), PromoteBody::default())
3537	.await
3538	.expect("identity-path promote succeeds");
3539	assert_eq!(body["version"], "3.0.0");
3540
3541	// Target tier advanced with the build identity, not just the version.
3542	let (cur_v, cur_b): (Option<String>, Option<i64>) = sqlx::query_as(
3543	"SELECT current_version, current_build_id FROM tier_state WHERE tier = 'a'",
3544	)
3545	.fetch_one(&pool)
3546	.await
3547	.unwrap();
3548	assert_eq!(cur_v.as_deref(), Some("3.0.0"));
3549	assert_eq!(cur_b, Some(b), "target records the promoted build id");
3550
3551	// The deploy row is attributed to the build.
3552	let deploy_b: Option<i64> = sqlx::query_scalar(
3553	"SELECT build_id FROM deploys WHERE tier = 'a' ORDER BY id DESC LIMIT 1",
3554	)
3555	.fetch_one(&pool)
3556	.await
3557	.unwrap();
3558	assert_eq!(deploy_b, Some(b));
3559
3560	// The node's `current` points at the content-addressed release dir, whose
3561	// name is the staged_path's basename — not the version.
3562	let link = tokio::fs::read_link(node_root.path().join("current"))
3563	.await
3564	.unwrap();
3565	assert_eq!(link.to_string_lossy(), "releases/abc123abc123abc1");
3566	}
3567
3568	/// Insert a bare `versions` row (FK target for tier_state.current_version).
3569	async fn seed_version(pool: &SqlitePool, version: &str) {
3570	sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES (?, 'sha', datetime('now'), '/tmp/x') ON CONFLICT DO NOTHING")
3571	.bind(version).execute(pool).await.unwrap();
3572	}
3573	}
3574