//! Bounded process actor. Only this actor owns the control channel and child
//! lifetime.
//!
//! Every interrupted/expired exchange invalidates the stream and terminates the
//! worker. Reusing a half-read frame or accepting its late effects is forbidden.
//!
//! Transport lives behind `WorkerBackend`: the local backend spawns a child and
//! hands over its accepted unix stream; a remote backend hands over whatever
//! private framed channel its relay demultiplexed. Handshake validation, target
//! negotiation, startup deadline, per-cell exchange, host-request budget,
//! bounded diagnostics, interrupt and generation invalidation are shared and
//! identical for both backends.
use crate::{behavior::Release, protocol::*, skills::SkillRoots, target::TargetIdentity};
use anyhow::{anyhow, bail, ensure, Context, Result};
use std::{
collections::BTreeMap,
fmt,
future::Future,
path::PathBuf,
pin::Pin,
sync::{
atomic::{AtomicBool, Ordering},
Arc, Mutex,
},
time::Duration,
};
use tokio::{
io::{AsyncRead, AsyncReadExt, AsyncWrite},
net::UnixListener,
process::{Child, Command},
sync::{mpsc, oneshot, watch},
time::{timeout, timeout_at, Instant},
};
#[derive(Clone)]
pub struct WorkerConfig {
pub python: PathBuf,
pub runner: PathBuf,
pub cwd: PathBuf,
pub environment: BTreeMap<String, String>,
pub startup_timeout: Duration,
/// Skill roots this session resolves ids against. They belong to the daemon
/// rather than to one checkout, because the contract a session receives is
/// already the daemon's: a peer running elsewhere must not be handed a skills
/// directory that checkout never published.
pub skills: SkillRoots,
}
impl WorkerConfig {
/// Resolve a worker from explicit installation paths.
///
/// The runner and workspace have independent lifetimes: a packaged runner can
/// execute against a writable source checkout without importing Python from it.
pub fn installed(python: PathBuf, runner: PathBuf, workspace: PathBuf) -> Self {
let environment = [
"PATH",
"HOME",
"LANG",
"LC_ALL",
"TMPDIR",
"LD_LIBRARY_PATH",
"DYLD_LIBRARY_PATH",
]
.into_iter()
.filter_map(|name| std::env::var(name).ok().map(|value| (name.into(), value)))
.collect();
let skills = SkillRoots::for_workspace(&workspace);
Self {
python,
runner,
cwd: workspace,
environment,
startup_timeout: Duration::from_secs(10),
skills,
}
}
/// Point this session at the daemon's skill roots. Absence of a managed root
/// is meaningful: builtin skills alone are a complete namespace.
pub fn with_skills(mut self, skills: SkillRoots) -> Self {
self.skills = skills;
self
}
pub fn source_tree(root: &std::path::Path) -> Self {
Self::installed(
std::env::var_os("KLBR_PYTHON")
.map(PathBuf::from)
.unwrap_or_else(|| "python3".into()),
root.join("python/klbr-runtime/src/klbr_runtime/__main__.py"),
root.into(),
)
}
}
#[derive(Clone, Debug)]
pub struct HostCall {
pub id: RequestId,
pub execution_id: OperationId,
pub request: HostRequest,
}
pub type HostFuture = Pin<Box<dyn Future<Output = HostReply> + Send>>;
pub type HostService = Arc<dyn Fn(HostCall) -> HostFuture + Send + Sync>;
pub fn no_host_operations() -> HostService {
Arc::new(|_| {
Box::pin(async { HostReply::error("forbidden", "this role has no host-effect capability") })
})
}
#[derive(Clone, Debug)]
pub enum WorkCommand {
Execute {
id: OperationId,
code: String,
purpose: Purpose,
},
Invoke {
id: OperationId,
slot: HookSlot,
input: serde_json::Value,
},
RemoteDispatch {
id: OperationId,
operation: String,
payload: serde_json::Value,
},
}
impl WorkCommand {
fn into_message(self) -> (OperationId, Message) {
match self {
Self::Execute { id, code, purpose } => {
(id.clone(), Message::Execute { id, code, purpose })
}
Self::Invoke { id, slot, input } => (id.clone(), Message::Invoke { id, slot, input }),
Self::RemoteDispatch {
id,
operation,
payload,
} => (
id.clone(),
Message::RemoteDispatch {
id,
operation,
payload,
},
),
}
}
}
struct Work {
command: WorkCommand,
deadline: Instant,
service: HostService,
reply: oneshot::Sender<Result<Outcome>>,
}
#[derive(Default)]
pub struct Tails {
stdout: Vec<u8>,
stderr: Vec<u8>,
}
impl Tails {
pub fn stderr_tail(&self) -> Option<String> {
Some(String::from_utf8_lossy(&self.stderr).into_owned())
}
}
#[derive(Clone, Debug)]
pub struct Diagnostics {
pub stdout_tail: String,
pub stderr_tail: String,
}
pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
/// Framed duplex worker control channel. Local backends hand over an accepted
/// unix stream; remote backends hand over the demultiplexed relay stream.
pub trait ControlChannel: AsyncRead + AsyncWrite + Unpin + Send {}
impl<T: AsyncRead + AsyncWrite + Unpin + Send> ControlChannel for T {}
/// Terminal states of a launched process. Exit, signal and unknown are distinct
/// outcomes: a failed transport read is not proof that a remote process did not
/// run, so it must stay representable as unknown.
#[derive(Clone, Debug)]
pub enum ProcessEnd {
Exited(i64),
Signaled(String),
Unknown,
}
impl fmt::Display for ProcessEnd {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Exited(code) => write!(f, "exited({code})"),
Self::Signaled(signal) => write!(f, "signaled({signal})"),
Self::Unknown => write!(f, "unknown"),
}
}
}
pub trait ProcessHandle: Send {
/// Idempotent termination; must not fail on an already-dead process.
fn kill(&mut self) -> BoxFuture<'_, ()>;
/// Resolves once; after kill it resolves promptly.
fn wait(&mut self) -> BoxFuture<'_, ProcessEnd>;
}
impl ProcessEnd {
pub fn from_exit_status(status: &std::process::ExitStatus) -> Self {
if let Some(code) = status.code() {
Self::Exited(i64::from(code))
} else if let Some(signal) = unix_signal(status) {
Self::Signaled(signal)
} else {
Self::Unknown
}
}
}
#[cfg(unix)]
fn unix_signal(status: &std::process::ExitStatus) -> Option<String> {
use std::os::unix::process::ExitStatusExt;
status.signal().map(|s| s.to_string())
}
#[cfg(not(unix))]
fn unix_signal(_status: &std::process::ExitStatus) -> Option<String> {
None
}
#[derive(Clone, Debug)]
pub struct BehaviorSource {
pub revision: String,
pub location: BehaviorLocation,
}
#[derive(Clone, Debug)]
pub enum BehaviorLocation {
LocalPath(PathBuf),
RemoteBundle(String),
}
/// Everything needed to launch ONE worker. Identity and binding are decided by
/// the shared launch path, never by the backend.
pub struct WorkerLaunch {
pub session_id: SessionId,
pub role: Role,
pub generation: Generation,
pub token: String,
pub behavior: Option<BehaviorSource>,
pub binding: Option<TargetStamp>,
pub startup_deadline: Duration,
}
/// Typed startup refusals. Each is an explicit failure BEFORE any execution;
/// none of them has a fallback to local execution, home directories, or
/// another transport.
#[derive(Debug)]
pub enum BackendStartError {
UnavailableTarget { target: String, reason: String },
MissingWorkspace { workspace: String },
PlatformMismatch { required: String, actual: String },
ExitedBeforeReady { detail: String },
ChannelFailed { detail: String },
Deadline,
}
impl fmt::Display for BackendStartError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::UnavailableTarget { target, reason } => {
write!(f, "unavailable target {target}: {reason}")
}
Self::MissingWorkspace { workspace } => {
write!(f, "configured workspace does not exist: {workspace}")
}
Self::PlatformMismatch { required, actual } => {
write!(f, "platform mismatch: required {required}, actual {actual}")
}
Self::ExitedBeforeReady { detail } => {
write!(f, "worker exited before ready: {detail}")
}
Self::ChannelFailed { detail } => write!(f, "control channel failed: {detail}"),
Self::Deadline => write!(f, "worker startup timed out"),
}
}
}
impl std::error::Error for BackendStartError {}
pub struct WorkerStreams {
pub stdout: Box<dyn AsyncRead + Unpin + Send>,
pub stderr: Box<dyn AsyncRead + Unpin + Send>,
}
/// One launched worker handed to the shared actor: private control channel,
/// kill/wait handle, resolved target identity (None = local), and diagnostics
/// already draining into bounded tails via the shared `drain` fn. The backend
/// owns its demultiplexing; the shared actor owns everything else.
pub struct LaunchedWorker {
pub control: Box<dyn ControlChannel>,
pub process: Box<dyn ProcessHandle>,
pub tails: Arc<std::sync::Mutex<Tails>>,
pub drains: DrainTasks,
/// The identity the connect boundary actually verified, once per launch.
/// None for a local launch.
pub resolved: Option<TargetIdentity>,
}
/// The single transport boundary. Local and remote backends feed the same
/// execution, interruption, output, callback and completion contracts; there is
/// no fallback backend.
pub trait WorkerBackend: Send + Sync {
/// Acquire the private control channel and start the process. Typed
/// refusals precede any execution; the future must resolve within
/// `launch.startup_deadline` or report `BackendStartError::Deadline`.
fn launch(
&self,
launch: WorkerLaunch,
) -> BoxFuture<'_, std::result::Result<LaunchedWorker, BackendStartError>>;
}
/// The local backend: today's spawn behavior (temp directory, unix listener,
/// child process, piped diagnostics) as a thin adapter.
pub struct LocalBackend {
config: WorkerConfig,
}
impl LocalBackend {
pub fn new(config: WorkerConfig) -> Self {
Self { config }
}
}
impl WorkerBackend for LocalBackend {
fn launch(
&self,
launch: WorkerLaunch,
) -> BoxFuture<'_, std::result::Result<LaunchedWorker, BackendStartError>> {
Box::pin(self.launch_local(launch))
}
}
impl LocalBackend {
async fn launch_local(
&self,
launch: WorkerLaunch,
) -> std::result::Result<LaunchedWorker, BackendStartError> {
if !self.config.cwd.exists() {
return Err(BackendStartError::MissingWorkspace {
workspace: self.config.cwd.display().to_string(),
});
}
if launch.binding.is_some() {
return Err(BackendStartError::UnavailableTarget {
target: "local".into(),
reason: "local launch cannot carry a remote target binding".into(),
});
}
if let Some(behavior) = &launch.behavior {
if matches!(behavior.location, BehaviorLocation::RemoteBundle(_)) {
return Err(BackendStartError::UnavailableTarget {
target: "local".into(),
reason: "behavior source is a remote bundle; local backend cannot serve it"
.into(),
});
}
}
let directory = tempfile::Builder::new()
.prefix("klbr-")
.tempdir()
.map_err(|e| BackendStartError::ChannelFailed {
detail: format!("temp directory: {e}"),
})?;
let socket_path = directory.path().join("control.sock");
// The worker is handed the skill bodies rather than the roots: one reader,
// one validator, and a set that cannot change under a running turn.
let skills_path = crate::skills::materialize(&self.config.skills, directory.path())
.map_err(|e| BackendStartError::ChannelFailed {
detail: format!("skill payload: {e}"),
})?;
let listener =
UnixListener::bind(&socket_path).map_err(|e| BackendStartError::ChannelFailed {
detail: format!("{e}"),
})?;
let mut command = Command::new(&self.config.python);
command
.args(["-I", "-B", "-u"])
.arg(&self.config.runner)
.arg("--socket")
.arg(&socket_path)
.arg("--role")
.arg(launch.role.argument())
.arg("--session")
.arg(launch.session_id.as_str())
.arg("--generation")
.arg(launch.generation.as_str())
.arg("--token")
.arg(&launch.token)
.arg("--skills")
.arg(&skills_path)
.current_dir(&self.config.cwd)
.env_clear()
.envs(&self.config.environment)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.kill_on_drop(true);
if let Some(behavior) = &launch.behavior {
match &behavior.location {
BehaviorLocation::LocalPath(path) => {
command
.arg("--behavior")
.arg(path)
.arg("--revision")
.arg(&behavior.revision);
}
BehaviorLocation::RemoteBundle(_) => unreachable!("rejected above"),
}
}
let mut child = command
.spawn()
.map_err(|e| BackendStartError::ChannelFailed {
detail: format!("launch Python worker: {e}"),
})?;
let stdout = child
.stdout
.take()
.ok_or(BackendStartError::ChannelFailed {
detail: "missing stdout".into(),
})?;
let stderr = child
.stderr
.take()
.ok_or(BackendStartError::ChannelFailed {
detail: "missing stderr".into(),
})?;
let tails = Arc::new(std::sync::Mutex::new(Tails::default()));
let drains = DrainTasks(vec![
tokio::spawn(drain(stdout, tails.clone(), false)),
tokio::spawn(drain(stderr, tails.clone(), true)),
]);
let channel = tokio::select! {
result = timeout(launch.startup_deadline, async {
let (stream, _) = listener
.accept()
.await
.map_err(|e| BackendStartError::ChannelFailed { detail: format!("{e}") })?;
Ok::<_, BackendStartError>(stream)
}) =>
result.map_err(|_| BackendStartError::Deadline).and_then(|r| r),
status = child.wait() =>
return Err(BackendStartError::ExitedBeforeReady { detail: format!("{status:?}") }),
};
let channel = channel?;
Ok(LaunchedWorker {
control: Box::new(channel),
process: Box::new(LocalProcess {
child,
_directory: directory,
}),
tails,
drains,
resolved: None,
})
}
}
struct LocalProcess {
child: Child,
_directory: tempfile::TempDir,
}
impl ProcessHandle for LocalProcess {
fn kill(&mut self) -> BoxFuture<'_, ()> {
Box::pin(async move {
let _ = self.child.kill().await;
})
}
fn wait(&mut self) -> BoxFuture<'_, ProcessEnd> {
Box::pin(async move {
match self.child.wait().await {
Ok(status) => ProcessEnd::from_exit_status(&status),
Err(_) => ProcessEnd::Unknown,
}
})
}
}
/// What the worker reported about itself at hello, after validation.
struct Handshake {
identity: Option<WorkerIdentity>,
skills: Option<String>,
}
struct Inner {
queue: mpsc::Sender<Work>,
stop: watch::Sender<bool>,
stopped: watch::Receiver<bool>,
alive: Arc<AtomicBool>,
tails: Arc<Mutex<Tails>>,
generation: Generation,
identity: Option<WorkerIdentity>,
/// Skill set the worker echoed from the spawn payload. Recorded so a durable
/// reset names the instruction set that generation loaded.
skills: Option<String>,
/// Target identity MEASURED by the launcher's own probe (None for local).
///
/// This is never the worker's echoed stamp: the echo is a binding check, and a
/// consumer needing account/machine/cwd must read the measured value or none.
resolved: Option<TargetIdentity>,
}
impl Drop for Inner {
fn drop(&mut self) {
let _ = self.stop.send(true);
}
}
#[derive(Clone)]
pub struct Worker(Arc<Inner>);
struct Life {
alive: Arc<AtomicBool>,
stopped: watch::Sender<bool>,
}
impl Drop for Life {
fn drop(&mut self) {
self.alive.store(false, Ordering::Release);
let _ = self.stopped.send(true);
}
}
pub struct DrainTasks(pub Vec<tokio::task::JoinHandle<()>>);
impl Drop for DrainTasks {
fn drop(&mut self) {
for task in &self.0 {
task.abort();
}
}
}
pub async fn drain<R: AsyncRead + Unpin>(mut reader: R, tails: Arc<Mutex<Tails>>, stderr: bool) {
let mut buffer = [0; 4096];
loop {
let count = match reader.read(&mut buffer).await {
Ok(0) | Err(_) => break,
Ok(n) => n,
};
let Ok(mut tails) = tails.lock() else { break };
let tail = if stderr {
&mut tails.stderr
} else {
&mut tails.stdout
};
tail.extend_from_slice(&buffer[..count]);
let excess = tail.len().saturating_sub(8192);
tail.drain(..excess);
}
}
impl Worker {
pub async fn spawn(
config: &WorkerConfig,
session_id: SessionId,
role: Role,
release: Option<&Release>,
) -> Result<Self> {
let behavior = release.map(|r| BehaviorSource {
revision: r.id.clone(),
location: BehaviorLocation::LocalPath(r.path.clone()),
});
Self::launch_with(
&LocalBackend::new(config.clone()),
config.startup_timeout,
session_id,
role,
behavior,
None,
)
.await
}
/// Shared launch path for every backend. The backend only produces the
/// private channel, a kill/wait handle and its verified identity; handshake
/// validation, negotiation, deadline and failure handling are identical.
pub async fn launch_with(
backend: &dyn WorkerBackend,
startup_timeout: Duration,
session_id: SessionId,
role: Role,
behavior: Option<BehaviorSource>,
binding: Option<TargetStamp>,
) -> Result<Self> {
ensure!(
role != Role::Hooks || behavior.is_some(),
"hook role requires a release"
);
let behavior_revision = behavior.as_ref().map(|b| b.revision.clone());
let generation = Generation::fresh();
let token = uuid::Uuid::new_v4().to_string();
let LaunchedWorker {
mut control,
mut process,
tails,
drains,
resolved,
} = backend
.launch(WorkerLaunch {
session_id: session_id.clone(),
role,
generation: generation.clone(),
token: token.clone(),
behavior,
binding: binding.clone(),
startup_deadline: startup_timeout,
})
.await
.context("worker backend refused launch")?;
let expected_binding: Option<HelloBinding> = match (binding.as_ref(), resolved.as_ref()) {
(Some(stamp), Some(identity)) => Some(HelloBinding {
stamp: stamp.clone(),
identity: Some(identity.clone()),
}),
(Some(stamp), None) => {
process.kill().await;
return Err(anyhow!(
"target {stamp}: launch did not resolve a verified target identity"
));
}
(None, _) => None,
};
let handshake = async {
let envelope = read_frame(&mut control).await?;
ensure!(
envelope.session_id == session_id && envelope.generation == generation,
"foreign handshake identity"
);
let handshake = match envelope.message {
Message::Hello {
token: actual_token,
role: actual_role,
revision,
slots,
identity,
skills,
} => {
ensure!(
actual_token == token && actual_role == role,
"worker handshake mismatch"
);
ensure!(revision == behavior_revision, "worker revision mismatch");
let expected = if role == Role::Hooks {
vec![HookSlot::Attention, HookSlot::Delivery]
} else {
vec![]
};
ensure!(slots == expected, "worker hook contract mismatch");
if let Some(reported) = &identity {
WorkerIdentity::validate(reported, expected_binding.as_ref())?;
} else if expected_binding.is_some() {
bail!(NegotiationError::TargetBindingMissing);
}
Handshake { identity, skills }
}
_ => bail!("expected hello"),
};
Ok(handshake)
};
let startup = tokio::select! {
result = timeout(startup_timeout, handshake) =>
result.map_err(|_| anyhow!("worker handshake timed out")).and_then(|result| result),
end = process.wait() => Err(anyhow!("worker exited before ready: {end}")),
};
let handshake = match startup {
Ok(handshake) => handshake,
Err(error) => {
process.kill().await;
let diagnostic = tails
.lock()
.ok()
.and_then(|t| t.stderr_tail())
.unwrap_or_default();
return Err(
error.context(format!("worker startup failed; stderr tail: {diagnostic}"))
);
}
};
let (queue, mut requests) = mpsc::channel::<Work>(16);
let (stop, mut stopping) = watch::channel(false);
let (stopped_tx, stopped) = watch::channel(false);
let alive = Arc::new(AtomicBool::new(true));
let life = Life {
alive: alive.clone(),
stopped: stopped_tx,
};
let exchange_identity = (session_id, generation.clone());
tokio::spawn(async move {
let life_guard = life;
let _drains = drains;
loop {
let work = tokio::select! {
biased;
_ = stopping.changed() => break,
end = process.wait() => { let _ = end; break; }
work = requests.recv() => match work { Some(work) => work, None => break },
};
let Work {
command,
deadline,
service,
mut reply,
} = work;
if reply.is_closed() {
continue;
}
if Instant::now() >= deadline {
let _ = reply.send(Err(anyhow!("worker queue deadline exceeded")));
continue;
}
let result = tokio::select! {
biased;
_ = stopping.changed() => Err(anyhow!("worker stopped by operator")),
_ = reply.closed() => Err(anyhow!("caller cancelled; worker generation invalidated")),
result = timeout_at(deadline, exchange(&mut control, &exchange_identity, role, command, &service, binding.as_ref())) =>
result.map_err(|_| anyhow!("worker deadline exceeded; generation invalidated")).and_then(|r| r),
};
let failed = result.is_err();
if failed {
life_guard.alive.store(false, Ordering::Release);
}
let _ = reply.send(result);
if failed {
break;
}
}
// Stop admissions before draining the queue, including process-exit
// paths that did not finish a request.
life_guard.alive.store(false, Ordering::Release);
requests.close();
// No reuse of a channel with an unknown framing/cancellation state.
process.kill().await;
while let Ok(work) = requests.try_recv() {
let _ = work.reply.send(Err(anyhow!("worker generation has ended")));
}
});
Ok(Self(Arc::new(Inner {
queue,
stop,
stopped,
alive,
tails,
generation,
identity: handshake.identity,
skills: handshake.skills,
resolved,
})))
}
pub fn generation(&self) -> &Generation {
&self.0.generation
}
/// The validated runtime identity the worker reported at hello, if any.
pub fn worker_identity(&self) -> Option<WorkerIdentity> {
self.0.identity.clone()
}
/// The skill set this generation loaded, as the worker echoed it.
pub fn skills_revision(&self) -> Option<String> {
self.0.skills.clone()
}
/// The target identity MEASURED by the launcher's probe, if this is a remote worker.
///
/// Returns `None` for a local worker. Callers needing account/machine/cwd must use
/// this rather than the runtime fields of [`Self::worker_identity`]: those are
/// reported by the worker, while this was measured about it. An absent measurement
/// must surface as absent, never be substituted with self-reported values.
pub fn target_identity(&self) -> Option<TargetIdentity> {
self.0.resolved.clone()
}
pub fn is_alive(&self) -> bool {
self.0.alive.load(Ordering::Acquire) && !*self.0.stop.borrow()
}
pub fn diagnostics(&self) -> Diagnostics {
match self.0.tails.lock() {
Ok(t) => Diagnostics {
stdout_tail: String::from_utf8_lossy(&t.stdout).into(),
stderr_tail: String::from_utf8_lossy(&t.stderr).into(),
},
Err(_) => Diagnostics {
stdout_tail: String::new(),
stderr_tail: "diagnostic lock poisoned".into(),
},
}
}
pub async fn call(
&self,
command: WorkCommand,
budget: Duration,
service: HostService,
) -> Result<Outcome> {
ensure!(self.is_alive(), "worker generation is unavailable");
let (reply, receive) = oneshot::channel();
self.0
.queue
.try_send(Work {
command,
deadline: Instant::now() + budget,
service,
reply,
})
.map_err(|_| anyhow!("worker queue full or closed"))?;
receive
.await
.context("worker ended before completing request")?
}
/// Independent of the request queue. Initially uses a hard process stop;
/// cooperative in-place cancellation is supported by Python but not trusted
/// as the host's recovery guarantee in this slice.
pub async fn stop(&self) {
let _ = self.0.stop.send(true);
let mut stopped = self.0.stopped.clone();
let _ = stopped.wait_for(|value| *value).await;
}
}
async fn exchange(
stream: &mut Box<dyn ControlChannel>,
identity: &(SessionId, Generation),
role: Role,
command: WorkCommand,
service: &HostService,
binding: Option<&TargetStamp>,
) -> Result<Outcome> {
ensure!(
matches!(
(role, &command),
(Role::Workbench, WorkCommand::Execute { .. })
| (Role::Hooks, WorkCommand::Invoke { .. })
| (Role::Remote, WorkCommand::RemoteDispatch { .. })
),
"command is not allowed for worker role"
);
match &command {
WorkCommand::Execute { code, .. } => ensure!(code.len() <= MAX_CODE, "cell too large"),
WorkCommand::RemoteDispatch {
operation, payload, ..
} => {
ensure!(operation.len() <= 128, "remote operation name too large");
ensure!(
serde_json::to_vec(payload)?.len() <= MAX_REMOTE_VALUE,
"remote payload too large"
);
}
WorkCommand::Invoke { .. } => {}
}
let (operation_id, message) = command.into_message();
let envelope = |message| Envelope {
version: VERSION,
session_id: identity.0.clone(),
generation: identity.1.clone(),
message,
target: binding.cloned(),
};
write_frame(stream, &envelope(message)).await?;
let mut requests = 0;
let mut yielded = false;
loop {
let frame = read_frame(stream).await?;
ensure!(
frame.session_id == identity.0 && frame.generation == identity.1,
"stale worker envelope"
);
validate_envelope_target(&frame, binding)?;
match frame.message {
Message::Completed { id, outcome } => {
ensure!(id == operation_id, "wrong completed operation");
outcome.validate(role)?;
return Ok(outcome);
}
Message::HostRequest {
id,
execution_id,
request,
} => {
ensure!(
role == Role::Workbench && execution_id == operation_id,
"foreign host request"
);
requests += 1;
ensure!(requests <= 128, "per-cell host-request budget exhausted");
let is_wait = matches!(request, HostRequest::Wait { .. });
let reply = if yielded {
HostReply::error("scope_closed", "execution has already yielded")
} else {
service(HostCall {
id: id.clone(),
execution_id: execution_id.clone(),
request,
})
.await
};
if is_wait && matches!(reply, HostReply::Ok { .. }) {
yielded = true;
}
write_frame(
stream,
&envelope(Message::HostReply {
id,
execution_id,
reply,
}),
)
.await?;
}
_ => bail!("unexpected worker message"),
}
}
}