//! Backend-boundary tests: the shared worker contracts (handshake validation,
//! target binding, startup refusals, termination distinctness) must hold over
//! ANY ControlChannel a backend produces, not only a local unix stream.
//!
//! The scripted backend below is a fixture AT the explicit seam: it implements
//! the production `WorkerBackend` trait and feeds the production handshake,
//! exchange loop and stop path. It is labeled plumbing, not SSH validation;
//! the real remote backend lives in m1-relay's relay.rs and U05-U07 run it
//! over genuine SSH.
#![forbid(unsafe_code)]
use std::{path::PathBuf, sync::Arc, time::Duration};
use anyhow::Result;
use klbr_runtime::{
protocol::*,
target::TargetIdentity,
worker::WorkCommand,
worker::{
BackendStartError, BoxFuture, DrainTasks, LaunchedWorker, ProcessEnd, ProcessHandle, Tails,
Worker, WorkerBackend, WorkerLaunch,
},
};
fn root() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.parent()
.unwrap()
.into()
}
fn resolved_identity() -> TargetIdentity {
TargetIdentity::new(
"tester",
"loopback.internal",
"TestOS",
"test64",
"/usr/bin/python3",
"3.11.9",
"/srv/work/tangled",
None,
1,
)
.expect("identity")
}
fn hello_identity(stamp: Option<TargetStamp>) -> WorkerIdentity {
WorkerIdentity {
protocol_version: VERSION,
relay_protocol_version: RELAY_PROTOCOL_VERSION,
target: stamp.map(|stamp| HelloBinding {
identity: Some(resolved_identity()),
stamp,
}),
runtime: RuntimeIdentity {
platform: "TestOS".into(),
arch: "test64".into(),
interpreter_path: "/usr/bin/python3".into(),
interpreter_version: "3.11.9".into(),
environment_fingerprint: None,
capabilities: vec![Capability::RelayFraming { version: 1 }],
},
}
}
/// A scripted backend at the explicit transport seam: the channel is a plain
/// duplex, the peer speaks the framed protocol, and everything shared (handshake
/// validation, envelope binding, exchange loop) is exercised unmodified.
struct ScriptedBackend {
stamp: Option<TargetStamp>,
hello_reports_identity: bool,
hello_forges_stamp: bool,
hello_cwd_override: Option<String>,
reply_completed: bool,
}
impl WorkerBackend for ScriptedBackend {
fn launch(
&self,
launch: WorkerLaunch,
) -> BoxFuture<'_, std::result::Result<LaunchedWorker, BackendStartError>> {
Box::pin(async move {
let (host_side, worker_side) = tokio::io::duplex(4096);
let stamp = self.stamp.clone();
let hello_identity =
if self.hello_reports_identity {
let reported_stamp =
if self.hello_forges_stamp {
// a foreign stamp the host never issued
Some(TargetStamp {
target_id: klbr_runtime::target::TargetId::parse("forged-target")
.map_err(|e| BackendStartError::ChannelFailed {
detail: e.to_string(),
})?,
namespace_id: klbr_runtime::target::NamespaceId::parse("foreign")
.map_err(|e| BackendStartError::ChannelFailed {
detail: e.to_string(),
})?,
namespace_generation:
klbr_runtime::target::NamespaceGeneration::parse("forged-gen")
.map_err(|e| BackendStartError::ChannelFailed {
detail: e.to_string(),
})?,
})
} else {
stamp.clone()
};
let mut reported = hello_identity(reported_stamp);
if let Some(cwd) = &self.hello_cwd_override {
if let Some(binding) = reported.target.as_mut() {
if let Some(identity) = binding.identity.as_mut() {
identity.cwd = cwd.clone();
}
}
}
Some(reported)
} else {
None
};
let reply_completed = self.reply_completed;
tokio::spawn(async move {
let mut worker_side = worker_side;
let hello = Message::Hello {
token: launch.token.clone(),
role: launch.role,
revision: launch.behavior.as_ref().map(|b| b.revision.clone()),
slots: if launch.role == Role::Hooks {
vec![HookSlot::Attention, HookSlot::Delivery]
} else {
vec![]
},
identity: hello_identity,
skills: None,
};
write_frame(
&mut worker_side,
&Envelope {
version: VERSION,
session_id: launch.session_id.clone(),
generation: launch.generation.clone(),
message: hello,
target: stamp.clone(),
},
)
.await
.expect("write hello");
if !reply_completed {
return;
}
loop {
let frame = match read_frame(&mut worker_side).await {
Ok(frame) => frame,
Err(_) => break,
};
// the peer asserts what the host must never violate: the
// remote binding rides on every host frame
if stamp.is_some() && frame.target.is_none() {
eprintln!("scripted peer: host frame carried no target stamp");
break;
}
if stamp.is_none() && frame.target.is_some() {
eprintln!("scripted peer: local host frame claimed a remote stamp");
break;
}
match &frame.message {
Message::Execute { id, .. } => {
let envelope = Envelope {
version: VERSION,
session_id: frame.session_id.clone(),
generation: frame.generation.clone(),
message: Message::Completed {
id: id.clone(),
outcome: Outcome::Cell {
status: CellStatus::Ok,
output: vec![Output {
stream: OutputStream::Result,
text: "scripted-ok".into(),
}],
truncated_bytes: 0,
error: None,
},
},
target: stamp.clone(),
};
if write_frame(&mut worker_side, &envelope).await.is_err() {
break;
}
}
_ => break,
}
}
});
let has_stamp = self.stamp.is_some();
let resolved = if has_stamp {
Some(resolved_identity())
} else {
None
};
Ok(LaunchedWorker {
control: Box::new(host_side),
process: Box::new(EndlessProcess),
tails: Arc::new(std::sync::Mutex::new(Tails::default())),
drains: DrainTasks(vec![]),
resolved,
})
})
}
}
/// A process handle that never ends and ignores kill: keeps the shared actor
/// loop alive so only the exchange contract is under test.
struct EndlessProcess;
impl ProcessHandle for EndlessProcess {
fn kill(&mut self) -> BoxFuture<'_, ()> {
Box::pin(async {})
}
fn wait(&mut self) -> BoxFuture<'_, ProcessEnd> {
Box::pin(std::future::pending())
}
}
fn remote_stamp() -> TargetStamp {
TargetStamp {
target_id: klbr_runtime::target::TargetId::parse("linux-build").unwrap(),
namespace_id: klbr_runtime::target::NamespaceId::parse("workbench").unwrap(),
namespace_generation: klbr_runtime::target::NamespaceGeneration::fresh(),
}
}
#[tokio::test]
async fn remote_handshake_with_issued_binding_executes_over_any_channel() -> Result<()> {
let stamp = remote_stamp();
let backend = ScriptedBackend {
stamp: Some(stamp.clone()),
hello_reports_identity: true,
hello_forges_stamp: false,
hello_cwd_override: None,
reply_completed: true,
};
let worker = Worker::launch_with(
&backend,
Duration::from_secs(5),
SessionId::fresh(),
Role::Workbench,
Some(klbr_runtime::worker::BehaviorSource {
revision: "rev-1".into(),
location: klbr_runtime::worker::BehaviorLocation::RemoteBundle("/srv/bundle".into()),
}),
Some(stamp),
)
.await?;
let outcome = worker
.call(
WorkCommand::Execute {
id: OperationId::fresh(),
code: "42".into(),
purpose: Purpose::Interactive,
},
Duration::from_secs(5),
Arc::new(|_| {
Box::pin(async { HostReply::error("forbidden", "no host effects in this test") })
as klbr_runtime::worker::HostFuture
}),
)
.await?;
assert!(matches!(
outcome,
Outcome::Cell {
status: CellStatus::Ok,
..
}
));
Ok(())
}
#[tokio::test]
async fn local_launch_is_the_unchanged_adapter_path() -> Result<()> {
// Worker::spawn (old signature) must keep working over the refactored
// local backend: a real python worker completes a real handshake.
let temp = tempfile::tempdir()?;
let initial = klbr_runtime::behavior::Release::publish(
&root().join("behavior"),
&temp.path().join("releases"),
)?;
let config = klbr_runtime::worker::WorkerConfig::source_tree(&root());
let worker =
Worker::spawn(&config, SessionId::fresh(), Role::Workbench, Some(&initial)).await?;
assert!(worker.is_alive());
worker.stop().await;
Ok(())
}
#[tokio::test]
async fn local_backend_refuses_a_missing_workspace_without_fallback() -> Result<()> {
let config = klbr_runtime::worker::WorkerConfig {
cwd: PathBuf::from("/nonexistent/klbr-missing-workspace"),
..klbr_runtime::worker::WorkerConfig::source_tree(&root())
};
let error = Worker::spawn(&config, SessionId::fresh(), Role::Workbench, None)
.await
.err()
.expect("missing workspace must be a typed startup refusal");
let cause = error.root_cause().downcast_ref::<BackendStartError>();
assert!(
matches!(cause, Some(BackendStartError::MissingWorkspace { .. })),
"expected typed MissingWorkspace, got {error:?}"
);
Ok(())
}
#[tokio::test]
async fn remote_worker_that_omits_its_identity_is_refused() -> Result<()> {
let stamp = remote_stamp();
let backend = ScriptedBackend {
stamp: Some(stamp.clone()),
hello_reports_identity: false,
hello_forges_stamp: false,
hello_cwd_override: None,
reply_completed: true,
};
let error = Worker::launch_with(
&backend,
Duration::from_secs(5),
SessionId::fresh(),
Role::Workbench,
None,
Some(stamp),
)
.await
.err()
.expect("remote launch without hello identity must be refused");
let cause = error.root_cause().downcast_ref::<NegotiationError>();
assert!(
matches!(cause, Some(NegotiationError::TargetBindingMissing)),
"expected TargetBindingMissing, got {error:?}"
);
Ok(())
}
#[tokio::test]
async fn local_worker_claiming_a_remote_stamp_is_forgery() -> Result<()> {
let backend = ScriptedBackend {
stamp: None,
hello_reports_identity: true,
hello_forges_stamp: true,
hello_cwd_override: None,
reply_completed: true,
};
let error = Worker::launch_with(
&backend,
Duration::from_secs(5),
SessionId::fresh(),
Role::Workbench,
None,
None,
)
.await
.err()
.expect("a forged target stamp must be refused");
let cause = error.root_cause().downcast_ref::<BindingMismatch>();
assert!(
matches!(cause, Some(BindingMismatch::ForgedTargetStamp)),
"expected ForgedTargetStamp, got {error:?}"
);
Ok(())
}
#[tokio::test]
async fn process_end_distinguishes_exit_signal_and_unknown() {
// real exit code
let mut child = std::process::Command::new("sh")
.arg("-c")
.arg("exit 7")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let status = child.wait().unwrap();
assert!(matches!(
ProcessEnd::from_exit_status(&status),
ProcessEnd::Exited(7)
));
// signal death is its own outcome, not an exit code
let mut child = std::process::Command::new("sh")
.arg("-c")
.arg("kill -TERM $$")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let status = child.wait().unwrap();
assert!(
matches!(
ProcessEnd::from_exit_status(&status),
ProcessEnd::Signaled(_)
),
"signal death must be distinct: {status:?}"
);
}
#[tokio::test]
async fn exchanged_frames_carry_the_issued_binding_both_ways() -> Result<()> {
// The scripted peer breaks the moment the host stops stamping its frames
// or validates the worker's stamps loosely: remove either and this test
// fails at the peer's stamp assertions.
let stamp = remote_stamp();
let backend = ScriptedBackend {
stamp: Some(stamp.clone()),
hello_reports_identity: true,
hello_forges_stamp: false,
hello_cwd_override: None,
reply_completed: true,
};
let worker = Worker::launch_with(
&backend,
Duration::from_secs(5),
SessionId::fresh(),
Role::Workbench,
None,
Some(stamp),
)
.await?;
let result = worker
.call(
WorkCommand::Execute {
id: OperationId::fresh(),
code: "1".into(),
purpose: Purpose::Interactive,
},
Duration::from_secs(5),
Arc::new(|_| Box::pin(async { HostReply::error("forbidden", "no") }) as _),
)
.await;
assert!(result.is_ok(), "bound exchange must complete: {result:?}");
Ok(())
}
#[tokio::test]
async fn argv_echo_disagreeing_with_the_measured_identity_is_a_refusal() -> Result<()> {
// The relay MEASURES the target; the worker ECHOES what its argv told it.
// When the two disagree the host must refuse: an echo is a binding check,
// never a source of verified identity.
let stamp = remote_stamp();
let backend = ScriptedBackend {
stamp: Some(stamp.clone()),
hello_reports_identity: true,
hello_forges_stamp: false,
hello_cwd_override: Some("/home/elsewhere".into()),
reply_completed: true,
};
let error = Worker::launch_with(
&backend,
Duration::from_secs(5),
SessionId::fresh(),
Role::Workbench,
None,
Some(stamp),
)
.await
.err()
.expect("probe-vs-stamp disagreement must be refused");
let cause = error.root_cause().downcast_ref::<NegotiationError>();
assert!(
matches!(
cause,
Some(NegotiationError::BindingMismatch { field, .. })
| Some(NegotiationError::RuntimeMismatch { field, .. })
if *field == "cwd",
),
"expected a typed cwd mismatch, got {error:?}"
);
Ok(())
}