ive harnessed the harness
Something went wrong. Try again.
17 kB · 475 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476//! 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, trace: 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(())}