cloud/reconciler/local_process.rs
1//! `[providers.compute] kind = "local-process"` — run a service's own compute
2//! component as a **kamaji-supervised host process** at the dev tier.
3//!
4//! ## Why this exists
5//!
6//! A component's `kind` says what the service *is*; the mirror's provider slot
7//! says how *this tier* runs it. That split already exists for storage —
8//! `mesofact-static` runs off `miniflare-native` (a Worker over the dev S3 driver) at dev and
9//! `miniflare-container` (containers) at pond, one component kind, two
10//! runtimes — and this is the same split for compute. A `kind = "container"`
11//! component now runs natively at dev and in docker at pond without the
12//! service declaring itself twice.
13//!
14//! The alternative — a `local-process` *component kind* — was rejected: it
15//! would make the artifact shape a property of the service rather than of the
16//! tier, which is exactly the fork that W265 exists to prevent.
17//!
18//! ## Why not just run the container at dev
19//!
20//! Because a container is a different filesystem, and at the dev tier that is
21//! a lie you have to maintain. yah-cloud-admin is the worked example: it reads
22//! `.yah/infra/machines/*.toml`, so its dev container has to bind-mount
23//! `.yah/infra` read-only into `/workspace/.yah/infra` and override
24//! `YAH_CLOUD_ADMIN_WORKSPACE_ROOT` to find it (see
25//! `crates/yah/cloud-admin/workload.toml`). Get the mount wrong and the
26//! service comes up *running, healthy, and describing a fleet of zero
27//! machines*. A host process inherits the operator's actual workspace and the
28//! whole class of failure disappears — along with a docker build per edit.
29//!
30//! ## Config
31//!
32//! The mirror opts in:
33//!
34//! ```toml
35//! # .yah/services/<svc>/mirrors/dev.toml
36//! shape = "local"
37//!
38//! [providers.compute]
39//! kind = "local-process"
40//! ```
41//!
42//! and the component's `workload.toml` describes the process:
43//!
44//! ```toml
45//! [process]
46//! # Cargo package to build before spawning. Omit for a prebuilt binary.
47//! cargo_package = "yah-cloud-admin"
48//! # Binary to exec, relative to the workspace root. Defaults to
49//! # target/<profile>/<cargo_package>.
50//! bin = "target/debug/yah-cloud-admin"
51//! # Extra argv after the binary.
52//! args = []
53//! # Port the process listens on. OPTIONAL. When present it is the readiness
54//! # signal and the mirror's `dev_url`. Omit it for a process that never
55//! # listens — a native GUI, a daemon on a unix socket, a batch loop — and
56//! # readiness falls back to "still alive after a short grace window",
57//! # with no `dev_url` for the Run tab to open.
58//! port = 4325
59//!
60//! [process.env]
61//! YAH_CLOUD_ADMIN_ADDR = "127.0.0.1:4325"
62//! ```
63//!
64//! ## Per-mirror `profile` override, and non-cargo builds
65//!
66//! `workload.toml` is one file shared by every mirror bound to this slot, so
67//! `profile` can be overridden per mirror — the only field that can, since
68//! it's the only one where two tiers of the *same* component legitimately
69//! want different values (a debug dev loop next to a release build, both
70//! pointed at the same `[process]` block):
71//!
72//! ```toml
73//! # .yah/services/<svc>/mirrors/release.toml
74//! shape = "local"
75//!
76//! [providers.compute]
77//! kind = "local-process"
78//! profile = "release"
79//! ```
80//!
81//! For a component with no runnable top-level binary from a plain
82//! `cargo build -p pkg` — a native macOS/iOS app, where Rust only supplies a
83//! static lib for Xcode to link — `pre_build` replaces `cargo_package`
84//! entirely: an operator-authored argv (same trust model as `compose.rs`'s
85//! post-write shell commands, R592-T2) run in `workspace_root` before `bin`
86//! is resolved, streamed into the Run tab's log tail exactly like a cargo
87//! build:
88//!
89//! ```toml
90//! [process]
91//! pre_build = ["./build-dist.sh", "arm64", "--sign"]
92//! bin = "app/macos/dist/NoiseTable.app/Contents/MacOS/NoiseTable"
93//! ```
94//!
95//! ## Portless components
96//!
97//! A process with no TCP listener is a first-class case, not a degenerate
98//! one: noisetable's desktop dev loop is a winit window with zero network
99//! surface. Do not paper over the gap by declaring a port the process never
100//! binds — that trips the readiness timeout below and tears down a perfectly
101//! healthy child, or (worse) adopts an unrelated listener that happens to
102//! answer. Omit `port` and the Run tab renders the row as a log-tail plus
103//! stop card instead of an iframe.
104//!
105//! ## …and what a portless component SHOULD do instead
106//!
107//! Dropping the port drops the only structured thing the supervisor knew
108//! about the process, which leaves log-grepping — for an operator and, worse,
109//! for an agent. So the house default for anything long-running is the
110//! **process-control channel** ([`crate::proc_control`]): a unix socket
111//! speaking one verb, `status`, answering with a document whose `state` uses
112//! kamaji's own `WorkloadState` vocabulary.
113//!
114//! ```toml
115//! [process]
116//! cargo_package = "dev"
117//! # no `port` — a winit window has nothing to bind
118//!
119//! [process.control]
120//! # nothing to configure: the socket path arrives as $YAH_CONTROL_SOCK
121//! ```
122//!
123//! With it, readiness stops being a guess: the process says `starting` while
124//! it loads and `running` when it is up, and an agent can ask what it is
125//! doing instead of parsing sentences out of stdout. A component that already
126//! serves HTTP declares `http_path` instead and shares one endpoint with the
127//! prod tier's `Healthcheck`. The channel is optional — a process that
128//! declines it still runs, on liveness alone.
129//!
130//! @yah:relay(R715, "local-process compute provider: a dev tier that runs the binary, not a container")
131//! @yah:at(2026-08-03T22:21:13Z)
132//! @yah:status(open)
133//! @yah:assignee(agent:bundle-anthropic-ashguard)
134//! @yah:next("T1 (landed): LocalProcessReconciler + Provider::LocalProcess + native_support dedup + both dispatchers + the yah-cloud-admin dev/pond split. See the T1 handoff.")
135//! @yah:next("Follow-on: mesofact-dev's local-static arm and this reconciler now differ only in argv construction. W265 already plans to retire local-static when local-s3-fs lands - that is the moment to consider collapsing them.")
136//! @yah:next("Follow-on: the desktop mirror_run_up has no adopt-probe for local-process the way it does for mesofact-dev. Correct today because the reconciler reaps its predecessor, but an adopt path would let a re-click return the running URL without a rebuild.")
137//! @yah:gotcha("Naming rule this relay encodes, from the operator: if it runs in a container it is pond. Dev is the tier that runs against the operator's real filesystem. No service is required to have all three tiers - a service whose lowest tier is pond is fine.")
138//! @yah:gotcha("The runtime is a property of the MIRROR, not the component. A kind=container component runs natively at dev and in docker at pond, selected by the mirror's compute slot - the same split mesofact-static already had via providers.static (local-static at dev, miniflare-container at pond). Do NOT add a local-process COMPONENT kind: that makes artifact shape a property of the service and reintroduces the per-tier fork W265 exists to prevent.")
139//! @arch:see(.yah/docs/working/W265-service-capabilities-and-drivers.md)
140//!
141//! @yah:ticket(R715-T1, "LocalProcessReconciler + Provider::LocalProcess, wired into both dispatchers; yah-cloud-admin dev/pond split")
142//! @yah:status(review)
143//! @yah:at(2026-08-05T04:54:16Z)
144//! @yah:assignee(agent:bundle-anthropic-ashguard)
145//! @yah:phase(P1)
146//! @yah:parent(R715)
147//! @yah:next("R714-B1 (stop button no-ops on container mirrors) and R714-B2 (tier chip prints 'axum') were found during this work and filed separately - they predate it and are not regressions from it.")
148//! @yah:verify("cargo test --manifest-path oss/yubaba/Cargo.toml -p yah-cloud --lib - 688 passed, 0 failed")
149//! @yah:verify("cargo test -p yah --lib cloud:: - 108 passed")
150//! @yah:verify("cargo test -p xtask --test schema_drift --test workload_envelope - 4 passed")
151//! @yah:verify("MANUAL (done): yah cloud mirror up yah-cloud-admin --env dev binds 127.0.0.1:4325 with no container, and the process logs workspace=/Users/leif/ss/yah - the real checkout, not /workspace")
152//! @yah:verify("MANUAL (done): running mirror up twice in a row replaces the child (pid changes) instead of leaving a dead second instance behind an Address-already-in-use")
153//! @yah:gotcha("NativeRuntime's workload table is IN-MEMORY, so a second `mirror up` from a fresh process cannot see the first one's child. Without the owner.json sidecar + reap the new child dies on Address-already-in-use, wait_for_port sees the OLD listener answering, and the reconcile reports success while the just-edited binary is not what is running. This was observed live before the fix, not theorised. Do not remove the sidecar.")
154//! @yah:gotcha("yah-cloud-admin's pond tier moved to host port 4326 so both tiers can run at once. The container still listens on 4325 internally.")
155//! @yah:next("RESOLVED (R715-T1, 2026-08-04): the cloud.toml blocker comment was corrected by the peer whose uncommitted edits blocked it - lines 34-60 now carry the 2026-08-03 status (published image DONE, workload declaration DONE, cheers key STILL BLOCKED on both an issuer and a 0.8.21 yubaba roll). Nothing left to do there.")
156//! @yah:handoff("Re-verified independently on 2026-08-04 (session:5fe4274e, rune) rather than trusting the prior run's claims. All three automated suites green; counts are HIGHER than the ones recorded above because peers added tests since: yah-cloud lib 698 passed / 0 failed (was 688), yah lib cloud:: 111 passed (was 108), xtask schema_drift + workload_envelope 4 passed (unchanged).")
157//! @yah:verify("RE-VERIFIED 2026-08-04, sidecar survives across sessions: the owner.json left by the 2026-08-03 run ({pid:54869,port:4325}) still matched the live listener a day later - lsof -t on 4325 returned exactly 54869. That is the cross-process ownership link working in the wild, not in a test.")
158//! @yah:verify("RE-VERIFIED 2026-08-04, reap-and-replace: `cargo run -p yah -- cloud mirror up yah-cloud-admin --env dev` reaped pid 54869 (kill -0 confirms gone), spawned 71455, rewrote owner.json to {pid:71455,port:4325}, and 127.0.0.1:4325/ answers HTTP 200. No dead second instance, no Address-already-in-use.")
159//! @yah:verify("RE-VERIFIED 2026-08-04, the failure mode this tier exists to prevent: child cwd is /Users/leif/ss/yah (lsof -d cwd), `docker ps` shows NO cloud-admin container, and /partial/machines renders 9 machine rows against 9 files in .yah/infra/machines/ - us-west-001/002/003 + mac-builder among them. Not a fleet of zero.")
160//! @yah:verify("RE-VERIFIED 2026-08-04, both dispatcher arms compile: `cargo build -p yah` (app/yah/cli/src/cloud.rs:4705) and `cargo check -p desktop --lib` (app/yah/desktop/src/mirror_run.rs:624) are both clean.")
161//! @yah:gotcha("LEFT RUNNING: the re-verification above spawned yah-cloud-admin pid 71455 on 127.0.0.1:4325 and did not stop it - that matches the state found at session start (a dev-tier process was already up). Kill it with `kill $(lsof -nP -iTCP:4325 -sTCP:LISTEN -t)`, or just re-run `mirror up`, which reaps it.")
162//! @yah:gotcha("TRANSIENT, NOT THIS TICKET: `cargo run -p yah` failed once with E0599 mid-verification and compiled clean on immediate retry with no edit from this session - a peer's in-flight change on the shared tree. Same shape as the note in crates/yah/camp-service/tests/e2e.rs:35. If you see it, retry before investigating.")
163//!
164//! @yah:ticket(R715-T2, "Portless `local-process` compute slot — Run tab support for native GUI processes with no TCP listener")
165//! @yah:status(review)
166//! @yah:at(2026-08-14T21:27:45Z)
167//! @yah:assignee(agent:bundle-anthropic-ashguard)
168//! @yah:parent(R715)
169//! @yah:next("Make `[process].port` optional in `ProcessSpec` (this file). When absent, readiness = process spawned and still alive after a short grace window, instead of `wait_for_port`; the `up()` path returns `dev_url = None` to `into_running(...)` (RunningWorkload's dev_url is already `Option`, so this is a narrowing of ProcessSpec, not a new field anywhere downstream).")
170//! @yah:next("Mirror the same required->optional change for `port` on `run.spawn`'s schema and `SpawnArgs` (crates/yah/agent-tools/src/run_tools.rs) so an agent can supervise a portless process through the same tool used for app servers.")
171//! @yah:next("Run tab: a mirror/spawn with no dev_url should render as a log-tail + kill/restart card, no iframe, no URL bar — this is the `stdio` process row A046 (.yah/docs/architecture/A046-yah-run-tab.md) already sketches (`yah-rig (noise) stdio pid 4598 ... [logs] [kill]`) but neither shipped mechanism implements.")
172//! @yah:next("Real external consumer blocked on this today: noisetable's desktop dev loop (`cargo run -p dev` / `cargo run --release -p dev`, a winit native GUI window with zero network surface — no port to bind, ever). Full writeup + the service.toml/mirror.toml/workload.toml shape noisetable wants to declare once this lands: entambi repo, .yah/docs/working/W145-desktop-dev-run-tab.md.")
173//! @yah:gotcha("Do not let a consumer route around this by having the process bind a throwaway port it never uses for anything real — that just trips this reconciler's own `wait_for_port` timeout/teardown, or makes `run.spawn` poll a port that happens to be open for unrelated reasons. Both are worse than the honest gap. The fix belongs in this reconciler (and run_tools.rs), not in a consumer's workload.toml.")
174//! @yah:assumes("Readiness-as-'process alive after N ms grace window' is assumed to be an acceptable v1 fallback (matches what a bare `[kill]`-only scoreboard row needs). A richer readiness probe (e.g. an optional stdout regex match) would be nicer but isn't required to unblock the noisetable use case named above — confirm with whoever picks this up whether the simple version is sufficient, or whether R322-F4's LivePanel / RunStatePanel state model expects something richer for a 'ready' vs 'starting' distinction on portless rows.")
175//! @arch:see(.yah/docs/working/W265-service-capabilities-and-drivers.md)
176//! @arch:see(.yah/docs/architecture/A046-yah-run-tab.md)
177//! @arch:see(crates/yah/agent-tools/src/run_tools.rs)
178//! @yah:handoff("[process].port is now Option<u16> in the local-process reconciler. A portless component skips the port-contention check and wait_for_port; readiness falls back to ALIVE_GRACE (750ms) + pid_alive, dev_url is None, and OwnerRecord.port became Option<u16> with a serde default so sidecars written before this still parse (tested).")
179//! @yah:handoff("run.spawn: SpawnArgs.port is Option<u16> and the JSON schema no longer requires it; the wire still encodes portless as 0. The camp daemon's run_spawn_handler applies the same 750ms grace and fails the spawn with a log tail when a portless child exits immediately - it previously registered a healthy scoreboard row over a corpse. rpc::RunSpawnParams.port's doc claimed 0 meant OS-assigned, which the handler never did; corrected to say portless.")
180//! @yah:handoff("Run tab: MirrorPanel's running-with-no-dev_url state was the placeholder reading 'paste a local URL to preview it here' - advice that cannot be followed for a process that will never have one. Replaced with PortlessCard (host-process chip, what it is, pointer to the log panel and Stop, plus whatever the process reported). LivePanel's port column reads 'stdio' for a portless mirror or spawn and keeps the dash for task runs, via an exported transportLabel with 6 tests.")
181//! @yah:handoff("SCOPE ADDED MID-TICKET, at the operator's direction: the process-control channel. Dropping the port drops the only structured signal a supervisor had, leaving log-grepping - so portless is now paired with an opinionated default that anything long-running SHOULD expose. New module oss/yubaba/crates/cloud/src/proc_control.rs (client + wire types), new optional [process.control] in the reconciler, and W315-process-control-channel.md as the canon. Phase 2 filed as R715-F3.")
182//! @yah:handoff("The contract: one verb, `status`, over newline-JSON on a unix socket (path handed down as $YAH_CONTROL_SOCK) or GET /_yah/status for anything already serving HTTP - same document, two transports, so the cloud tier's Healthcheck{Http} probes the same endpoint the dev tier reads over a socket. `state` is the only required field and its vocabulary is EXACTLY kamaji_proto::WorkloadState, which is the whole compatibility claim: a workload's own report can be handed to the supervisor verbatim. Pinned by a test that fails if either side drifts.")
183//! @yah:handoff("Readiness in the reconciler is now a ladder: control channel > port bind > alive-after-grace. A declared channel supersedes the port because a bound port says nothing about whether the thing behind it finished booting, and a terminal state (failed/exited) short-circuits instead of burning the 20s timeout.")
184//! @yah:handoff("DISCOVERED AND FIXED, wider than the ticket: RunningWorkloadSummary dropped RunningWorkload.notes, so every operator-facing note a reconciler attached since R546-B12 was written into a struct nobody read. Added the field (+ TS type, + three desktop construction sites in mirror_observation.rs). It is what carries the process's own status line to the Run tab.")
185//! @yah:verify("cargo test --manifest-path oss/yubaba/Cargo.toml -p yah-cloud --lib - 830 passed / 0 failed (827 before the last batch). 20 in local_process, 10 in proc_control.")
186//! @yah:verify("END-TO-END, not just unit: three tests drive the real reconciler through kamaji's native backend against a temp workspace. a_portless_component_comes_up_with_no_dev_url spawns /bin/sleep with no port and asserts dev_url None, the liveness-only note, and an owner.json whose port is ABSENT rather than a placeholder zero. a_portless_component_that_exits_immediately_fails_the_reconcile runs /usr/bin/false and asserts the bring-up fails. a_control_channel_decides_readiness_when_declared holds the component at not-ready through a `starting` document and passes on `running`.")
187//! @yah:verify("cargo test -p yah-agent-tools --lib run_tools - 14 passed (was 12; +2 for the optional port).")
188//! @yah:verify("packages/yah/ui - bun run typecheck clean, bun run build clean, bun test src/components/run/ 111 passed / 0 failed.")
189//! @yah:verify("Full bun test 1717 pass / 14 fail / 8 errors. Identical failure counts to what R714-B2 recorded, and the failing names are the same pre-existing four files (StatusPill, dispatchNav, first-run, truncateArg) - no regression from this ticket.")
190//! @yah:verify("cargo check -p yah, cargo check -p desktop --lib, cargo test -p xtask --test schema_drift (3 passed) - all clean.")
191//! @yah:verify("NOT RUN: any manual desktop check. Nobody has looked at the rendered PortlessCard or the stdio transport chip in a running app - there is no portless service declared in this camp to bring up. The e2e reconciler tests cover the backend claim; the two UI changes are covered by tests and typecheck only.")
192//! @yah:gotcha("PRE-EXISTING RED GATE, not from this ticket: `cargo test -p xtask --test workload_envelope` fails on app/yah/web/chat/workload.toml and oss/mesofact/examples/hello/workload.toml, both `missing field routes`. That is R658-B1's known bug (routes written after [build], so TOML scopes it into build.routes) spread to two more files that were never added to KNOWN_GAPS. Both files are committed and untouched by this ticket. Deliberately NOT pinned into KNOWN_GAPS - silently widening the pin is what R658-B1 exists to stop; noted on that ticket instead.")
193//! @yah:gotcha("The reconciler unlinks the control socket path before spawning, because a leftover socket file makes bind fail with EADDRINUSE even with nobody listening. Consequence for anyone writing a test: a producer that binds BEFORE up() has its socket deleted out from under it. Bind after, which is the order a real child sees anyway.")
194//! @yah:gotcha("RESOLVED 2026-08-18 by R658-B1 (@Ashguard:libra): the red gate this ticket's gotcha flagged is now GREEN. app/yah/web/chat/workload.toml and oss/mesofact/examples/hello/workload.toml were migrated to put routes ABOVE [build], along with the other four files and the yah cloud site init scaffold. workload_spec::BuildConfig now carries serde(deny_unknown_fields) so the misplacement is a parse error naming routes. cargo test -p xtask --test workload_envelope passes. Nothing was pinned into KNOWN_GAPS - the four missing-field-routes entries were DELETED. No action needed here; this note exists so the gotcha is not read as still-true.")
195
196use std::collections::{BTreeMap, VecDeque};
197use std::net::{IpAddr, Ipv4Addr, SocketAddr};
198use std::path::PathBuf;
199use std::sync::Arc;
200use std::time::Duration;
201
202use anyhow::{bail, Context, Result};
203use async_trait::async_trait;
204use kamaji::native::NativeRuntime;
205use kamaji::{Kamaji, MeshAssignment, MeshIdent};
206use serde::Deserialize;
207use tokio::sync::{oneshot, Mutex as AsyncMutex};
208use tracing::{info, warn};
209use workload_spec::EnvVar;
210use workload_spec::MeshExpose;
211
212use super::native_support::{
213 capture_paths, native_spec, sanitize_ident, spawn_native_log_supervisor,
214};
215use super::{
216 into_running, wait_for_port, LogBuffer, PhaseCursor, ReconcileCtx, Reconciler, RunningWorkload,
217};
218use crate::proc_control::{self, ControlEndpoint, ReadyOutcome};
219use crate::{MirrorProviderSlot, MirrorShape, Provider};
220
221/// The slot role a natively-run compute component occupies on its mirror.
222const SLOT: &str = "compute";
223
224/// How long to wait for the process to bind its declared port before calling
225/// the reconcile failed. Generous because the first `cargo build` of a cold
226/// target dir is included in the caller's patience, not in this window — the
227/// build finishes before the spawn.
228const READY_TIMEOUT: Duration = Duration::from_secs(20);
229
230/// Readiness window for a component with no declared port: if the child is
231/// still alive this long after `deploy_workload` returned, call it up.
232///
233/// This is deliberately not a health check — a portless process has no
234/// surface to check. What it catches is the common failure the operator
235/// would otherwise see as a green row and an empty window: a binary that
236/// execs and dies immediately (missing dylib, bad argv, panic at startup).
237/// Anything that fails later shows up in the log tail, which is the only
238/// signal a portless process has.
239const ALIVE_GRACE: Duration = Duration::from_millis(750);
240
241/// True when `mirror` binds its compute slot to `local-process`. This is the
242/// dispatch predicate — callers use it to choose this reconciler over the
243/// container one for the same component kind.
244pub fn slot_declared(mirror: &crate::MirrorConfig) -> bool {
245 matches!(
246 mirror.providers.get(SLOT),
247 Some(MirrorProviderSlot::Inline {
248 kind: Provider::LocalProcess,
249 ..
250 })
251 )
252}
253
254/// `profile` from this mirror's `[providers.compute]` inline extra fields, if
255/// declared — the per-mirror override described on [`ProcessSpec::pre_build`]
256/// above `up()`. `None` for a `Reference` slot or an `Inline` slot with no
257/// `profile` key, both of which fall back to `workload.toml`'s own value.
258fn mirror_profile_override(mirror: &crate::MirrorConfig) -> Option<String> {
259 match mirror.providers.get(SLOT) {
260 Some(MirrorProviderSlot::Inline { fields, .. }) => fields
261 .get("profile")
262 .and_then(|v| v.as_str())
263 .map(str::to_string),
264 _ => None,
265 }
266}
267
268/// Reconciler for components bound to a `local-process` compute slot.
269#[derive(Debug, Default)]
270pub struct LocalProcessReconciler {
271 /// External sink to stream build+run output into, in place of the
272 /// private buffer `up()` would otherwise create. Lets a caller register
273 /// the buffer for polling BEFORE calling `up()`, so a slow `cargo build`
274 /// is visible to a poller while the reconcile is still in flight rather
275 /// than only after `up()` returns.
276 log_buf: Option<LogBuffer>,
277 /// Published the moment the build phase finishes (before readiness/spawn
278 /// even starts), so a caller polling an in-flight `up()` can render a
279 /// build→run transition live instead of only learning about it from the
280 /// finished [`RunningWorkload::build_log_end`].
281 build_log_end_sink: Option<PhaseCursor>,
282}
283
284impl LocalProcessReconciler {
285 pub fn new() -> Self {
286 Self::default()
287 }
288
289 /// Stream into a caller-owned [`LogBuffer`] instead of a private one.
290 pub fn with_log_buf(mut self, log_buf: LogBuffer) -> Self {
291 self.log_buf = Some(log_buf);
292 self
293 }
294
295 /// Publish the build→run boundary into a caller-owned [`PhaseCursor`] as
296 /// soon as it's known, rather than only once `up()` returns it.
297 pub fn with_build_log_end_sink(mut self, sink: PhaseCursor) -> Self {
298 self.build_log_end_sink = Some(sink);
299 self
300 }
301}
302
303/// On-disk `workload.toml` shape — only the `[process]` section is read here.
304/// Other sections (`[build]`, `[run]`) belong to the container reconciler and
305/// are ignored, so one component file can describe both tiers.
306#[derive(Debug, Default, Deserialize)]
307struct ProcessComponent {
308 #[serde(default)]
309 process: Option<ProcessSpec>,
310}
311
312#[derive(Debug, Deserialize)]
313struct ProcessSpec {
314 /// Argv of an operator-authored command to run before `cargo_package`'s
315 /// build (or standalone, when there is no `cargo_package` — e.g. a native
316 /// macOS/iOS component where the runnable artifact comes out of
317 /// `xcodebuild`, not `cargo build -p`, and Rust only supplies a static
318 /// lib for Xcode to link). Runs in `workspace_root`, streamed into
319 /// `log_buf` exactly like `cargo_build`, via the same [`run_streaming`].
320 /// Same trust model as `compose.rs`'s post-write shell commands (R592-T2:
321 /// operator-authored, not attacker input) — this is TOML the operator
322 /// wrote, not a request body.
323 #[serde(default)]
324 pre_build: Option<Vec<String>>,
325 /// Cargo package to `cargo build -p` before spawning. `None` → the binary
326 /// is expected to already exist.
327 #[serde(default)]
328 cargo_package: Option<String>,
329 /// Binary path relative to the workspace root (absolute taken as-is).
330 /// `None` → `target/<profile>/<cargo_package>`.
331 #[serde(default)]
332 bin: Option<String>,
333 /// Extra argv appended after the binary.
334 #[serde(default)]
335 args: Vec<String>,
336 /// Port the process listens on. Optional: when set it is both the
337 /// readiness signal and the `dev_url`; when absent the component is
338 /// portless (native GUI, unix socket, batch loop) and readiness falls
339 /// back to [`ALIVE_GRACE`].
340 #[serde(default)]
341 port: Option<u16>,
342 /// Environment for the child.
343 #[serde(default)]
344 env: BTreeMap<String, String>,
345 /// Cargo profile used for both the build and the default binary path.
346 #[serde(default = "default_profile")]
347 profile: String,
348 /// Opt in to the process-control channel ([`crate::proc_control`]) — the
349 /// house default for anything long-running. When present, the process's
350 /// own report decides readiness, which is strictly better than both other
351 /// signals: a bound port says nothing about whether the thing behind it
352 /// has finished booting, and a live pid says nothing at all.
353 #[serde(default)]
354 control: Option<ControlSpec>,
355}
356
357/// `[process.control]` — where to ask the process how it is doing.
358///
359/// Both fields optional and mutually exclusive in practice: give `http_path`
360/// for a process that already serves HTTP (it then shares an endpoint with
361/// the prod tier's healthcheck), otherwise the channel is a unix socket
362/// whose path this reconciler chooses and passes down as `$YAH_CONTROL_SOCK`.
363#[derive(Debug, Default, Deserialize)]
364struct ControlSpec {
365 /// Override the supervisor-chosen socket path. Rarely needed — a process
366 /// that reads `$YAH_CONTROL_SOCK` needs nothing here.
367 #[serde(default)]
368 socket: Option<String>,
369 /// Serve the status document over HTTP at this path instead, against the
370 /// component's declared `port`. Defaults to
371 /// [`proc_control::DEFAULT_HTTP_PATH`] when set to an empty string.
372 #[serde(default)]
373 http_path: Option<String>,
374}
375
376fn default_profile() -> String {
377 "debug".to_string()
378}
379
380#[async_trait]
381impl Reconciler for LocalProcessReconciler {
382 fn kind(&self) -> &'static str {
383 "local-process"
384 }
385
386 async fn up(&self, ctx: ReconcileCtx<'_>) -> Result<RunningWorkload> {
387 ctx.materialize().await?;
388
389 // A host process is the operator's own machine by definition. Refusing
390 // non-local shapes here keeps a `local-process` slot from silently
391 // meaning "run it on my laptop" in a mirror that describes a fleet.
392 if !matches!(ctx.mirror.shape, MirrorShape::Local) {
393 bail!(
394 "component {}: `local-process` is a dev-tier compute slot — mirror shape is \
395 {:?}, not `local`. Deploy the prod tier via `yah cloud workload deploy`.",
396 ctx.component.id,
397 ctx.mirror.shape,
398 );
399 }
400
401 let mut spec = load_process_spec(&ctx)?;
402 // A mirror can override `profile` (and only `profile` — everything
403 // else about the process is one declaration shared by every tier)
404 // via `[providers.compute]`'s inline extra fields, same pattern the
405 // schema already documents for other inline provider kinds ("bucket,
406 // zone, dns, …"). Without this, two mirrors pointing at the same
407 // component's workload.toml (e.g. noisetable-desktop's dev + release
408 // tiers) would be unable to build different profiles from one file.
409 if let Some(profile) = mirror_profile_override(ctx.mirror) {
410 spec.profile = profile;
411 }
412
413 // Created here, before the build, rather than down where the
414 // supervisor is spawned: `cargo_build` streams into it live, and
415 // `build_log_end` below records the cursor where build output stops
416 // and the spawned process's own stdout/stderr begins. A caller-
417 // supplied buffer (`with_log_buf`) is reused as-is so it can already
418 // be registered for polling before this call started.
419 let log_buf = self.log_buf.clone().unwrap_or_default();
420
421 // Build first, if asked. Doing this before the port probe means a
422 // compile error surfaces as a compile error, rather than as a bind
423 // timeout on a binary that was never rebuilt. Streamed into
424 // `log_buf` as it runs (not buffered until exit) so a slow or
425 // failing build shows progress in the Run tab instead of going
426 // silent until it's done — and, unlike the old `.output()` capture,
427 // a *successful* build's output is no longer discarded outright.
428 if let Some(argv) = &spec.pre_build {
429 run_pre_build(ctx.workspace_root, argv, &log_buf).await?;
430 }
431
432 let mut build_log_end = None;
433 if let Some(pkg) = &spec.cargo_package {
434 cargo_build(ctx.workspace_root, pkg, &spec.profile, &log_buf).await?;
435 let cursor = log_buf.cursor().await;
436 build_log_end = Some(cursor);
437 if let Some(sink) = &self.build_log_end_sink {
438 sink.set(cursor).await;
439 }
440 }
441
442 let bin = resolve_binary(&spec, ctx.workspace_root)?;
443
444 // Kamaji's native backend captures stdio under this dir; scope it per
445 // workspace so concurrent camps don't collide on the same ident.
446 let state_dir = ctx.workspace_root.join(".yah/jit/native");
447 let ident_str = sanitize_ident(&format!(
448 "local-process-{}-{}-{}",
449 ctx.service.name, ctx.env, ctx.component.id
450 ));
451 let ident = MeshIdent(ident_str.clone());
452
453 // Replace any predecessor before spawning. `ContainerReconciler` has
454 // the same semantics ("run clears any prior container of the same name
455 // first"), and here it is load-bearing rather than tidy: NativeRuntime
456 // keeps its workload table **in memory**, so a second `mirror up` from
457 // a fresh process knows nothing about the first. Without this the new
458 // child dies on `Address already in use`, `wait_for_port` sees the
459 // *old* listener still answering, and the reconcile reports success
460 // while the binary the operator just edited is not the one running.
461 // That is the R602-B4 foreign-listener failure with a friendlier
462 // disguise, and on the dev tier it is the single most costly thing
463 // this reconciler could get wrong.
464 let owner_path = state_dir.join(&ident_str).join("owner.json");
465 reap_predecessor(&owner_path).await;
466
467 // Anything still holding the port now is not ours. Refuse rather than
468 // adopt it: a `dev_url` that answers with someone else's service is
469 // worse than a failed bring-up. A portless component has no address to
470 // contend over, so there is nothing to check.
471 let addr = spec
472 .port
473 .map(|port| SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port));
474 if let Some(addr) = addr {
475 if tokio::net::TcpStream::connect(addr).await.is_ok() {
476 bail!(
477 "component {}: {addr} is already held by a process this reconciler did not \
478 start. Stop it, or give [process] a different `port`.",
479 ctx.component.id,
480 );
481 }
482 }
483
484 let mut argv = vec![bin.display().to_string()];
485 argv.extend(spec.args.iter().cloned());
486
487 // Process-control channel. Resolved before the spawn because the
488 // socket variant has to reach the child in its environment — a
489 // conforming process binds `$YAH_CONTROL_SOCK` and does nothing when
490 // it is unset, so the same binary runs outside a camp untouched.
491 let control = control_endpoint(&spec, &state_dir, &ident_str);
492 if let Some(ControlEndpoint::Socket(path)) = &control {
493 // A socket file left by the predecessor makes `bind` fail with
494 // EADDRINUSE even though nothing is listening on it. Reaping the
495 // process does not remove it; unlinking here does.
496 if let Some(dir) = path.parent() {
497 tokio::fs::create_dir_all(dir).await.ok();
498 }
499 tokio::fs::remove_file(path).await.ok();
500 }
501
502 let mut env: Vec<EnvVar> = spec
503 .env
504 .iter()
505 .map(|(name, value)| EnvVar {
506 name: name.clone(),
507 value: workload_spec::EnvValue::Literal {
508 value: value.clone(),
509 },
510 })
511 .collect();
512 if let Some(ControlEndpoint::Socket(path)) = &control {
513 // Declared `[process.env]` wins: an operator who set the variable
514 // by hand meant it.
515 if !spec.env.contains_key(proc_control::CONTROL_SOCK_ENV) {
516 env.push(EnvVar {
517 name: proc_control::CONTROL_SOCK_ENV.to_string(),
518 value: workload_spec::EnvValue::Literal {
519 value: path.display().to_string(),
520 },
521 });
522 }
523 }
524
525 let mut workload = native_spec(&ident_str, argv, env);
526 // A declared `[process] port` reaches the child as `PORT` / `PORT_HTTP`
527 // (R844-T13) by riding the spec, not by a per-caller string. Declared
528 // `[process.env]` still wins — the native backend layers spec env last.
529 // A portless component contributes nothing and gets neither variable.
530 workload.expose.mesh.ports = MeshExpose::anonymous_ports(spec.port);
531 let runtime = Arc::new(NativeRuntime::new(&state_dir));
532 let mesh = MeshAssignment::inlined(Ipv4Addr::LOCALHOST);
533
534 info!(
535 binary = %bin.display(),
536 port = ?spec.port,
537 ident = %ident_str,
538 "spawning local-process component (kamaji native backend)",
539 );
540
541 let deployed = runtime
542 .deploy_workload(&workload, &mesh)
543 .await
544 .with_context(|| {
545 format!(
546 "deploying component {} via kamaji native backend (binary {})",
547 ctx.component.id,
548 bin.display(),
549 )
550 })?;
551
552 let (stdout_path, stderr_path) = capture_paths(&state_dir, &ident_str);
553
554 // Record ownership so the *next* reconcile — in a different process,
555 // with a different in-memory NativeRuntime — can find and reap this
556 // child instead of colliding with it.
557 write_owner(&owner_path, deployed.task_pid as i32, spec.port).await;
558
559 // Readiness, best signal first.
560 //
561 // 1. The control channel, when declared: the process says when it is
562 // serving. Nothing else here can distinguish "bound" from "ready".
563 // 2. The port bind: a dead process must not hand the Run tab a URL
564 // that will never answer.
565 // 3. Still alive after a grace window: all a portless process that
566 // declines the channel can offer.
567 let mut notes: Vec<String> = Vec::new();
568
569 if let Some(endpoint) = &control {
570 match proc_control::wait_ready(endpoint, READY_TIMEOUT).await {
571 ReadyOutcome::Ready(status) => {
572 info!(
573 endpoint = %endpoint,
574 status = %status.summary(),
575 "local-process reported ready over its control channel",
576 );
577 notes.push(format!("control: {}", status.summary()));
578 }
579 ReadyOutcome::Terminal(status) => {
580 let tail = read_capture_tail(&stdout_path, &stderr_path).await;
581 runtime.teardown_workload(&ident).await.ok();
582 bail!(
583 "component {} reported `{}` on its control channel ({}){tail}",
584 ctx.component.id,
585 status.summary(),
586 endpoint,
587 );
588 }
589 ReadyOutcome::TimedOut { last } => {
590 let tail = read_capture_tail(&stdout_path, &stderr_path).await;
591 runtime.teardown_workload(&ident).await.ok();
592 let seen = match last {
593 Some(s) => format!("last reported `{}`", s.summary()),
594 None => "never answered — is it binding $YAH_CONTROL_SOCK?".to_string(),
595 };
596 bail!(
597 "component {} was not ready within {READY_TIMEOUT:?} on {endpoint} \
598 ({seen}){tail}",
599 ctx.component.id,
600 );
601 }
602 }
603 }
604
605 let dev_url = match addr {
606 Some(addr) => {
607 // Already ready per the control channel? Then the bind has
608 // happened by definition, and re-waiting would only add
609 // latency to a process that already said it is serving.
610 if control.is_none() && !wait_for_port(addr, READY_TIMEOUT).await {
611 warn!(addr = %addr, "local-process did not bind within timeout; tearing down");
612 let tail = read_capture_tail(&stdout_path, &stderr_path).await;
613 runtime.teardown_workload(&ident).await.ok();
614 bail!(
615 "component {} did not bind {addr} within {READY_TIMEOUT:?}{tail}",
616 ctx.component.id,
617 );
618 }
619 Some(format!("http://{addr}"))
620 }
621 None => {
622 if control.is_none() {
623 tokio::time::sleep(ALIVE_GRACE).await;
624 if !pid_alive(deployed.task_pid as i32) {
625 let tail = read_capture_tail(&stdout_path, &stderr_path).await;
626 runtime.teardown_workload(&ident).await.ok();
627 bail!(
628 "component {} (portless) exited within {ALIVE_GRACE:?} of \
629 starting{tail}",
630 ctx.component.id,
631 );
632 }
633 notes.push(
634 "portless with no [process.control] — readiness is liveness only; \
635 see yah-cloud's proc_control module"
636 .to_string(),
637 );
638 }
639 None
640 }
641 };
642 info!(dev_url = ?dev_url, pid = deployed.task_pid, "local-process ready");
643
644 let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
645 let supervisor = spawn_native_log_supervisor(
646 runtime,
647 ident,
648 log_buf.clone(),
649 stdout_path,
650 stderr_path,
651 shutdown_rx,
652 );
653
654 Ok(into_running(
655 "local-process",
656 SLOT,
657 dev_url,
658 None,
659 Some(log_buf),
660 shutdown_tx,
661 supervisor,
662 )
663 .with_notes(notes)
664 .with_control(control)
665 .with_build_log_end(build_log_end))
666 }
667}
668
669/// Resolve `[process.control]` into an endpoint to poll, or `None` when the
670/// component declines the channel.
671///
672/// HTTP needs a port to hang off — a component asking for `http_path` without
673/// a `port` has contradicted itself, and falling back to the socket silently
674/// would hide that. It is reported as no endpoint at all rather than guessed
675/// at; the caller then treats the process as unmonitored, which is the honest
676/// reading of a self-contradicting declaration.
677fn control_endpoint(
678 spec: &ProcessSpec,
679 state_dir: &std::path::Path,
680 ident: &str,
681) -> Option<ControlEndpoint> {
682 let control = spec.control.as_ref()?;
683 if let Some(path) = &control.http_path {
684 let port = spec.port?;
685 let path = if path.is_empty() {
686 proc_control::DEFAULT_HTTP_PATH
687 } else {
688 path
689 };
690 return Some(ControlEndpoint::Http(format!(
691 "http://127.0.0.1:{port}{path}"
692 )));
693 }
694 let sock = match &control.socket {
695 Some(s) => PathBuf::from(s),
696 None => state_dir.join(ident).join("control.sock"),
697 };
698 Some(ControlEndpoint::Socket(sock))
699}
700
701/// Read `<workload_dir>/workload.toml` and pull out its `[process]` section.
702fn load_process_spec(ctx: &ReconcileCtx<'_>) -> Result<ProcessSpec> {
703 let path = ctx.workload_dir().join("workload.toml");
704 let src =
705 std::fs::read_to_string(&path).with_context(|| format!("reading {}", path.display()))?;
706 let parsed: ProcessComponent =
707 toml::from_str(&src).with_context(|| format!("parsing {}", path.display()))?;
708 parsed.process.with_context(|| {
709 format!(
710 "component {} is bound to a `local-process` compute slot but {} declares no \
711 [process] section — add one with at least `bin` or `cargo_package`",
712 ctx.component.id,
713 path.display(),
714 )
715 })
716}
717
718/// Resolve the binary to exec: explicit `bin`, else the cargo convention.
719fn resolve_binary(spec: &ProcessSpec, workspace_root: &std::path::Path) -> Result<PathBuf> {
720 let rel = match (&spec.bin, &spec.cargo_package) {
721 (Some(bin), _) => PathBuf::from(bin),
722 (None, Some(pkg)) => PathBuf::from(format!("target/{}/{pkg}", spec.profile)),
723 (None, None) => bail!(
724 "[process] declares neither `bin` nor `cargo_package` — one is needed to know \
725 what to run"
726 ),
727 };
728 let path = if rel.is_absolute() {
729 rel
730 } else {
731 workspace_root.join(rel)
732 };
733 if !path.exists() {
734 bail!(
735 "[process] binary {} does not exist — set `cargo_package` to have it built, or \
736 point `bin` at an existing file",
737 path.display(),
738 );
739 }
740 Ok(path)
741}
742
743/// Run `[process.pre_build]`'s argv to completion in `workspace_root`,
744/// streamed live into `log_buf` via the same [`run_streaming`] `cargo_build`
745/// uses. Runs before `cargo_package`'s build (if any) so a failing pre_build
746/// step surfaces as its own error rather than as a confusing downstream
747/// build/spawn failure.
748async fn run_pre_build(
749 workspace_root: &std::path::Path,
750 argv: &[String],
751 log_buf: &LogBuffer,
752) -> Result<()> {
753 let (program, args) = argv
754 .split_first()
755 .context("[process.pre_build] argv must have at least one element")?;
756 let mut cmd = tokio::process::Command::new(program);
757 cmd.args(args);
758 cmd.current_dir(workspace_root);
759
760 info!(cmd = %argv.join(" "), "running [process.pre_build] for local-process component");
761 let (ok, stderr_tail) = run_streaming(cmd, log_buf)
762 .await
763 .with_context(|| format!("spawning pre_build command `{}`", argv.join(" ")))?;
764 if !ok {
765 bail!(
766 "[process.pre_build] `{}` failed:\n{}",
767 argv.join(" "),
768 stderr_tail.join("\n")
769 );
770 }
771 Ok(())
772}
773
774/// Lines of trailing stderr [`run_streaming`] keeps in memory to fold into a
775/// failure's error message. Compiler diagnostics are what an operator needs
776/// to act on a failed build; the full transcript is in `log_buf` regardless.
777const STDERR_TAIL_CAP: usize = 30;
778
779/// `cargo build -p <pkg>` in the workspace root, streamed live into `log_buf`
780/// (R715 follow-up — the desktop has no console to inherit stdout/stderr
781/// into, so this is the only way build progress reaches the Run tab, and
782/// unlike a buffer-then-discard capture it doesn't drop a *successful*
783/// build's output on the floor).
784async fn cargo_build(
785 workspace_root: &std::path::Path,
786 pkg: &str,
787 profile: &str,
788 log_buf: &LogBuffer,
789) -> Result<()> {
790 let mut cmd = tokio::process::Command::new("cargo");
791 cmd.arg("build").arg("-p").arg(pkg);
792 if profile == "release" {
793 cmd.arg("--release");
794 } else if profile != "debug" {
795 cmd.arg("--profile").arg(profile);
796 }
797 cmd.current_dir(workspace_root);
798
799 info!(
800 package = pkg,
801 profile, "cargo build for local-process component"
802 );
803 let (ok, stderr_tail) = run_streaming(cmd, log_buf)
804 .await
805 .with_context(|| format!("spawning `cargo build -p {pkg}`"))?;
806 if !ok {
807 bail!("cargo build -p {pkg} failed:\n{}", stderr_tail.join("\n"));
808 }
809 Ok(())
810}
811
812/// Run `cmd` to completion, streaming stdout and stderr into `log_buf`
813/// line-by-line as they're produced — not buffered until exit, so a slow
814/// command shows progress instead of going silent until it's done. Returns
815/// whether it exited successfully plus the last [`STDERR_TAIL_CAP`] stderr
816/// lines, which a caller can fold into its own error on failure without
817/// re-reading `log_buf` (a compiler's diagnostics land on stderr, so that's
818/// the stream worth quoting back).
819async fn run_streaming(mut cmd: tokio::process::Command, log_buf: &LogBuffer) -> Result<(bool, Vec<String>)> {
820 use tokio::io::{AsyncBufReadExt, BufReader};
821
822 cmd.stdout(std::process::Stdio::piped());
823 cmd.stderr(std::process::Stdio::piped());
824 let mut child = cmd.spawn().context("spawning command")?;
825 let stdout = child.stdout.take().expect("stdout piped above");
826 let stderr = child.stderr.take().expect("stderr piped above");
827
828 let stderr_tail: Arc<AsyncMutex<VecDeque<String>>> = Arc::new(AsyncMutex::new(VecDeque::new()));
829
830 let out_task = {
831 let log_buf = log_buf.clone();
832 tokio::spawn(async move {
833 let mut lines = BufReader::new(stdout).lines();
834 while let Ok(Some(line)) = lines.next_line().await {
835 log_buf.push(line).await;
836 }
837 })
838 };
839 let err_task = {
840 let log_buf = log_buf.clone();
841 let tail = stderr_tail.clone();
842 tokio::spawn(async move {
843 let mut lines = BufReader::new(stderr).lines();
844 while let Ok(Some(line)) = lines.next_line().await {
845 log_buf.push(line.clone()).await;
846 let mut t = tail.lock().await;
847 t.push_back(line);
848 if t.len() > STDERR_TAIL_CAP {
849 t.pop_front();
850 }
851 }
852 })
853 };
854
855 let status = child
856 .wait()
857 .await
858 .context("waiting on spawned command")?;
859 // The drain tasks were spawned onto the runtime above, so they've been
860 // reading concurrently with `wait()` regardless of join order; awaiting
861 // them here just confirms both pipes hit EOF before returning the tail.
862 let _ = out_task.await;
863 let _ = err_task.await;
864
865 let tail = stderr_tail.lock().await.iter().cloned().collect();
866 Ok((status.success(), tail))
867}
868
869/// What a previous `up()` recorded about the child it left running.
870///
871/// This exists because [`NativeRuntime`]'s workload table is in-memory: a
872/// `yah cloud mirror up` from the shell and a ▶ click in the desktop are
873/// different processes, and neither can see the other's children. The sidecar
874/// is the only thing that makes "replace the predecessor" possible across them.
875#[derive(Debug, serde::Serialize, Deserialize)]
876struct OwnerRecord {
877 pid: i32,
878 /// The port the recorded child bound, when it declared one. `default` so a
879 /// sidecar written before portless components existed still parses — those
880 /// records always carry a port, and a missing one now means portless.
881 #[serde(default)]
882 port: Option<u16>,
883}
884
885/// Record the child we just spawned. Best-effort: failing to write the sidecar
886/// must not fail an otherwise-successful bring-up — the cost is a manual kill
887/// on the next re-run, not a broken mirror.
888async fn write_owner(path: &std::path::Path, pid: i32, port: Option<u16>) {
889 if let Some(dir) = path.parent() {
890 if tokio::fs::create_dir_all(dir).await.is_err() {
891 return;
892 }
893 }
894 if let Ok(json) = serde_json::to_vec(&OwnerRecord { pid, port }) {
895 if let Err(e) = tokio::fs::write(path, json).await {
896 warn!(path = %path.display(), error = %e, "could not record local-process owner");
897 }
898 }
899}
900
901/// Stop the child a previous `up()` left behind, if it is still alive.
902///
903/// SIGTERM, a short grace, then SIGKILL — the same ladder kamaji's own teardown
904/// uses. The sidecar is removed either way, including when the pid is already
905/// gone, so a stale record can't linger and make the next run think it has a
906/// predecessor to wait on.
907async fn reap_predecessor(owner_path: &std::path::Path) {
908 let Ok(bytes) = tokio::fs::read(owner_path).await else {
909 return;
910 };
911 let _ = tokio::fs::remove_file(owner_path).await;
912 let Ok(owner) = serde_json::from_slice::<OwnerRecord>(&bytes) else {
913 return;
914 };
915 if !pid_alive(owner.pid) {
916 return;
917 }
918
919 info!(
920 pid = owner.pid,
921 port = ?owner.port,
922 "replacing predecessor local-process"
923 );
924 signal_pid(owner.pid, libc::SIGTERM);
925 for _ in 0..50 {
926 if !pid_alive(owner.pid) {
927 return;
928 }
929 tokio::time::sleep(Duration::from_millis(100)).await;
930 }
931 warn!(
932 pid = owner.pid,
933 "predecessor ignored SIGTERM; sending SIGKILL"
934 );
935 signal_pid(owner.pid, libc::SIGKILL);
936 // Give the kernel a moment to release the port before we probe it.
937 tokio::time::sleep(Duration::from_millis(200)).await;
938}
939
940/// `kill(pid, 0)` — true when a process with this pid exists and we may signal
941/// it. Cannot prove the pid is still *our* child (pids are reused), which is
942/// why the sidecar is written next to this workload's capture dir and removed
943/// on every read: the window where a recycled pid could be signalled is one
944/// reconcile wide, and the alternative (never reaping) is a guaranteed
945/// collision rather than a theoretical one.
946fn pid_alive(pid: i32) -> bool {
947 // SAFETY: `kill` with signal 0 performs error checking only and never
948 // delivers a signal; any pid value is a defined input.
949 unsafe { libc::kill(pid, 0) == 0 }
950}
951
952fn signal_pid(pid: i32, sig: i32) {
953 // SAFETY: same contract as above — `kill` is defined for any pid/signal
954 // pair and reports failure through its return value, which we ignore
955 // because a vanished process is the outcome we wanted anyway.
956 unsafe {
957 libc::kill(pid, sig);
958 }
959}
960
961/// Last few lines of the capture files, formatted for an error message. A bind
962/// timeout with no output is nearly impossible to act on; the process almost
963/// always said why on the way down.
964async fn read_capture_tail(stdout_path: &std::path::Path, stderr_path: &std::path::Path) -> String {
965 let mut lines: Vec<String> = Vec::new();
966 for path in [stderr_path, stdout_path] {
967 if let Ok(s) = tokio::fs::read_to_string(path).await {
968 lines.extend(s.lines().rev().take(10).map(str::to_string));
969 }
970 }
971 if lines.is_empty() {
972 return String::new();
973 }
974 lines.reverse();
975 format!("\nlast output:\n{}", lines.join("\n"))
976}
977
978#[cfg(test)]
979mod tests {
980 use super::*;
981
982 /// `run_streaming` interleaves stdout/stderr into `log_buf` as the child
983 /// produces them (not buffered until exit), reports success/failure by
984 /// exit status, and keeps a bounded stderr tail for the caller's error
985 /// message — the shape `cargo_build` builds on, exercised here without
986 /// depending on an actual `cargo` invocation.
987 #[tokio::test]
988 async fn run_streaming_captures_both_streams_and_reports_failure() {
989 let log_buf = LogBuffer::new();
990 let mut cmd = tokio::process::Command::new("sh");
991 cmd.arg("-c")
992 .arg("echo out-line; echo err-line >&2; exit 3");
993
994 let (ok, tail) = run_streaming(cmd, &log_buf).await.unwrap();
995 assert!(!ok, "exit 3 must be reported as failure");
996 assert_eq!(tail, vec!["err-line".to_string()]);
997
998 let (lines, _) = log_buf.since(0).await;
999 assert!(lines.contains(&"out-line".to_string()), "{lines:?}");
1000 assert!(lines.contains(&"err-line".to_string()), "{lines:?}");
1001 }
1002
1003 /// The tail exists so a failure's error message doesn't have to re-read
1004 /// `log_buf` — but it must stay bounded, or a build that fails after
1005 /// paging through thousands of warnings dumps all of them into the error.
1006 #[tokio::test]
1007 async fn run_streaming_bounds_the_stderr_tail() {
1008 let log_buf = LogBuffer::new();
1009 let mut cmd = tokio::process::Command::new("sh");
1010 cmd.arg("-c").arg("for i in $(seq 1 50); do echo \"e$i\" >&2; done; exit 1");
1011
1012 let (ok, tail) = run_streaming(cmd, &log_buf).await.unwrap();
1013 assert!(!ok);
1014 assert_eq!(tail.len(), STDERR_TAIL_CAP);
1015 assert_eq!(tail.first(), Some(&"e21".to_string()));
1016 assert_eq!(tail.last(), Some(&"e50".to_string()));
1017
1018 // Every line still reached the log buffer, unbounded — only the
1019 // in-memory tail kept for the error message is capped.
1020 let (lines, _) = log_buf.since(0).await;
1021 assert_eq!(lines.len(), 50);
1022 }
1023
1024 fn spec(bin: Option<&str>, pkg: Option<&str>) -> ProcessSpec {
1025 ProcessSpec {
1026 pre_build: None,
1027 cargo_package: pkg.map(str::to_string),
1028 bin: bin.map(str::to_string),
1029 args: vec![],
1030 port: Some(4325),
1031 env: BTreeMap::new(),
1032 profile: "debug".to_string(),
1033 control: None,
1034 }
1035 }
1036
1037 #[test]
1038 fn parses_a_process_section_alongside_container_sections() {
1039 // One workload.toml describes both tiers; each reconciler reads its own
1040 // section and ignores the other's.
1041 let src = r#"
1042schema_version = 1
1043kind = "container"
1044
1045[build]
1046image = "yah-local/x:dev"
1047
1048[run]
1049port = 4325
1050
1051[process]
1052cargo_package = "yah-cloud-admin"
1053port = 4325
1054args = ["--verbose"]
1055
1056[process.env]
1057YAH_CLOUD_ADMIN_ADDR = "127.0.0.1:4325"
1058"#;
1059 let c: ProcessComponent = toml::from_str(src).unwrap();
1060 let p = c.process.unwrap();
1061 assert_eq!(p.cargo_package.as_deref(), Some("yah-cloud-admin"));
1062 assert_eq!(p.port, Some(4325));
1063 assert_eq!(p.args, vec!["--verbose".to_string()]);
1064 assert_eq!(p.profile, "debug");
1065 assert_eq!(
1066 p.env.get("YAH_CLOUD_ADMIN_ADDR").map(String::as_str),
1067 Some("127.0.0.1:4325")
1068 );
1069 }
1070
1071 /// R715-T2: the noisetable shape — a native GUI with no network surface.
1072 /// `port` must be *absent*, not zero: a zero would be a real port number
1073 /// to every layer downstream that reads one.
1074 #[test]
1075 fn a_process_section_without_a_port_is_portless() {
1076 let src = r#"
1077[process]
1078cargo_package = "dev"
1079profile = "release"
1080args = ["--window"]
1081"#;
1082 let p: ProcessComponent = toml::from_str(src).unwrap();
1083 let p = p.process.unwrap();
1084 assert_eq!(p.port, None, "an omitted port must stay absent");
1085 assert_eq!(p.cargo_package.as_deref(), Some("dev"));
1086 assert_eq!(p.profile, "release");
1087 }
1088
1089 /// The macOS/iOS shape: no `cargo_package` (there is no runnable
1090 /// top-level binary from a plain `cargo build -p pkg` — Rust only
1091 /// supplies a static lib for Xcode), the entire build is `pre_build`.
1092 #[test]
1093 fn parses_a_pre_build_argv() {
1094 let src = r#"
1095[process]
1096pre_build = ["./build-dist.sh", "arm64", "--sign"]
1097bin = "app/macos/dist/NoiseTable.app/Contents/MacOS/NoiseTable"
1098"#;
1099 let p: ProcessComponent = toml::from_str(src).unwrap();
1100 let p = p.process.unwrap();
1101 assert_eq!(
1102 p.pre_build,
1103 Some(vec![
1104 "./build-dist.sh".to_string(),
1105 "arm64".to_string(),
1106 "--sign".to_string(),
1107 ])
1108 );
1109 assert_eq!(p.cargo_package, None);
1110 }
1111
1112 /// The house default: a portless component opting into the control
1113 /// channel with an empty `[process.control]` and nothing else. The socket
1114 /// path is the supervisor's to choose — the process reads it from
1115 /// `$YAH_CONTROL_SOCK`.
1116 #[test]
1117 fn an_empty_control_section_yields_the_supervisor_chosen_socket() {
1118 let src = r#"
1119[process]
1120cargo_package = "dev"
1121
1122[process.control]
1123"#;
1124 let c: ProcessComponent = toml::from_str(src).unwrap();
1125 let spec = c.process.unwrap();
1126 let got = control_endpoint(&spec, std::path::Path::new("/s"), "svc-dev-app");
1127 assert_eq!(
1128 got,
1129 Some(ControlEndpoint::Socket(PathBuf::from(
1130 "/s/svc-dev-app/control.sock"
1131 ))),
1132 );
1133 }
1134
1135 #[test]
1136 fn no_control_section_means_no_channel() {
1137 assert_eq!(
1138 control_endpoint(&spec(Some("bin/x"), None), std::path::Path::new("/s"), "id"),
1139 None,
1140 );
1141 }
1142
1143 #[test]
1144 fn an_http_path_binds_the_channel_to_the_declared_port() {
1145 let src = r#"
1146[process]
1147cargo_package = "svc"
1148port = 4325
1149
1150[process.control]
1151http_path = "/_yah/status"
1152"#;
1153 let spec = toml::from_str::<ProcessComponent>(src).unwrap().process.unwrap();
1154 assert_eq!(
1155 control_endpoint(&spec, std::path::Path::new("/s"), "id"),
1156 Some(ControlEndpoint::Http(
1157 "http://127.0.0.1:4325/_yah/status".into()
1158 )),
1159 );
1160 }
1161
1162 /// An empty `http_path` means "the conventional one" rather than a URL
1163 /// ending in the port — the shape most likely to be written by hand.
1164 #[test]
1165 fn an_empty_http_path_falls_back_to_the_conventional_one() {
1166 let src = r#"
1167[process]
1168cargo_package = "svc"
1169port = 4325
1170
1171[process.control]
1172http_path = ""
1173"#;
1174 let spec = toml::from_str::<ProcessComponent>(src).unwrap().process.unwrap();
1175 assert_eq!(
1176 control_endpoint(&spec, std::path::Path::new("/s"), "id"),
1177 Some(ControlEndpoint::Http(format!(
1178 "http://127.0.0.1:4325{}",
1179 crate::proc_control::DEFAULT_HTTP_PATH
1180 ))),
1181 );
1182 }
1183
1184 /// `http_path` with no port is a contradiction. Silently falling back to a
1185 /// socket would give the component a channel it never agreed to serve, and
1186 /// then fail readiness twenty seconds later with a message about a socket
1187 /// nobody mentioned.
1188 #[test]
1189 fn an_http_path_without_a_port_yields_no_endpoint() {
1190 let src = r#"
1191[process]
1192cargo_package = "svc"
1193
1194[process.control]
1195http_path = "/_yah/status"
1196"#;
1197 let spec = toml::from_str::<ProcessComponent>(src).unwrap().process.unwrap();
1198 assert_eq!(control_endpoint(&spec, std::path::Path::new("/s"), "id"), None);
1199 }
1200
1201 #[test]
1202 fn an_explicit_socket_path_overrides_the_default() {
1203 let src = r#"
1204[process]
1205cargo_package = "svc"
1206
1207[process.control]
1208socket = "/tmp/mine.sock"
1209"#;
1210 let spec = toml::from_str::<ProcessComponent>(src).unwrap().process.unwrap();
1211 assert_eq!(
1212 control_endpoint(&spec, std::path::Path::new("/s"), "id"),
1213 Some(ControlEndpoint::Socket(PathBuf::from("/tmp/mine.sock"))),
1214 );
1215 }
1216
1217 /// The sidecar gained an optional `port` when portless components landed.
1218 /// Records written before that carry a bare `u16` and must still parse —
1219 /// otherwise the first reconcile after an upgrade silently fails to reap
1220 /// its predecessor, which is the exact collision the sidecar exists for.
1221 #[test]
1222 fn an_owner_record_written_before_portless_components_still_parses() {
1223 let old: OwnerRecord = serde_json::from_str(r#"{"pid":54869,"port":4325}"#).unwrap();
1224 assert_eq!(old.pid, 54869);
1225 assert_eq!(old.port, Some(4325));
1226
1227 let portless: OwnerRecord = serde_json::from_str(r#"{"pid":71455}"#).unwrap();
1228 assert_eq!(portless.port, None);
1229 }
1230
1231 #[test]
1232 fn a_workload_without_a_process_section_parses_as_none() {
1233 let c: ProcessComponent = toml::from_str("[run]\nport = 1\n").unwrap();
1234 assert!(c.process.is_none());
1235 }
1236
1237 #[test]
1238 fn binary_defaults_to_the_cargo_convention() {
1239 let tmp = tempfile::tempdir().unwrap();
1240 std::fs::create_dir_all(tmp.path().join("target/debug")).unwrap();
1241 std::fs::write(tmp.path().join("target/debug/yah-cloud-admin"), b"").unwrap();
1242 let got = resolve_binary(&spec(None, Some("yah-cloud-admin")), tmp.path()).unwrap();
1243 assert_eq!(got, tmp.path().join("target/debug/yah-cloud-admin"));
1244 }
1245
1246 #[test]
1247 fn explicit_bin_wins_over_the_cargo_convention() {
1248 let tmp = tempfile::tempdir().unwrap();
1249 std::fs::create_dir_all(tmp.path().join("bin")).unwrap();
1250 std::fs::write(tmp.path().join("bin/custom"), b"").unwrap();
1251 let got = resolve_binary(&spec(Some("bin/custom"), Some("pkg")), tmp.path()).unwrap();
1252 assert_eq!(got, tmp.path().join("bin/custom"));
1253 }
1254
1255 /// Spawning a path that isn't there fails with an exec error several
1256 /// layers down; naming the missing file here is the actionable version.
1257 #[test]
1258 fn a_missing_binary_is_named_before_we_try_to_exec_it() {
1259 let tmp = tempfile::tempdir().unwrap();
1260 let err = resolve_binary(&spec(None, Some("nope")), tmp.path()).unwrap_err();
1261 assert!(err.to_string().contains("does not exist"), "{err}");
1262 }
1263
1264 #[test]
1265 fn neither_bin_nor_package_is_an_error() {
1266 let tmp = tempfile::tempdir().unwrap();
1267 let err = resolve_binary(&spec(None, None), tmp.path()).unwrap_err();
1268 assert!(err.to_string().contains("neither"), "{err}");
1269 }
1270
1271 /// The bug this whole path exists to prevent: a second `up()` in a fresh
1272 /// process must stop the first one's child. NativeRuntime's table is
1273 /// in-memory, so the sidecar is the only link between the two runs.
1274 ///
1275 /// The assertion is on the child's exit status, not on [`pid_alive`],
1276 /// because this test *is* the child's parent: a signalled child it has not
1277 /// waited on stays a zombie, and `kill(pid, 0)` succeeds against a zombie.
1278 /// In production the spawning process is either gone (CLI — init reaps) or
1279 /// still holding kamaji's supervisor task (desktop — that reaps), so
1280 /// neither leaves one behind.
1281 #[tokio::test]
1282 async fn reap_stops_a_live_predecessor_and_clears_the_record() {
1283 let tmp = tempfile::tempdir().unwrap();
1284 let owner_path = tmp.path().join("owner.json");
1285
1286 let mut child = tokio::process::Command::new("sleep")
1287 .arg("120")
1288 .spawn()
1289 .unwrap();
1290 let pid = child.id().unwrap() as i32;
1291 assert!(pid_alive(pid));
1292
1293 write_owner(&owner_path, pid, Some(4325)).await;
1294 reap_predecessor(&owner_path).await;
1295
1296 let status = tokio::time::timeout(Duration::from_secs(5), child.wait())
1297 .await
1298 .expect("a signalled `sleep 120` must have exited well inside 5s")
1299 .unwrap();
1300 assert!(!status.success(), "predecessor should have been signalled");
1301 assert!(!owner_path.exists(), "sidecar should be cleared");
1302 }
1303
1304 /// A record left behind by a crash names a pid that is gone. Reaping it
1305 /// must be a no-op that still clears the file — a stale record that
1306 /// survives would make every later run wait on a corpse.
1307 #[tokio::test]
1308 async fn reap_clears_a_stale_record_without_signalling_anything() {
1309 let tmp = tempfile::tempdir().unwrap();
1310 let owner_path = tmp.path().join("owner.json");
1311
1312 // Exited process: spawn and wait, so the pid is definitely dead.
1313 let mut child = tokio::process::Command::new("true").spawn().unwrap();
1314 let pid = child.id().unwrap() as i32;
1315 child.wait().await.unwrap();
1316
1317 write_owner(&owner_path, pid, Some(4325)).await;
1318 reap_predecessor(&owner_path).await;
1319 assert!(!owner_path.exists());
1320 }
1321
1322 #[tokio::test]
1323 async fn reap_with_no_record_is_a_no_op() {
1324 let tmp = tempfile::tempdir().unwrap();
1325 reap_predecessor(&tmp.path().join("owner.json")).await;
1326 }
1327
1328 // ── End-to-end: a portless component through the real reconciler ────────
1329 //
1330 // The unit tests above pin parsing and endpoint resolution. This one runs
1331 // `up()` itself — kamaji native backend, real child process, real sidecar
1332 // — because the claim R715-T2 makes is not "the struct has an Option", it
1333 // is "a component with no port comes up and reports no dev_url", and only
1334 // the whole path can say that.
1335
1336 fn portless_service() -> crate::ServiceConfig {
1337 crate::ServiceConfig {
1338 schema_version: 1,
1339 name: "noisy".into(),
1340 domain: "noisy.example".into(),
1341 health_path: None,
1342 components: vec![crate::ServiceComponent {
1343 mount: None,
1344 id: "gui".into(),
1345 kind: "container".into(),
1346 path: "gui".into(),
1347 role: "compute".into(),
1348 publishes: None,
1349 wave: 0,
1350 git: None,
1351 deploy: Default::default(),
1352 }],
1353 db: crate::DbCatalog::default(),
1354 }
1355 }
1356
1357 fn local_process_mirror() -> crate::MirrorConfig {
1358 let mut providers = BTreeMap::new();
1359 providers.insert(
1360 SLOT.to_string(),
1361 MirrorProviderSlot::Inline {
1362 kind: Provider::LocalProcess,
1363 fields: Default::default(),
1364 },
1365 );
1366 crate::MirrorConfig {
1367 schema_version: 1,
1368 shape: MirrorShape::Local,
1369 providers,
1370 ingress: Default::default(),
1371 ingress_machines: Vec::new(),
1372 drivers: Default::default(),
1373 asset_aliases: Default::default(),
1374 build: Default::default(),
1375 }
1376 }
1377
1378 /// The noisetable shape end to end: no `port`, no control channel. Comes
1379 /// up, reports no `dev_url`, and says in its notes that readiness here is
1380 /// liveness only — so nobody reads a green row as a health claim.
1381 #[tokio::test]
1382 async fn a_portless_component_comes_up_with_no_dev_url() {
1383 let ws = tempfile::tempdir().unwrap();
1384 std::fs::create_dir_all(ws.path().join("gui")).unwrap();
1385 std::fs::write(
1386 ws.path().join("gui/workload.toml"),
1387 "schema_version = 1\nkind = \"container\"\n\n[process]\nbin = \"/bin/sleep\"\nargs = [\"30\"]\n",
1388 )
1389 .unwrap();
1390
1391 let svc = portless_service();
1392 let mirror = local_process_mirror();
1393 let ctx = ReconcileCtx {
1394 workspace_root: ws.path(),
1395 service: &svc,
1396 component: &svc.components[0],
1397 mirror: &mirror,
1398 env: "dev",
1399 scope: crate::ProviderScope::singleton(),
1400 };
1401
1402 let mut running = LocalProcessReconciler::new().up(ctx).await.unwrap();
1403 assert_eq!(running.dev_url, None, "a portless component has no URL");
1404 assert_eq!(running.slot, SLOT);
1405 assert_eq!(
1406 running.build_log_end, None,
1407 "no cargo_package was declared, so nothing was built"
1408 );
1409 assert!(
1410 running.notes.iter().any(|n| n.contains("portless")),
1411 "the liveness-only caveat must be visible to an operator: {:?}",
1412 running.notes,
1413 );
1414
1415 // The sidecar is what lets a later reconcile in another process reap
1416 // this child. Portless or not, it must be written — and its `port`
1417 // must be absent rather than a placeholder zero.
1418 let owner: OwnerRecord = serde_json::from_slice(
1419 &std::fs::read(
1420 ws.path()
1421 .join(".yah/jit/native")
1422 .join(sanitize_ident("local-process-noisy-dev-gui"))
1423 .join("owner.json"),
1424 )
1425 .expect("owner sidecar must exist"),
1426 )
1427 .unwrap();
1428 assert_eq!(owner.port, None);
1429 assert!(pid_alive(owner.pid), "the child should still be running");
1430
1431 if let Some(tx) = running.shutdown.take() {
1432 tx.send(()).ok();
1433 }
1434 if let Some(sup) = running.supervisor.take() {
1435 let _ = tokio::time::timeout(Duration::from_secs(5), sup).await;
1436 }
1437 }
1438
1439 /// The house default, end to end: a portless component that declares
1440 /// `[process.control]` is held at "not ready" until something answers
1441 /// `status` with `running`.
1442 ///
1443 /// The child here is `/bin/sleep` and the socket is served by the test,
1444 /// which is the point — the reconciler must not care *who* answers, only
1445 /// that the declared endpoint does. A real workload binds the same path
1446 /// out of `$YAH_CONTROL_SOCK`.
1447 #[tokio::test]
1448 async fn a_control_channel_decides_readiness_when_declared() {
1449 let ws = tempfile::tempdir().unwrap();
1450 let sock = ws.path().join("control.sock");
1451 std::fs::create_dir_all(ws.path().join("gui")).unwrap();
1452 std::fs::write(
1453 ws.path().join("gui/workload.toml"),
1454 format!(
1455 "schema_version = 1\nkind = \"container\"\n\n[process]\nbin = \"/bin/sleep\"\n\
1456 args = [\"30\"]\n\n[process.control]\nsocket = \"{}\"\n",
1457 sock.display()
1458 ),
1459 )
1460 .unwrap();
1461
1462 // Stand-in producer: answers `starting` once, then `running`.
1463 //
1464 // It binds *after* a delay on purpose. `up()` unlinks a predecessor's
1465 // socket file before spawning (a leftover file makes `bind` fail with
1466 // EADDRINUSE even with nobody listening), so a producer that bound
1467 // first would have its socket deleted out from under it — which is
1468 // exactly the order a real child sees: supervisor unlinks, child
1469 // starts, child binds.
1470 let server = {
1471 let sock = sock.clone();
1472 tokio::spawn(async move {
1473 tokio::time::sleep(Duration::from_millis(400)).await;
1474 let listener = tokio::net::UnixListener::bind(&sock).unwrap();
1475 for doc in [r#"{"state":"starting","detail":"loading"}"#, r#"{"state":"running","detail":"1 window"}"#] {
1476 let Ok((stream, _)) = listener.accept().await else {
1477 return;
1478 };
1479 let (rd, mut wr) = stream.into_split();
1480 let mut line = String::new();
1481 tokio::io::AsyncBufReadExt::read_line(
1482 &mut tokio::io::BufReader::new(rd),
1483 &mut line,
1484 )
1485 .await
1486 .ok();
1487 tokio::io::AsyncWriteExt::write_all(&mut wr, format!("{doc}\n").as_bytes())
1488 .await
1489 .ok();
1490 }
1491 })
1492 };
1493
1494 let svc = portless_service();
1495 let mirror = local_process_mirror();
1496 let ctx = ReconcileCtx {
1497 workspace_root: ws.path(),
1498 service: &svc,
1499 component: &svc.components[0],
1500 mirror: &mirror,
1501 env: "dev",
1502 scope: crate::ProviderScope::singleton(),
1503 };
1504
1505 let mut running = LocalProcessReconciler::new().up(ctx).await.unwrap();
1506 assert_eq!(running.dev_url, None);
1507 assert!(
1508 running
1509 .notes
1510 .iter()
1511 .any(|n| n.contains("control:") && n.contains("1 window")),
1512 "the process's own words should reach the operator: {:?}",
1513 running.notes,
1514 );
1515 assert!(
1516 !running.notes.iter().any(|n| n.contains("liveness only")),
1517 "a component WITH a channel must not be labelled liveness-only: {:?}",
1518 running.notes,
1519 );
1520
1521 server.abort();
1522 if let Some(tx) = running.shutdown.take() {
1523 tx.send(()).ok();
1524 }
1525 if let Some(sup) = running.supervisor.take() {
1526 let _ = tokio::time::timeout(Duration::from_secs(5), sup).await;
1527 }
1528 }
1529
1530 /// A portless component that dies on exec must fail the reconcile, not
1531 /// register a healthy row over a corpse. `/usr/bin/false` is the smallest
1532 /// honest stand-in for "binary exits immediately".
1533 #[tokio::test]
1534 async fn a_portless_component_that_exits_immediately_fails_the_reconcile() {
1535 let ws = tempfile::tempdir().unwrap();
1536 std::fs::create_dir_all(ws.path().join("gui")).unwrap();
1537 std::fs::write(
1538 ws.path().join("gui/workload.toml"),
1539 "schema_version = 1\nkind = \"container\"\n\n[process]\nbin = \"/usr/bin/false\"\n",
1540 )
1541 .unwrap();
1542
1543 let svc = portless_service();
1544 let mirror = local_process_mirror();
1545 let ctx = ReconcileCtx {
1546 workspace_root: ws.path(),
1547 service: &svc,
1548 component: &svc.components[0],
1549 mirror: &mirror,
1550 env: "dev",
1551 scope: crate::ProviderScope::singleton(),
1552 };
1553
1554 let err = LocalProcessReconciler::new()
1555 .up(ctx)
1556 .await
1557 .expect_err("a process that exits immediately is not a successful bring-up");
1558 assert!(
1559 err.to_string().contains("portless"),
1560 "the error should name why it was judged this way: {err}"
1561 );
1562 }
1563
1564 /// A failing `[process.pre_build]` must fail the reconcile before ever
1565 /// touching `cargo_package`/spawn — a broken xcodebuild/codesign step
1566 /// should surface as its own error, not as a confusing "binary not
1567 /// found" from a later stage that never should have run.
1568 #[tokio::test]
1569 async fn a_failing_pre_build_fails_the_reconcile_before_spawn() {
1570 let ws = tempfile::tempdir().unwrap();
1571 std::fs::create_dir_all(ws.path().join("gui")).unwrap();
1572 std::fs::write(
1573 ws.path().join("gui/workload.toml"),
1574 "schema_version = 1\nkind = \"container\"\n\n[process]\npre_build = [\"sh\", \"-c\", \"exit 1\"]\nbin = \"/bin/sleep\"\nargs = [\"30\"]\n",
1575 )
1576 .unwrap();
1577
1578 let svc = portless_service();
1579 let mirror = local_process_mirror();
1580 let ctx = ReconcileCtx {
1581 workspace_root: ws.path(),
1582 service: &svc,
1583 component: &svc.components[0],
1584 mirror: &mirror,
1585 env: "dev",
1586 scope: crate::ProviderScope::singleton(),
1587 };
1588
1589 let err = LocalProcessReconciler::new()
1590 .up(ctx)
1591 .await
1592 .expect_err("a failing pre_build must not be treated as a successful bring-up");
1593 assert!(
1594 err.to_string().contains("pre_build"),
1595 "the error should name the stage that failed: {err}"
1596 );
1597 }
1598
1599 #[test]
1600 fn mirror_profile_override_reads_the_inline_extra_field() {
1601 let mut mirror = local_process_mirror();
1602 if let Some(MirrorProviderSlot::Inline { fields, .. }) = mirror.providers.get_mut(SLOT) {
1603 fields.insert("profile".to_string(), toml::Value::String("release".to_string()));
1604 }
1605 assert_eq!(
1606 mirror_profile_override(&mirror),
1607 Some("release".to_string())
1608 );
1609 }
1610
1611 #[test]
1612 fn mirror_profile_override_is_absent_by_default() {
1613 // No `profile` key declared on the mirror's compute slot — falls back
1614 // to whatever workload.toml itself declares.
1615 assert_eq!(mirror_profile_override(&local_process_mirror()), None);
1616 }
1617
1618 #[test]
1619 fn slot_declared_only_matches_the_local_process_compute_slot() {
1620 let mut m = crate::MirrorConfig {
1621 schema_version: 1,
1622 shape: MirrorShape::Local,
1623 ingress: Default::default(),
1624 ingress_machines: Vec::new(),
1625 providers: Default::default(),
1626 drivers: Default::default(),
1627 asset_aliases: Default::default(),
1628 build: Default::default(),
1629 };
1630 assert!(!slot_declared(&m));
1631 m.providers.insert(
1632 SLOT.to_string(),
1633 MirrorProviderSlot::Inline {
1634 kind: Provider::LocalContainer,
1635 fields: Default::default(),
1636 },
1637 );
1638 assert!(!slot_declared(&m), "a container slot is not a process slot");
1639 m.providers.insert(
1640 SLOT.to_string(),
1641 MirrorProviderSlot::Inline {
1642 kind: Provider::LocalProcess,
1643 fields: Default::default(),
1644 },
1645 );
1646 assert!(slot_declared(&m));
1647 }
1648}