From 03cacef062e7d93aaf07af0cdb9dde3c702a35f2 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Fri, 28 Aug 2026 13:07:31 +0400 Subject: [PATCH] myelin(contextual): run uploaded Python files Add a remote-node file picker that uploads one Python file through the data namespace, materializes it over Iroh, and launches it with contextual-process bootstrap while streaming lifecycle and output events. Clean up execution artifacts, install Python and the swactor wheel in the production node image, and build the worker inside Docker to prevent stale host binaries. --- .dockerignore | 12 - apps/myelin/node-image/Dockerfile | 67 ++- apps/myelin/node-image/Dockerfile.e2e | 51 +- apps/myelin/src/contextual_process.rs | 488 +++++++++++++++++- apps/myelin/src/data_namespace.rs | 1 - apps/myelin/src/node/worker_node_runtime.rs | 45 +- apps/myelin/src/orchestration/actor.rs | 139 ++++- apps/myelin/src/orchestration/app.rs | 20 +- apps/myelin/src/orchestration/control.rs | 207 +++++++- .../myelin/src/orchestration/fleet_control.js | 136 ++++- .../tests/contextual_process_guarantees.rs | 2 + tools/myelin-e2e-fuzz/src/resources.rs | 11 + 12 files changed, 1071 insertions(+), 108 deletions(-) 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",