diff --git a/.dockerignore b/.dockerignore index 85a6ee5..0c96593 100644 --- a/.dockerignore +++ b/.dockerignore @@ -20,17 +20,6 @@ !apps/myelin/ !apps/myelin/node-image/ !apps/myelin/node-image/** -!target/ -target/* -!target/release/ -target/release/* -!target/release/myelin-worker -!target/x86_64-unknown-linux-musl/ -target/x86_64-unknown-linux-musl/* -!target/x86_64-unknown-linux-musl/release/ -target/x86_64-unknown-linux-musl/release/* -!target/debug/ -target/debug/* crates/bindings/python/.venv/ crates/bindings/python/.venv/** crates/**/.pytest_cache/ @@ -41,4 +30,3 @@ crates/**/target/ crates/**/target/** **/.venv **/.venv/** -!target/debug/myelin-worker diff --git a/apps/myelin/node-image/Dockerfile b/apps/myelin/node-image/Dockerfile index dbf65f2..67b87e3 100644 --- a/apps/myelin/node-image/Dockerfile +++ b/apps/myelin/node-image/Dockerfile @@ -1,14 +1,59 @@ -# Myelin node agent: a thin Rust control-plane layer over the CUDA base. -# Workload frameworks are supplied by workload-specific images. -# Build from workspace root after compiling the Rust binary: -# cargo build --release -p myelin --bin myelin-worker +# Myelin node agent with the Python Swactor binding used by contextual scripts. +# Build from the workspace root: # docker build -f apps/myelin/node-image/Dockerfile.base -t myelin-node-base:cuda12.6 . -# docker build -f apps/myelin/node-image/Dockerfile --build-arg BASE_IMAGE=myelin-node-base:cuda12.6 -t myelin-node:latest . - +# docker build -f apps/myelin/node-image/Dockerfile -t myelin-node:latest . ARG BASE_IMAGE=myelin-node-base:cuda12.6 -ARG MYELIN_NODE_BIN=target/release/myelin-worker -FROM ${BASE_IMAGE} -ARG MYELIN_NODE_BIN=target/release/myelin-worker +FROM ${BASE_IMAGE} AS wheel-builder +ARG RUST_TOOLCHAIN=nightly-2026-02-07 -COPY ${MYELIN_NODE_BIN} /usr/local/bin/myelin-node -RUN chmod +x /usr/local/bin/myelin-node +ENV PATH=/root/.cargo/bin:${PATH} +RUN apt-get update && \ + apt-get install -y --no-install-recommends \ + build-essential \ + curl \ + pkg-config \ + python3 \ + python3-dev \ + python3-pip && \ + curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | \ + sh -s -- -y --profile minimal --default-toolchain none && \ + rustup toolchain install "${RUST_TOOLCHAIN}" --profile minimal --component rustc-dev && \ + python3 -m pip install --break-system-packages --no-cache-dir 'maturin>=1.7,<2' && \ + rm -rf /var/lib/apt/lists/* +WORKDIR /workspace +COPY . . +RUN --mount=type=cache,id=myelin-python-cargo-registry,target=/root/.cargo/registry \ + --mount=type=cache,id=myelin-python-target,target=/workspace/target \ + cargo build --release -p myelin --bin myelin-worker && \ + install -Dm0755 target/release/myelin-worker /out/myelin-worker && \ + maturin build \ + --manifest-path crates/bindings/python/Cargo.toml \ + --release \ + --interpreter python3 \ + --out /wheel + +FROM ${BASE_IMAGE} +RUN apt-get update && \ + apt-get install -y --no-install-recommends python3 python3-pip && \ + apt-get clean && \ + rm -rf /var/lib/apt/lists/* \ + /var/cache/apt/archives/* \ + /var/cache/apt/*.bin \ + /var/log/apt/* \ + /var/log/dpkg.log \ + /tmp/* \ + /var/tmp/* +COPY --from=wheel-builder /out/myelin-worker /usr/local/bin/myelin-node +COPY --from=wheel-builder /wheel/*.whl /tmp/swactor-wheel/ +RUN python3 -m pip install --break-system-packages --no-cache-dir /tmp/swactor-wheel/*.whl && \ + rm -rf /tmp/swactor-wheel && \ + apt-get purge -y --auto-remove python3-pip && \ + apt-get clean && \ + rm -rf /var/lib/apt/lists/* \ + /var/cache/apt/archives/* \ + /var/cache/apt/*.bin \ + /var/log/apt/* \ + /var/log/dpkg.log \ + /tmp/* \ + /var/tmp/* && \ + chmod +x /usr/local/bin/myelin-node diff --git a/apps/myelin/node-image/Dockerfile.e2e b/apps/myelin/node-image/Dockerfile.e2e index 21a271e..89d90a3 100644 --- a/apps/myelin/node-image/Dockerfile.e2e +++ b/apps/myelin/node-image/Dockerfile.e2e @@ -1,50 +1,7 @@ -ARG BASE_IMAGE=myelin-node-base:cuda12.6 -FROM ${BASE_IMAGE} AS wheel-builder -ARG RUST_TOOLCHAIN=nightly-2026-02-07 - -ENV PATH=/root/.cargo/bin:${PATH} -RUN apt-get update && \ - apt-get install -y --no-install-recommends \ - build-essential \ - curl \ - pkg-config \ - python3 \ - python3-dev \ - python3-pip && \ - curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | \ - sh -s -- -y --profile minimal --default-toolchain none && \ - rustup toolchain install "${RUST_TOOLCHAIN}" --profile minimal --component rustc-dev && \ - python3 -m pip install --break-system-packages --no-cache-dir 'maturin>=1.7,<2' && \ - rm -rf /var/lib/apt/lists/* -WORKDIR /workspace -COPY . . -RUN --mount=type=cache,id=myelin-e2e-cargo-registry,target=/root/.cargo/registry \ - --mount=type=cache,id=myelin-e2e-target,target=/workspace/target \ - maturin build \ - --manifest-path crates/bindings/python/Cargo.toml \ - --release \ - --interpreter python3 \ - --out /wheel - +ARG BASE_IMAGE=myelin-node:latest FROM ${BASE_IMAGE} -ARG MYELIN_NODE_BIN=target/release/myelin-worker -RUN apt-get update && \ - apt-get install -y --no-install-recommends python3 python3-pip && \ - apt-get clean && \ - rm -rf /var/lib/apt/lists/* \ - /var/cache/apt/archives/* \ - /var/cache/apt/*.bin \ - /var/log/apt/* \ - /var/log/dpkg.log \ - /tmp/* \ - /var/tmp/* \ - /usr/share/doc/* \ - /usr/share/man/* \ - /usr/share/info/* -COPY ${MYELIN_NODE_BIN} /usr/local/bin/myelin-node -COPY --from=wheel-builder /wheel/*.whl /tmp/swactor-wheel/ -RUN python3 -m pip install --break-system-packages --no-cache-dir /tmp/swactor-wheel/*.whl && \ - rm -rf /tmp/swactor-wheel && \ - chmod +x /usr/local/bin/myelin-node + +# Test-only source injector used by the generated E2E corpus. Production UI +# submissions materialize their selected file through the data namespace. COPY apps/myelin/node-image/e2e_python_launcher.py /usr/local/bin/myelin-e2e-python RUN chmod +x /usr/local/bin/myelin-e2e-python diff --git a/apps/myelin/src/contextual_process.rs b/apps/myelin/src/contextual_process.rs index c73ac13..70d6a78 100644 --- a/apps/myelin/src/contextual_process.rs +++ b/apps/myelin/src/contextual_process.rs @@ -1,12 +1,18 @@ use std::collections::{BTreeMap, HashMap}; -use std::path::PathBuf; +use std::fs::{self, File}; +use std::io::Write; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; -use data_plane::blob_transfer::{BlobTransferReceiver, BlobTransferSender}; +use data_plane::blob_transfer::{ + BlobTransferEvent, BlobTransferId, BlobTransferOffer, BlobTransferReceiver, BlobTransferSender, +}; use data_plane::host::HostRouteRegistrar; -use data_plane::namespace::NamespaceClient; -use data_plane::source::BlobSourcePublisher; +use data_plane::namespace::{ + BlobBinding, DataDirectoryOut, NamespaceClient, NamespaceClientIn, NamespaceRequest, +}; +use data_plane::source::{BlobSourceIn, BlobSourcePublisher}; use data_plane::stream_transport::StreamTransport; use distribution::transport_bridge::{OutboxRouteBinder, RouteBinder, RouteView}; use distribution::types::NodeId; @@ -138,6 +144,11 @@ pub fn build_contextual_process_spawner( Ok(ContextualProcessSpawner::new(config.engine, provisioner)) } +#[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)] +pub(crate) struct ContextualProgramFileWire { + pub namespace_path: String, +} + #[derive(Clone, Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)] pub(crate) struct ContextualProcessSpecWire { pub command: String, @@ -153,6 +164,8 @@ pub(crate) struct ContextualProcessSpecWire { #[serde(default)] pub write_prefixes: Vec, pub attach_timeout_ms: u64, + #[serde(default)] + pub staged_program: Option, } impl ContextualProcessSpecWire { @@ -316,6 +329,14 @@ pub(crate) enum ContextualProcessControllerIn { request_id: String, output: ContextualProcessOutput, }, + ProgramResolved { + request_id: String, + result: Result, + }, + ProgramPrepared { + request_id: String, + result: Result, + }, } struct ContextualOutputRelay { @@ -349,6 +370,239 @@ impl swactor::actor::ActorInterface for ContextualOutputRelay { } } +const PROGRAM_TRANSFER_RETRY: Duration = Duration::from_millis(100); +const PROGRAM_TRANSFER_RETRY_LIMIT: u16 = 300; + +#[derive(Clone)] +pub(crate) struct ContextualProgramMaterializer { + pub namespace_proxy: ActorAddress, + pub receiver: Arc, + pub routes: Arc, + pub engine: EngineHandle, + pub sender: swactor::runtime::ExternalSender, + pub root: PathBuf, +} + +struct PendingProgramSpawn { + spec: ContextualProcessSpecWire, + reply_to: ActorAddress, +} + +struct ProgramNamespaceResolver { + namespace_proxy: ActorAddress, + source_path: data_plane::path::DataPath, + request_id: String, + controller: ActorAddress, +} + +impl swactor::actor::ActorInterface for ProgramNamespaceResolver { + type Incoming = DataDirectoryOut; + type Response = (); + + fn on_start(&mut self, ctx: &swactor::runtime::Ctx<'_>) { + if ctx + .send( + self.namespace_proxy, + NamespaceClientIn::Request { + request: NamespaceRequest::Resolve { + path: self.source_path.clone(), + }, + reply_to: ctx.self_addr(), + }, + ) + .is_err() + { + let _ = ctx.send( + self.controller, + ContextualProcessControllerIn::ProgramResolved { + request_id: self.request_id.clone(), + result: Err("uploaded program namespace is unavailable".to_owned()), + }, + ); + ctx.stop_self(); + } + } + + fn handle(&mut self, ctx: &swactor::runtime::Ctx<'_>, reply: Self::Incoming) { + let result = match reply { + DataDirectoryOut::Resolved { result, .. } => { + result.map_err(|error| format!("resolve uploaded program: {error}")) + } + other => Err(format!( + "resolve uploaded program returned unexpected reply {other:?}" + )), + }; + let _ = ctx.send( + self.controller, + ContextualProcessControllerIn::ProgramResolved { + request_id: self.request_id.clone(), + result, + }, + ); + ctx.stop_self(); + } +} + +struct ProgramFileTransfer { + request_id: String, + controller: ActorAddress, + source: ActorAddress, + length: u64, + transfer_id: BlobTransferId, + receiver: Arc, + routes: Arc, + engine: EngineHandle, + sender: swactor::runtime::ExternalSender, + file: Option, + path: PathBuf, + offer: Option, + written: u64, + source_confirmed: bool, + route_attempts: u16, +} + +impl ProgramFileTransfer { + fn fail(&mut self, ctx: &swactor::runtime::Ctx<'_>, error: impl Into) { + self.finish(ctx, Err(error.into())); + } + + fn finish(&mut self, ctx: &swactor::runtime::Ctx<'_>, result: Result) { + if let Some(offer) = self.offer.take() { + self.receiver.cancel(&offer); + } + self.file.take(); + if result.is_err() { + remove_program_tree(&self.path); + } + let _ = ctx.send( + self.controller, + ContextualProcessControllerIn::ProgramPrepared { + request_id: self.request_id.clone(), + result, + }, + ); + ctx.stop_self(); + } + + fn try_start(&mut self, ctx: &swactor::runtime::Ctx<'_>) { + if self.route_attempts >= PROGRAM_TRANSFER_RETRY_LIMIT { + self.fail( + ctx, + "uploaded program source did not become routable before the transfer deadline", + ); + return; + } + self.route_attempts += 1; + if self.routes.is_routable(self.source) { + let offer = self + .offer + .as_ref() + .expect("program transfer retains its offer") + .clone(); + if ctx + .send(self.source, BlobSourceIn::BeginTransfer { offer }) + .is_err() + { + self.fail(ctx, "route to uploaded program source is unavailable"); + return; + } + } + if !self.source_confirmed { + self.engine.send_after( + PROGRAM_TRANSFER_RETRY, + self.sender.clone(), + ctx.self_addr(), + BlobTransferEvent::RouteRetry, + ); + } + } +} + +impl swactor::actor::ActorInterface for ProgramFileTransfer { + type Incoming = BlobTransferEvent; + type Response = (); + + fn on_start(&mut self, ctx: &swactor::runtime::Ctx<'_>) { + if self.length == 0 { + self.finish(ctx, Ok(self.path.clone())); + return; + } + match self.receiver.open(ctx.self_addr(), self.transfer_id) { + Ok(offer) => { + self.offer = Some(offer); + self.try_start(ctx); + } + Err(error) => self.fail(ctx, format!("open uploaded program transfer: {error}")), + } + } + + fn handle(&mut self, ctx: &swactor::runtime::Ctx<'_>, event: Self::Incoming) { + match event { + BlobTransferEvent::RouteRetry if !self.source_confirmed => self.try_start(ctx), + BlobTransferEvent::Chunk { transfer_id, bytes } if transfer_id == self.transfer_id => { + self.source_confirmed = true; + let Some(next) = self + .written + .checked_add(bytes.len() as u64) + .filter(|written| *written <= self.length) + else { + self.fail( + ctx, + format!("uploaded program exceeded declared length {}", self.length), + ); + return; + }; + if let Err(error) = self + .file + .as_mut() + .expect("live program transfer owns its file") + .write_all(&bytes) + { + self.fail(ctx, format!("write uploaded program: {error}")); + return; + } + self.written = next; + } + BlobTransferEvent::Finished { transfer_id } if transfer_id == self.transfer_id => { + self.source_confirmed = true; + if self.written != self.length { + self.fail( + ctx, + format!( + "uploaded program length mismatch: expected {}, received {}", + self.length, self.written + ), + ); + return; + } + if let Err(error) = self + .file + .as_mut() + .expect("live program transfer owns its file") + .flush() + { + self.fail(ctx, format!("flush uploaded program: {error}")); + return; + } + self.finish(ctx, Ok(self.path.clone())); + } + BlobTransferEvent::Failed { + transfer_id, + reason, + } if transfer_id == self.transfer_id => { + self.fail(ctx, format!("receive uploaded program: {reason}")); + } + _ => {} + } + } +} + +fn remove_program_tree(path: &Path) { + if let Some(parent) = path.parent() { + let _ = fs::remove_dir_all(parent); + } +} + struct LiveContextualExecution { request_id: String, process: ActorAddress, @@ -356,12 +610,15 @@ struct LiveContextualExecution { reply_to: ActorAddress, started_pid: Option, context_ready: bool, + staged_program: Option, } pub(crate) struct ContextualProcessController { logical_node_id: u64, spawner: Arc, sender: swactor::runtime::ExternalSender, + materializer: Option, + pending_programs: HashMap, by_process: HashMap, process_by_request: HashMap, } @@ -376,10 +633,19 @@ impl ContextualProcessController { logical_node_id, spawner, sender, + materializer: None, + pending_programs: HashMap::new(), by_process: HashMap::new(), process_by_request: HashMap::new(), } } + pub(crate) fn with_program_materializer( + mut self, + materializer: ContextualProgramMaterializer, + ) -> Self { + self.materializer = Some(materializer); + self + } fn emit( &self, @@ -404,10 +670,12 @@ impl ContextualProcessController { &mut self, ctx: &swactor::runtime::Ctx<'_>, request_id: String, - spec: ContextualProcessSpecWire, + mut spec: ContextualProcessSpecWire, reply_to: ActorAddress, ) { - if self.process_by_request.contains_key(&request_id) { + if self.process_by_request.contains_key(&request_id) + || self.pending_programs.contains_key(&request_id) + { self.emit( ctx, reply_to, @@ -418,9 +686,70 @@ impl ContextualProcessController { ); return; } + let Some(staged_program) = spec.staged_program.take() else { + self.spawn_ready(ctx, request_id, spec, reply_to, None); + return; + }; + let Some(materializer) = self.materializer.clone() else { + self.emit( + ctx, + reply_to, + request_id, + ContextualProcessEventKindWire::SpawnRejected { + error: "uploaded program materialization is unavailable on this node" + .to_owned(), + }, + ); + return; + }; + let source_path = match data_plane::path::DataPath::parse(staged_program.namespace_path) { + Ok(path) => path, + Err(error) => { + self.emit( + ctx, + reply_to, + request_id, + ContextualProcessEventKindWire::SpawnRejected { + error: format!("invalid uploaded program path: {error}"), + }, + ); + return; + } + }; + self.pending_programs + .insert(request_id.clone(), PendingProgramSpawn { spec, reply_to }); + if let Err(error) = ctx.spawn(ProgramNamespaceResolver { + namespace_proxy: materializer.namespace_proxy, + source_path, + request_id: request_id.clone(), + controller: ctx.self_addr(), + }) { + self.pending_programs.remove(&request_id); + self.emit( + ctx, + reply_to, + request_id, + ContextualProcessEventKindWire::SpawnRejected { + error: format!("spawn uploaded program resolver: {error}"), + }, + ); + } + } + + fn spawn_ready( + &mut self, + ctx: &swactor::runtime::Ctx<'_>, + request_id: String, + spec: ContextualProcessSpecWire, + reply_to: ActorAddress, + staged_program: Option, + ) { let spec = match spec.into_spec() { Ok(spec) => spec, Err(error) => { + if let Some(path) = staged_program.as_deref() { + remove_program_tree(path); + } self.emit( ctx, reply_to, @@ -436,6 +765,9 @@ impl ContextualProcessController { }) { Ok(relay) => relay, Err(error) => { + if let Some(path) = staged_program.as_deref() { + remove_program_tree(path); + } self.emit( ctx, reply_to, @@ -461,6 +793,7 @@ impl ContextualProcessController { reply_to, started_pid: None, context_ready: false, + staged_program, }, ); self.emit( @@ -475,6 +808,9 @@ impl ContextualProcessController { } Err(error) => { let _ = ctx.stop_actor(relay); + if let Some(path) = staged_program.as_deref() { + remove_program_tree(path); + } self.emit( ctx, reply_to, @@ -487,6 +823,129 @@ impl ContextualProcessController { } } + fn program_resolved( + &mut self, + ctx: &swactor::runtime::Ctx<'_>, + request_id: String, + result: Result, + ) { + let Some(pending) = self.pending_programs.get(&request_id) else { + return; + }; + let reply_to = pending.reply_to; + let binding = match result { + Ok(binding) => binding, + Err(error) => { + let pending = self + .pending_programs + .remove(&request_id) + .expect("pending uploaded program exists"); + self.emit( + ctx, + pending.reply_to, + request_id, + ContextualProcessEventKindWire::SpawnRejected { error }, + ); + return; + } + }; + let Some(materializer) = self.materializer.clone() else { + return; + }; + let path = program_path(&materializer.root, &request_id); + let file = path + .parent() + .ok_or_else(|| "uploaded program path has no parent".to_owned()) + .and_then(|parent| { + fs::create_dir_all(parent) + .map_err(|error| format!("create uploaded program directory: {error}")) + }) + .and_then(|()| { + File::create(&path) + .map_err(|error| format!("create uploaded program file: {error}")) + }); + let file = match file { + Ok(file) => file, + Err(error) => { + self.pending_programs.remove(&request_id); + remove_program_tree(&path); + self.emit( + ctx, + reply_to, + request_id, + ContextualProcessEventKindWire::SpawnRejected { error }, + ); + return; + } + }; + let random = ActorAddress::new_random(); + let transfer_id = BlobTransferId(u64::from_le_bytes( + random.0[..8] + .try_into() + .expect("actor address contains eight transfer ID bytes"), + )); + let transfer = ProgramFileTransfer { + request_id: request_id.clone(), + controller: ctx.self_addr(), + source: binding.source, + length: binding.length, + transfer_id, + receiver: materializer.receiver, + routes: materializer.routes, + engine: materializer.engine, + sender: materializer.sender, + file: Some(file), + path: path.clone(), + offer: None, + written: 0, + source_confirmed: false, + route_attempts: 0, + }; + if let Err(error) = ctx.spawn(transfer) { + self.pending_programs.remove(&request_id); + remove_program_tree(&path); + self.emit( + ctx, + reply_to, + request_id, + ContextualProcessEventKindWire::SpawnRejected { + error: format!("spawn uploaded program transfer: {error}"), + }, + ); + } + } + + fn program_prepared( + &mut self, + ctx: &swactor::runtime::Ctx<'_>, + request_id: String, + result: Result, + ) { + let Some(mut pending) = self.pending_programs.remove(&request_id) else { + if let Ok(path) = result { + remove_program_tree(&path); + } + return; + }; + let path = match result { + Ok(path) => path, + Err(error) => { + self.emit( + ctx, + pending.reply_to, + request_id, + ContextualProcessEventKindWire::SpawnRejected { error }, + ); + return; + } + }; + pending.spec.args.push(path.display().to_string()); + if pending.spec.working_dir.is_none() { + pending.spec.working_dir = path.parent().map(|parent| parent.display().to_string()); + } + self.spawn_ready(ctx, request_id, pending.spec, pending.reply_to, Some(path)); + } + fn stop( &self, ctx: &swactor::runtime::Ctx<'_>, @@ -586,12 +1045,21 @@ impl ContextualProcessController { }; self.emit(ctx, reply_to, request_id.clone(), event); if terminal { - self.by_process.remove(&process); + if let Some(execution) = self.by_process.remove(&process) + && let Some(path) = execution.staged_program + { + remove_program_tree(&path); + } self.process_by_request.remove(&request_id); } } } +fn program_path(root: &Path, request_id: &str) -> PathBuf { + let digest = blake3::hash(request_id.as_bytes()).to_hex(); + root.join(digest.as_str()).join("program.py") +} + fn bootstrap_failure(failure: &BootstrapFailure) -> String { match failure { BootstrapFailure::Provisioning(error) @@ -632,6 +1100,12 @@ impl swactor::actor::ActorInterface for ContextualProcessController { ContextualProcessControllerIn::Observed { request_id, output } => { self.observe(ctx, request_id, output); } + ContextualProcessControllerIn::ProgramResolved { request_id, result } => { + self.program_resolved(ctx, request_id, result); + } + ContextualProcessControllerIn::ProgramPrepared { request_id, result } => { + self.program_prepared(ctx, request_id, result); + } } } } diff --git a/apps/myelin/src/data_namespace.rs b/apps/myelin/src/data_namespace.rs index 3ca1087..0ebd7d1 100644 --- a/apps/myelin/src/data_namespace.rs +++ b/apps/myelin/src/data_namespace.rs @@ -154,7 +154,6 @@ impl DataNamespaceAuthority { Ok(Self { service }) } - #[cfg(test)] pub(crate) fn directory(&self) -> ActorAddress { self.service.directory() } diff --git a/apps/myelin/src/node/worker_node_runtime.rs b/apps/myelin/src/node/worker_node_runtime.rs index aa44ca7..d2c61b3 100644 --- a/apps/myelin/src/node/worker_node_runtime.rs +++ b/apps/myelin/src/node/worker_node_runtime.rs @@ -23,7 +23,8 @@ use telemetry::{ use crate::codecs::register_myelin_actor_codecs; use crate::contextual_process::{ - ContextualProcessController, MyelinContextualProcessConfig, build_contextual_process_spawner, + ContextualProcessController, ContextualProgramMaterializer, MyelinChildRouteRegistrar, + MyelinContextualProcessConfig, build_contextual_process_spawner, }; use crate::data_namespace::install_namespace_client; use crate::gguf_shard::{StageShardPlan, materialize_stage_shard_http, validate_stage_shard_cache}; @@ -45,6 +46,7 @@ use data_plane::arena; use data_plane::blob_transfer::{BlobTransferReceiver, BlobTransferSender}; use data_plane::edge_lifecycle as edge; use data_plane::edge_runtime; +use data_plane::host::HostRouteRegistrar; use data_plane::object_record as ingress; use distribution::node::DistributedNodeConfig; use distribution::telemetry::{MembershipTransition, SwimProbeEvent}; @@ -1952,7 +1954,7 @@ fn run() -> Result<(), String> { "ready", json!({"actor":orchestrator,"source":orchestrator_source}), )?; - let contextual_spawner = if config.agent_only { + let contextual_services = if config.agent_only { let namespace = install_namespace_client(&stack, &driver)?; let blob_receiver = Arc::new(IrohBlobTransferReceiver::new( driver.endpoint_addr(), @@ -1965,14 +1967,14 @@ fn run() -> Result<(), String> { &engine.handle(), stack.runtime.clone(), )); - Some(Arc::new(build_contextual_process_spawner( + let spawner = Arc::new(build_contextual_process_spawner( MyelinContextualProcessConfig { runtime: stack.runtime.clone(), engine: engine.handle(), arena_bytes: config.arena_bytes, arena_alignment: config.arena_alignment, - namespace: Some(namespace.client), - transfer_receiver: Some(transfer_receiver), + namespace: Some(namespace.client.clone()), + transfer_receiver: Some(Arc::clone(&transfer_receiver)), source_sender: Some(source_sender), source_publisher: Some(namespace.source_publisher), route_view: stack.route_view.clone(), @@ -1981,18 +1983,35 @@ fn run() -> Result<(), String> { stream_transport: Some(driver.stream_transport()), host_endpoint: driver.endpoint_addr(), }, - )?)) + )?); + let routes: Arc = Arc::new(MyelinChildRouteRegistrar::new( + stack.route_view.clone(), + stack.pinned_routes.clone(), + stack.route_binder.clone(), + )); + let materializer = ContextualProgramMaterializer { + namespace_proxy: namespace.client.proxy(), + receiver: transfer_receiver, + routes, + engine: engine.handle(), + sender: stack.runtime.create_sender(), + root: std::env::temp_dir().join("myelin-contextual"), + }; + Some((spawner, materializer)) } else { None }; - let contextual_controller = contextual_spawner + let contextual_controller = contextual_services .as_ref() - .map(|spawner| { - stack.runtime.spawn(ContextualProcessController::new( - config.logical_node_id, - Arc::clone(spawner), - stack.runtime.create_sender(), - )) + .map(|(spawner, materializer)| { + stack.runtime.spawn( + ContextualProcessController::new( + config.logical_node_id, + Arc::clone(spawner), + stack.runtime.create_sender(), + ) + .with_program_materializer(materializer.clone()), + ) }) .transpose() .map_err(|error| format!("spawn contextual process controller: {error}"))?; diff --git a/apps/myelin/src/orchestration/actor.rs b/apps/myelin/src/orchestration/actor.rs index 7979007..673e7aa 100644 --- a/apps/myelin/src/orchestration/actor.rs +++ b/apps/myelin/src/orchestration/actor.rs @@ -1,4 +1,5 @@ use std::collections::{HashMap, VecDeque}; +use std::path::PathBuf; use iroh::EndpointAddr; use serde::{Deserialize, Serialize}; @@ -15,6 +16,79 @@ use crate::orchestration::manual_control::{ManualActorControl, ManualControlMsg} use crate::run_fsm as core; use swactor_transport::JsonCodec; +#[derive(Clone)] +pub(crate) struct ContextualArtifactCleanup { + directory: ActorAddress, + upload_root: PathBuf, +} + +impl ContextualArtifactCleanup { + pub(crate) fn new(directory: ActorAddress, upload_root: PathBuf) -> Self { + Self { + directory, + upload_root, + } + } + + fn cleanup(&self, ctx: &Ctx<'_>, request_id: &str, namespace_path: String) { + let Ok(namespace_path) = data_plane::path::DataPath::parse(namespace_path) else { + return; + }; + let host_path = self.upload_root.join(format!("{request_id}.py")); + let random = ActorAddress::new_random(); + let operation_id = data_plane::namespace::OperationId::from_u128(u128::from_le_bytes( + random.0[..16] + .try_into() + .expect("actor address contains sixteen operation ID bytes"), + )); + if ctx + .spawn(ContextualArtifactCleanupOperation { + directory: self.directory, + namespace_path, + operation_id, + }) + .is_ok() + { + // The source actor already owns an open descriptor. Removing the + // name after process termination cannot invalidate a transfer, + // and avoids retaining completed UI uploads while retirement waits. + let _ = std::fs::remove_file(host_path); + } + } +} + +struct ContextualArtifactCleanupOperation { + directory: ActorAddress, + namespace_path: data_plane::path::DataPath, + operation_id: data_plane::namespace::OperationId, +} + +impl ActorInterface for ContextualArtifactCleanupOperation { + type Incoming = data_plane::namespace::DataDirectoryOut; + type Response = (); + + fn on_start(&mut self, ctx: &Ctx<'_>) { + let request_id = data_plane::namespace::DirectoryRequestId(1); + if ctx + .send( + self.directory, + data_plane::namespace::DataDirectoryIn::Unregister { + request_id, + path: self.namespace_path.clone(), + operation_id: self.operation_id, + reply_to: ctx.self_addr(), + }, + ) + .is_err() + { + ctx.stop_self(); + } + } + + fn handle(&mut self, ctx: &Ctx<'_>, _reply: Self::Incoming) { + ctx.stop_self(); + } +} #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct StageRefWire { @@ -267,6 +341,7 @@ struct ContextualExecutionState { next_sequence: u64, events: VecDeque, spawn_waiter: Option, + staged_source: Option, } impl ContextualExecutionState { @@ -320,6 +395,7 @@ pub(crate) struct OrchestratorActor { manual: Option, contextual_executions: HashMap, pending_contextual_replies: HashMap, + contextual_artifact_cleanup: Option, } impl OrchestratorActor { @@ -332,6 +408,7 @@ impl OrchestratorActor { manual: None, contextual_executions: HashMap::new(), pending_contextual_replies: HashMap::new(), + contextual_artifact_cleanup: None, } } @@ -339,6 +416,19 @@ impl OrchestratorActor { self.manual = Some(manual); self } + pub(crate) fn with_contextual_artifact_cleanup( + mut self, + cleanup: ContextualArtifactCleanup, + ) -> Self { + self.contextual_artifact_cleanup = Some(cleanup); + self + } + + fn cleanup_contextual_artifact(&self, ctx: &Ctx<'_>, request_id: &str, source: Option) { + if let (Some(cleanup), Some(source)) = (&self.contextual_artifact_cleanup, source) { + cleanup.cleanup(ctx, request_id, source); + } + } fn contextual_rejection( &self, @@ -397,6 +487,13 @@ impl OrchestratorActor { .retain(|_, execution| !execution.terminal); } if self.contextual_executions.len() >= MAX_CONTEXTUAL_EXECUTIONS { + self.cleanup_contextual_artifact( + ctx, + request_id, + spec.staged_program + .as_ref() + .map(|program| program.namespace_path.clone()), + ); self.contextual_rejection( ctx, *reply_to, @@ -413,6 +510,10 @@ impl OrchestratorActor { next_sequence: 0, events: VecDeque::new(), spawn_waiter: Some(*reply_to), + staged_source: spec + .staged_program + .as_ref() + .map(|program| program.namespace_path.clone()), }, ); if let Err(error) = self.route_contextual( @@ -424,7 +525,11 @@ impl OrchestratorActor { reply_to: ctx.self_addr(), }, ) { - self.contextual_executions.remove(request_id); + let staged_source = self + .contextual_executions + .remove(request_id) + .and_then(|execution| execution.staged_source); + self.cleanup_contextual_artifact(ctx, request_id, staged_source); self.contextual_rejection(ctx, *reply_to, error); } true @@ -562,10 +667,19 @@ impl OrchestratorActor { ); return true; } - if let Some(target) = pending.target_request_id - && let Some(execution) = self.contextual_executions.get_mut(&target) - { - execution.record(observation.clone()); + if let Some(target) = pending.target_request_id { + let staged_source = + self.contextual_executions + .get_mut(&target) + .and_then(|execution| { + execution.record(observation.clone()); + observation + .event + .is_terminal() + .then(|| execution.staged_source.take()) + .flatten() + }); + self.cleanup_contextual_artifact(ctx, &target, staged_source); } let _ = ctx.send( pending.reply_to, @@ -582,6 +696,7 @@ impl OrchestratorActor { if execution.logical_node_id != observation.logical_node_id { let expected = execution.logical_node_id; let waiter = execution.spawn_waiter.take(); + let staged_source = execution.staged_source.take(); execution.terminal = true; if let Some(waiter) = waiter { let _ = ctx.send( @@ -594,6 +709,7 @@ impl OrchestratorActor { }, ); } + self.cleanup_contextual_artifact(ctx, &observation.request_id, staged_source); return true; } let resolves_spawn = matches!( @@ -601,6 +717,11 @@ impl OrchestratorActor { ContextualProcessEventKindWire::Spawned { .. } ) || observation.event.is_terminal(); execution.record(observation.clone()); + let staged_source = observation + .event + .is_terminal() + .then(|| execution.staged_source.take()) + .flatten(); if resolves_spawn && let Some(waiter) = execution.spawn_waiter.take() { let _ = ctx.send( waiter, @@ -609,6 +730,7 @@ impl OrchestratorActor { }, ); } + self.cleanup_contextual_artifact(ctx, &observation.request_id, staged_source); true } OrchestratorMsg::ContextualControlCancel { control_request_id } => { @@ -623,6 +745,7 @@ impl OrchestratorActor { fn fail_contextual_node(&mut self, ctx: &Ctx<'_>, logical_node_id: u64, reason: &str) { let mut waiters = Vec::new(); + let mut artifacts = Vec::new(); for (request_id, execution) in &mut self.contextual_executions { if execution.logical_node_id != logical_node_id || execution.terminal { continue; @@ -635,10 +758,16 @@ impl OrchestratorActor { }, }; execution.record(observation.clone()); + if let Some(source) = execution.staged_source.take() { + artifacts.push((request_id.clone(), source)); + } if let Some(waiter) = execution.spawn_waiter.take() { waiters.push((waiter, observation)); } } + for (request_id, source) in artifacts { + self.cleanup_contextual_artifact(ctx, &request_id, Some(source)); + } for (waiter, observation) in waiters { let _ = ctx.send(waiter, ContextualControlReply::Event { observation }); } diff --git a/apps/myelin/src/orchestration/app.rs b/apps/myelin/src/orchestration/app.rs index 9e78f78..3acdddb 100644 --- a/apps/myelin/src/orchestration/app.rs +++ b/apps/myelin/src/orchestration/app.rs @@ -14,7 +14,9 @@ use crate::observability::frame_collector::FrameCollector; use crate::observability::orch_telemetry::{ BootstrapEmission, DashboardSupport, MYELIN_SWIM_MEMBERSHIP, OrchTelemetry, }; -use crate::orchestration::actor::{OrchestratorActor, OrchestratorMsg, OrchestratorReport}; +use crate::orchestration::actor::{ + ContextualArtifactCleanup, OrchestratorActor, OrchestratorMsg, OrchestratorReport, +}; use crate::orchestration::config::{DEFAULT_CONFIG_PATH, TomlConfigOverlay}; use crate::orchestration::control; use crate::orchestration::daemon; @@ -234,6 +236,7 @@ where .finalize()?; let state_dir = daemon::StateDir::new(config.state_dir.clone()); let data_namespace_path = config.state_dir.join("data-namespace.json"); + let contextual_upload_root = config.state_dir.join("contextual-uploads"); if config.reset_state { state_dir.reset()?; } @@ -690,7 +693,11 @@ where }, Some(orchestrator_report_actor), ) - .with_manual_control(manual), + .with_manual_control(manual) + .with_contextual_artifact_cleanup(ContextualArtifactCleanup::new( + data_namespace.directory(), + contextual_upload_root.clone(), + )), ) { Ok(actor) => actor, Err(error) => return Err(format!("spawn orchestrator actor: {error}")), @@ -719,8 +726,13 @@ where .map_err(|error| format!("route initial provider validation: {error}"))?; } - let control_plugin = - control::plugin(stack.runtime.clone(), engine.handle(), orchestrator_actor); + let control_plugin = control::plugin( + stack.runtime.clone(), + engine.handle(), + orchestrator_actor, + data_namespace.control(), + contextual_upload_root, + ); dashboard = DashboardSupport::start_with_plugins( config.dashboard, &engine.handle(), diff --git a/apps/myelin/src/orchestration/control.rs b/apps/myelin/src/orchestration/control.rs index 6fb3bf4..59137ad 100644 --- a/apps/myelin/src/orchestration/control.rs +++ b/apps/myelin/src/orchestration/control.rs @@ -1,7 +1,10 @@ +use std::collections::BTreeMap; +use std::path::PathBuf; use std::sync::{Arc, Mutex}; use std::time::Duration; -use axum::extract::{Path, Query, State}; +use axum::body::Bytes; +use axum::extract::{DefaultBodyLimit, Path, Query, State}; use axum::http::{StatusCode, header}; use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; @@ -12,7 +15,7 @@ use swactor::runtime::{Ctx, ExternalSender, Runtime}; use swactor_engine::EngineHandle; use swactor_vastai::VastClient; -use crate::contextual_process::ContextualProcessSpecWire; +use crate::contextual_process::{ContextualProcessSpecWire, ContextualProgramFileWire}; use crate::orchestration::actor::ContextualControlReply; use crate::orchestration::actor::OrchestratorMsg; use crate::orchestration::manual_control::{ @@ -24,6 +27,9 @@ const CONTROL_REPLY_TIMEOUT: Duration = Duration::from_secs(2); const OFFER_SEARCH_REPLY_MARGIN: Duration = Duration::from_secs(5); const OFFER_SEARCH_REPLY_TIMEOUT: Duration = VastClient::REQUEST_TIMEOUT.saturating_add(OFFER_SEARCH_REPLY_MARGIN); +const PROGRAM_UPLOAD_LIMIT: usize = 256 * 1024; +const PROGRAM_ATTACH_TIMEOUT: Duration = Duration::from_secs(30); +const PROGRAM_SPAWN_REPLY_TIMEOUT: Duration = Duration::from_secs(35); struct ControlReplyObserver { reply: Arc>>>, engine: EngineHandle, @@ -110,17 +116,23 @@ struct ControlHttpState { runtime: Runtime, engine: EngineHandle, orchestrator: ActorAddress, + namespace: Option, + upload_root: Arc, } pub(crate) fn plugin( runtime: Runtime, engine: EngineHandle, orchestrator: ActorAddress, + namespace: data_plane::control::DataPlaneControl, + upload_root: PathBuf, ) -> dashboard::DashboardPlugin { let state = ControlHttpState { runtime, engine, orchestrator, + namespace: Some(namespace), + upload_root: Arc::new(upload_root), }; let routes = Router::new() .route(FLEET_CONTROL_SCRIPT_URL, get(fleet_control_script)) @@ -134,6 +146,10 @@ pub(crate) fn plugin( .route("/api/control/flush", post(flush)) .route("/api/control/nodes/{logical_node_id}/kill", post(kill_path)) .route("/api/control/contextual/spawn", post(contextual_spawn)) + .route( + "/api/control/contextual/nodes/{logical_node_id}/python", + post(contextual_python), + ) .route( "/api/control/contextual/{request_id}/events", get(contextual_events), @@ -146,6 +162,7 @@ pub(crate) fn plugin( "/api/control/contextual/nodes/{logical_node_id}", get(contextual_query_node), ) + .layer(DefaultBodyLimit::max(PROGRAM_UPLOAD_LIMIT)) .with_state(state); dashboard::DashboardPlugin::new(routes).with_page(dashboard::PluginPage::new( "provision", @@ -299,6 +316,166 @@ async fn contextual_spawn( }) .await } +#[derive(serde::Deserialize)] +struct ContextualPythonQuery { + filename: String, +} + +async fn contextual_python( + Path(logical_node_id): Path, + Query(query): Query, + State(state): State, + body: Bytes, +) -> Response { + let Some(namespace) = state.namespace.as_ref() else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(ErrorResponse { + error: "uploaded program control is unavailable".to_owned(), + }), + ) + .into_response(); + }; + let filename = query.filename.trim(); + if filename.is_empty() + || filename + != PathBuf::from(filename) + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("") + || !filename.to_ascii_lowercase().ends_with(".py") + { + return ( + StatusCode::BAD_REQUEST, + Json(ErrorResponse { + error: "select one Python file with a .py extension".to_owned(), + }), + ) + .into_response(); + } + if body.is_empty() { + return ( + StatusCode::BAD_REQUEST, + Json(ErrorResponse { + error: "selected Python file is empty".to_owned(), + }), + ) + .into_response(); + } + if body.len() > PROGRAM_UPLOAD_LIMIT { + return ( + StatusCode::PAYLOAD_TOO_LARGE, + Json(ErrorResponse { + error: format!( + "selected Python file exceeds the {} byte limit", + PROGRAM_UPLOAD_LIMIT + ), + }), + ) + .into_response(); + } + + let request_id = format!("ui-{}", ActorAddress::new_random().to_full_hex()); + let host_path = state.upload_root.join(format!("{request_id}.py")); + if let Err(error) = write_uploaded_program(&state, host_path.clone(), body).await { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse { error }), + ) + .into_response(); + } + let namespace_path = format!("/myelin/contextual/{request_id}/program.py"); + let data_path = data_plane::path::DataPath::parse(namespace_path.clone()) + .expect("generated contextual program path is valid"); + if let Err(error) = namespace + .register(data_path.clone(), data_plane::blob::file(&host_path)) + .await + { + remove_uploaded_program(&state, host_path); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse { + error: format!("register uploaded program: {error}"), + }), + ) + .into_response(); + } + + let spec = ContextualProcessSpecWire { + command: "python3".to_owned(), + args: Vec::new(), + env: BTreeMap::new(), + working_dir: None, + label: Some(filename.to_owned()), + execution_id: request_id.clone(), + read_prefixes: Vec::new(), + write_prefixes: Vec::new(), + attach_timeout_ms: PROGRAM_ATTACH_TIMEOUT.as_millis() as u64, + staged_program: Some(ContextualProgramFileWire { + namespace_path: namespace_path.clone(), + }), + }; + let response_rx = match begin_contextual_request_reply( + &state, + None, + PROGRAM_SPAWN_REPLY_TIMEOUT, + |reply_to| OrchestratorMsg::ContextualSpawn { + logical_node_id, + request_id, + spec, + reply_to, + }, + ) { + Ok(response_rx) => response_rx, + Err(response) => { + cleanup_uploaded_program(&state, data_path, host_path).await; + return *response; + } + }; + contextual_response(response_rx).await +} + +async fn write_uploaded_program( + state: &ControlHttpState, + path: PathBuf, + body: Bytes, +) -> Result<(), String> { + let (tx, rx) = tokio::sync::oneshot::channel(); + let root = Arc::clone(&state.upload_root); + let work = Box::new(move || { + let result = std::fs::create_dir_all(root.as_ref()) + .and_then(|()| std::fs::write(&path, &body)) + .map_err(|error| format!("store uploaded program: {error}")); + let _ = tx.send(result); + }); + state + .engine + .blocking_work_sender() + .submit(work) + .map_err(|_| "execution engine stopped before storing uploaded program".to_owned())?; + rx.await + .map_err(|error| format!("uploaded program writer stopped: {error}"))? +} + +fn remove_uploaded_program(state: &ControlHttpState, path: PathBuf) { + let _ = state + .engine + .blocking_work_sender() + .submit(Box::new(move || { + let _ = std::fs::remove_file(path); + })); +} + +async fn cleanup_uploaded_program( + state: &ControlHttpState, + data_path: data_plane::path::DataPath, + host_path: PathBuf, +) { + if let Some(namespace) = &state.namespace { + let _ = namespace.unregister(data_path).await; + } + remove_uploaded_program(state, host_path); +} #[derive(Default, serde::Deserialize)] struct ContextualEventsQuery { @@ -416,10 +593,17 @@ async fn contextual_request_reply( cancel_key: Option, build: impl FnOnce(ActorAddress) -> OrchestratorMsg, ) -> Response { - let response_rx = match begin_contextual_request_reply(state, cancel_key, build) { - Ok(response_rx) => response_rx, - Err(response) => return *response, - }; + let response_rx = + match begin_contextual_request_reply(state, cancel_key, CONTROL_REPLY_TIMEOUT, build) { + Ok(response_rx) => response_rx, + Err(response) => return *response, + }; + contextual_response(response_rx).await +} + +async fn contextual_response( + response_rx: tokio::sync::oneshot::Receiver, +) -> Response { match response_rx.await { Ok(ContextualControlReply::TimedOut) => ( StatusCode::GATEWAY_TIMEOUT, @@ -445,6 +629,7 @@ async fn contextual_request_reply( fn begin_contextual_request_reply( state: &ControlHttpState, cancel_key: Option, + timeout: Duration, build: impl FnOnce(ActorAddress) -> OrchestratorMsg, ) -> Result, Box> { let (response_tx, response_rx) = tokio::sync::oneshot::channel(); @@ -455,7 +640,7 @@ fn begin_contextual_request_reply( reply: response_tx, engine: state.engine.clone(), sender: state.runtime.create_sender(), - timeout: CONTROL_REPLY_TIMEOUT, + timeout, orchestrator: state.orchestrator, cancel_key, }) @@ -935,6 +1120,8 @@ mod properties { runtime: runtime.clone(), engine: engine.handle(), orchestrator, + namespace: None, + upload_root: Arc::new(PathBuf::new()), }; let outcome = (|| { let mut responses = Vec::with_capacity(actions.len()); @@ -1081,6 +1268,8 @@ mod properties { runtime: runtime.clone(), engine: engine.handle(), orchestrator: ActorAddress::default(), + namespace: None, + upload_root: Arc::new(PathBuf::new()), }; let response = match begin_request_reply(&state, Duration::from_millis(1), |reply_to| { ManualControlMsg::Query { reply_to } @@ -1122,6 +1311,8 @@ mod properties { runtime: runtime.clone(), engine: engine.handle(), orchestrator, + namespace: None, + upload_root: Arc::new(PathBuf::new()), }; let pending = begin_http_action(&state, 0, HttpAction::Status); drive_steps(&backend, STEP_BUDGET); @@ -1164,6 +1355,8 @@ mod properties { runtime, engine: engine.handle(), orchestrator: ActorAddress::default(), + namespace: None, + upload_root: Arc::new(PathBuf::new()), }; let actions = vec![HttpAction::Status]; let responses = vec![HttpObservation { diff --git a/apps/myelin/src/orchestration/fleet_control.js b/apps/myelin/src/orchestration/fleet_control.js index b7f882e..9b7496f 100644 --- a/apps/myelin/src/orchestration/fleet_control.js +++ b/apps/myelin/src/orchestration/fleet_control.js @@ -1,11 +1,13 @@ (() => { const NODE_CONTROL_ID = 'myelin-fleet-control'; + const EXECUTION_CONTROL_ID = 'myelin-contextual-execution'; const BULK_CONTROL_ID = 'myelin-fleet-bulk-control'; const CONFIRM_ID = 'myelin-confirm-dialog'; const STYLE_ID = 'myelin-fleet-control-style'; const selectedNodes = new Set(); let lastModel = null; let syncing = false; + let executionPoll = null; const page = document.getElementById('page'); if (!page) return; @@ -22,6 +24,11 @@ .myelin-bulk-control { width:fit-content;margin:0 0 12px auto;padding:6px 8px;border:1px solid var(--divider);background:transparent;border-radius:var(--r) } .node-card[data-bulk-selected="true"] { border-color:var(--bad);background:var(--selected);box-shadow:inset 3px 0 0 var(--bad) } .node-card[data-bulk-killable="true"] { user-select:none } + .myelin-execution { display:grid;gap:8px;width:min(720px,100%);padding:10px 12px;border:1px solid var(--divider);border-radius:var(--r);background:var(--ghost) } + .myelin-execution-row { display:flex;align-items:center;flex-wrap:wrap;gap:8px } + .myelin-execution input[type="file"] { max-width:100%;color:var(--text);font:12px var(--mono) } + .myelin-execution-output { display:none;max-height:260px;overflow:auto;margin:0;padding:9px;background:var(--inset);border:1px solid var(--divider);white-space:pre-wrap;word-break:break-word;color:var(--text);font:12px/1.45 var(--mono) } + .myelin-execution-output[data-active="true"] { display:block } .myelin-confirm { width:min(440px,calc(100vw - 32px));padding:0;color:var(--text);background:var(--panel);border:1px solid var(--bad);border-radius:var(--r);box-shadow:0 18px 60px rgba(0,0,0,.55) } .myelin-confirm::backdrop { background:rgba(0,6,12,.78) } .myelin-confirm form { display:grid;gap:14px;padding:18px } @@ -206,6 +213,130 @@ message.textContent = resultMessage(results); } + function stopExecutionPoll() { + if (executionPoll?.timer) clearTimeout(executionPoll.timer); + executionPoll = null; + } + + function appendExecutionEvent(panel, observation) { + const event = observation?.event || {}; + const type = event.type || 'unknown'; + const status = panel.querySelector('[data-execution-status]'); + const output = panel.querySelector('[data-execution-output]'); + if (type === 'stdout' || type === 'stderr') { + const bytes = Array.isArray(event.bytes) ? new Uint8Array(event.bytes) : new Uint8Array(); + output.dataset.active = 'true'; + output.textContent += new TextDecoder().decode(bytes); + output.scrollTop = output.scrollHeight; + } else if (type === 'context_ready') { + status.textContent = 'Swactor context ready'; + } else if (type === 'process_started') { + status.textContent = `Python started · pid ${event.pid}`; + } else if (type === 'spawned') { + status.textContent = 'Program transferred; starting Python…'; + } else if (type === 'exited') { + status.textContent = `Exited · ${JSON.stringify(event.status)}`; + } else if (event.error) { + status.textContent = event.error; + } + return ['spawn_rejected', 'spawn_failed', 'bootstrap_failed', 'exited', 'process_error'].includes(type); + } + + async function pollExecution(panel, requestId, afterSequence) { + if (!panel.isConnected || executionPoll?.requestId !== requestId) return; + try { + const response = await fetch( + `/api/control/contextual/${encodeURIComponent(requestId)}/events?after_sequence=${afterSequence}`, + { cache: 'no-store' }, + ); + const body = await response.json().catch(() => ({})); + if (!response.ok) throw new Error(body.error || `HTTP ${response.status}`); + const execution = body.execution; + let terminal = Boolean(execution?.terminal); + for (const record of execution?.events || []) { + terminal = appendExecutionEvent(panel, record.observation) || terminal; + } + if (terminal) { + stopExecutionPoll(); + panel.querySelector('[data-run]').disabled = false; + return; + } + executionPoll.afterSequence = Number(execution?.next_sequence || afterSequence); + executionPoll.timer = setTimeout( + () => pollExecution(panel, requestId, executionPoll.afterSequence), + 300, + ); + } catch (error) { + panel.querySelector('[data-execution-status]').textContent = error.message; + panel.querySelector('[data-run]').disabled = false; + stopExecutionPoll(); + } + } + + function ensureExecutionControl(nodeView, logicalNodeId, disabled) { + let panel = document.getElementById(EXECUTION_CONTROL_ID); + if (!panel) { + panel = document.createElement('div'); + panel.id = EXECUTION_CONTROL_ID; + panel.className = 'myelin-control myelin-execution'; + panel.innerHTML = `
+ + + Select one Python file +
+ 
`;
+      const control = document.getElementById(NODE_CONTROL_ID);
+      control.after(panel);
+      const input = panel.querySelector('[data-file]');
+      const run = panel.querySelector('[data-run]');
+      input.onchange = () => {
+        const file = input.files?.[0];
+        run.disabled = !file || disabled;
+        panel.querySelector('[data-execution-status]').textContent =
+          file ? `${file.name} · ${file.size} bytes` : 'Select one Python file';
+      };
+      run.onclick = async () => {
+        const file = input.files?.[0];
+        if (!file) return;
+        stopExecutionPoll();
+        run.disabled = true;
+        const status = panel.querySelector('[data-execution-status]');
+        const output = panel.querySelector('[data-execution-output]');
+        output.textContent = '';
+        output.dataset.active = 'false';
+        status.textContent = 'Uploading program…';
+        try {
+          const response = await fetch(
+            `/api/control/contextual/nodes/${logicalNodeId}/python?filename=${encodeURIComponent(file.name)}`,
+            {
+              method: 'POST',
+              headers: { 'content-type': 'application/octet-stream' },
+              body: file,
+            },
+          );
+          const body = await response.json().catch(() => ({}));
+          if (!response.ok) throw new Error(body.error || `HTTP ${response.status}`);
+          const observation = body.observation;
+          const requestId = observation?.request_id;
+          if (!requestId) throw new Error('spawn response did not include a request ID');
+          const terminal = appendExecutionEvent(panel, observation);
+          if (terminal) {
+            run.disabled = false;
+            return;
+          }
+          executionPoll = { requestId, afterSequence: 0, timer: null };
+          pollExecution(panel, requestId, 0);
+        } catch (error) {
+          status.textContent = error.message;
+          run.disabled = false;
+        }
+      };
+    }
+    const input = panel.querySelector('[data-file]');
+    input.disabled = disabled;
+    if (disabled) panel.querySelector('[data-run]').disabled = true;
+  }
+
   async function syncManagedNodeControl(nodeView, model, logicalNodeId) {
     const node = managedNodes(model).find(candidate => candidate.logical_node_id === logicalNodeId);
     if (!node) return;
@@ -230,6 +361,7 @@
     killButton.disabled = pending;
     killButton.textContent = pending ? 'Kill requested' : 'Kill';
     message.textContent = `managed node ${logicalNodeId}: ${node.phase}`;
+    ensureExecutionControl(nodeView, logicalNodeId, terminal || pending);
 
     killButton.onclick = async () => {
       if (!await confirmTermination(
@@ -298,7 +430,9 @@
       const element = mutation.target.nodeType === Node.ELEMENT_NODE
         ? mutation.target
         : mutation.target.parentElement;
-      return !element?.closest(`#${NODE_CONTROL_ID}, #${BULK_CONTROL_ID}`);
+      return !element?.closest(
+        `#${NODE_CONTROL_ID}, #${EXECUTION_CONTROL_ID}, #${BULK_CONTROL_ID}`,
+      );
     });
     if (dashboardChanged) syncControl();
   }).observe(page, { childList: true, subtree: true });
diff --git a/apps/myelin/src/tests/contextual_process_guarantees.rs b/apps/myelin/src/tests/contextual_process_guarantees.rs
index b20f829..a9dd354 100644
--- a/apps/myelin/src/tests/contextual_process_guarantees.rs
+++ b/apps/myelin/src/tests/contextual_process_guarantees.rs
@@ -180,6 +180,7 @@ fn worker_contextual_controller_spawns_queries_stops_and_reclaims_processes() {
                     read_prefixes: vec!["/models".to_owned(), "/runs".to_owned()],
                     write_prefixes: vec!["/runs".to_owned()],
                     attach_timeout_ms: 10_000,
+                    staged_program: None,
                 },
                 reply_to: *events.addr(),
             }),
@@ -336,6 +337,7 @@ fn bootstrap_failure_is_terminal_and_reclaims_execution() {
                     read_prefixes: vec!["/models".to_owned()],
                     write_prefixes: vec!["/runs".to_owned()],
                     attach_timeout_ms: 10_000,
+                    staged_program: None,
                 },
                 reply_to: *events.addr(),
             }),
diff --git a/tools/myelin-e2e-fuzz/src/resources.rs b/tools/myelin-e2e-fuzz/src/resources.rs
index 4bffb37..0baac3c 100644
--- a/tools/myelin-e2e-fuzz/src/resources.rs
+++ b/tools/myelin-e2e-fuzz/src/resources.rs
@@ -48,6 +48,17 @@ pub(crate) fn build_workload_image(workspace: &Path, image: &str) -> Result<(),
         ]),
         "build Myelin base image",
     )?;
+    run_checked(
+        Command::new("docker").current_dir(workspace).args([
+            "build",
+            "-f",
+            "apps/myelin/node-image/Dockerfile",
+            "-t",
+            "myelin-node:latest",
+            ".",
+        ]),
+        "build Myelin node image",
+    )?;
     run_checked(
         Command::new("docker").current_dir(workspace).args([
             "build",