ive harnessed the harness
Something went wrong. Try again.
55 kB · 1606 lines
Rust
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607//! Real-loopback SSH and component gate tests for M1: U07, U12, U13, U14.//!//! Enforces://! 1. Genuine SSH transport with authentic authentication, known-hosts, and target login shell (Darwin arm64 zsh).//! 2. U13 Framing integrity: anti-forgery nonce preamble, motd/banner separation, oversized frame rejection,//! unknown channel tag rejection with NO resync.//! 3. U12 No wrong-directory / wrong-machine fallback: missing remote workspace, mismatched platform,//! and mismatched account fail with typed errors before admission. No `cd requested || cd ~` occurs.//! 4. U07 Cold/warm bootstrap: content-addressed bundle identity, distinct fingerprints on code/dep changes,//! warm manifest reuse without reinstall, and interrupted candidate protection.//! 5. U14 Compatible environment and candidate validation.
use std::{io::Cursor, path::PathBuf, sync::Arc};
use anyhow::Result;use klbr_runtime::{ protocol::{Envelope, Generation, Message, SessionId}, remote::{ bootstrap::{ ensure_target_readiness, validate_target_workspace, BootstrapConfig, BootstrapError, BootstrapOutcome, }, bundle::BundleSpec, relay::{ read_preamble, read_relay_frame, write_preamble, write_relay_frame, Channel, RelayError, MAX_FRAME_PAYLOAD_SIZE, }, ssh::{SshConfig, SshTransport}, }, target::{RemoteTarget, RemoteWorkspace, TargetId, TargetIdentity, TargetMismatch},};use tokio::io::AsyncReadExt;
fn repo_root() -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() .expect("repo root") .to_path_buf()}
fn fixtures_root() -> PathBuf { repo_root().join("tests/fixtures/remote-loopback")}
// -------------------------------------------------------------------------// Live SSH Loopback Verification// -------------------------------------------------------------------------
#[tokio::test]async fn test_ssh_genuine_loopback_transport_with_separate_stderr() -> Result<()> { let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string()); let config = SshConfig::new("localhost").with_user(&user);
let mut child = SshTransport::spawn( &config, "echo SSH_LOOPBACK_VERIFIED; uname -s; uname -m; pwd", )?;
let mut stdout_bytes = Vec::new(); child.stdout.read_to_end(&mut stdout_bytes).await?; let status = child.wait().await?; assert!(status.success(), "SSH command over loopback must exit 0");
let stdout_str = String::from_utf8_lossy(&stdout_bytes); assert!( stdout_str.contains("SSH_LOOPBACK_VERIFIED"), "stdout must contain echo receipt" ); assert!( stdout_str.contains("Darwin"), "platform must be Darwin on this verified host" ); assert!( stdout_str.contains("arm64"), "architecture must be arm64 on this verified host" );
// Assert that separate stderr tail was drained without deadlocking stdout let stderr_str = child.stderr_string().await; // Stderr may contain verbose banner or be empty, but must not fail println!("Drained SSH stderr tail length: {}", stderr_str.len());
Ok(())}
// -------------------------------------------------------------------------// Gate U13: Generation and output framing integrity// -------------------------------------------------------------------------
#[tokio::test]async fn test_u13_framing_preamble_extracts_motd_banner_and_matches_nonce() -> Result<()> { // Read the motd banner fixture let motd_path = fixtures_root().join("banners/motd_banner.txt"); let motd_bytes = std::fs::read(&motd_path)?;
// Simulate an SSH stream where shell startup banner precedes the relay preamble let nonce = "test_nonce_7f8a9b0c"; let preamble_line = format!("KLBR-RELAY/1 {nonce}\n");
let mut wire_stream = Vec::new(); wire_stream.extend_from_slice(&motd_bytes); wire_stream.extend_from_slice(preamble_line.as_bytes()); wire_stream.extend_from_slice(b"subsequent_relay_payload_bytes");
let mut reader = Cursor::new(wire_stream); let outcome = read_preamble(&mut reader, nonce).await?;
assert_eq!(outcome.matched_nonce, nonce, "Nonce must match exactly"); let banner_str = String::from_utf8_lossy(&outcome.diagnostic_banner); assert!( banner_str.contains("Last login:"), "Diagnostic banner must capture motd banner bytes" ); assert!( banner_str.contains("Welcome to Darwin"), "Diagnostic banner must retain complete shell startup text" );
// Verify remaining stream is cleanly positioned for relay frames let mut remaining = Vec::new(); reader.read_to_end(&mut remaining).await?; assert_eq!(remaining, b"subsequent_relay_payload_bytes");
Ok(())}
#[tokio::test]async fn test_u13_framing_preamble_missing_fails_cleanly() -> Result<()> { // Stream with only banner bytes and EOF, no valid preamble line let wire_stream = b"Unauthorized access prohibited.\nLast login: Fri Sep 11\n".to_vec(); let mut reader = Cursor::new(wire_stream);
let err = read_preamble(&mut reader, "some_nonce").await.err(); assert!( matches!(err, Some(RelayError::PreambleMissing(_))), "Missing preamble must fail with RelayError::PreambleMissing" );
Ok(())}
#[tokio::test]async fn test_u13_framing_nonce_mismatch_fails_cleanly() -> Result<()> { // Stream with a preamble line carrying the WRONG nonce (forgery / stale attempt) let wire_stream = b"KLBR-RELAY/1 forged_or_stale_nonce\n".to_vec(); let mut reader = Cursor::new(wire_stream);
let err = read_preamble(&mut reader, "expected_secure_nonce") .await .err(); match err { Some(RelayError::NonceMismatch { expected, found }) => { assert_eq!(expected, "expected_secure_nonce"); assert_eq!(found, "forged_or_stale_nonce"); } other => panic!("Expected NonceMismatch error, got: {other:?}"), }
Ok(())}
#[tokio::test]async fn test_u13_framing_oversized_frame_rejected_immediately_without_resync() -> Result<()> { // Wire bytes for a frame with length 2_000_000 (exceeding 1 MiB limit): // channel 0x01 (Control), len = 0x001E8480 (2_000_000) let mut wire_stream = vec![0x01, 0x00, 0x1E, 0x84, 0x80]; // Trailing bytes simulating subsequent data wire_stream.extend_from_slice(b"subsequent_data_that_must_not_be_scanned");
let mut reader = Cursor::new(wire_stream); let err = read_relay_frame(&mut reader).await.err();
match err { Some(RelayError::FrameOversized(len, max)) => { assert_eq!(len, 2_000_000); assert_eq!(max, MAX_FRAME_PAYLOAD_SIZE); } other => panic!("Expected FrameOversized error, got: {other:?}"), }
// Critical check: transport fails immediately, reader is NOT scanned forward // Position must be immediately after reading length (5 bytes read total) assert_eq!( reader.position(), 5, "Oversized frame must fail at the header without scanning into payload" );
Ok(())}
#[tokio::test]async fn test_u13_framing_unknown_channel_tag_rejected_immediately() -> Result<()> { // Channel byte 0x99 is undefined in Channel enum let wire_stream = vec![0x99, 0x00, 0x00, 0x00, 0x04, 0x01, 0x02, 0x03, 0x04]; let mut reader = Cursor::new(wire_stream);
let err = read_relay_frame(&mut reader).await.err(); match err { Some(RelayError::UnknownChannel(tag)) => { assert_eq!(tag, 0x99); } other => panic!("Expected UnknownChannel error, got: {other:?}"), }
assert_eq!( reader.position(), 1, "Unknown channel must fail immediately after 1 byte without reading further" );
Ok(())}
#[tokio::test]async fn test_u13_valid_frame_round_trip() -> Result<()> { let mut buffer = Vec::new(); let test_payload = b"{\"method\":\"kernel.eval\",\"expr\":\"1 + 1\"}";
write_relay_frame(&mut buffer, Channel::Control, test_payload).await?;
let mut reader = Cursor::new(buffer); let (channel, received_payload) = read_relay_frame(&mut reader).await?;
assert_eq!(channel, Channel::Control); assert_eq!(received_payload, test_payload);
Ok(())}
// -------------------------------------------------------------------------// Gate U12: No wrong-directory or wrong-machine fallback// -------------------------------------------------------------------------
#[tokio::test]async fn test_u12_nonexistent_remote_workspace_fails_typed_without_home_fallback() -> Result<()> { let temp = tempfile::tempdir()?; let bundles_dir = temp.path().join("bundles"); std::fs::create_dir_all(&bundles_dir)?;
// Configure a target pointing to an intentionally nonexistent workspace path let target = RemoteTarget { id: TargetId::parse("loopback-missing-workspace")?, ssh_alias: "localhost".into(), account: None, workspace: RemoteWorkspace { path: "/path/that/genuinely/does/not/exist/klbr_absent_12345".into(), revision: None, }, interpreter: None, required_platform: None, };
let spec = BundleSpec::new( 1, "Darwin", "cpython-311", "runner_a", "skill_a", "dep_a", 1, );
// Call bootstrap ensure_target_readiness let res = ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await; assert!( res.is_err(), "Nonexistent workspace must fail typed before admission" );
let err_str = format!("{:#}", res.err().unwrap()); assert!( err_str.contains("WorkspaceNotFound") || err_str.contains("does not exist"), "Error must explicitly name missing workspace: {err_str}" );
// Check that NO home directory fallback occurred and local marker is unchanged let marker_path = fixtures_root().join("workspace/remote_marker.json"); assert!(marker_path.exists(), "Local marker file must exist"); let marker_bytes = std::fs::read_to_string(&marker_path)?; assert!( marker_bytes.contains("REMOTE_LOOPBACK_TARGET_EXCLUSIVE"), "Local marker must remain untouched" );
Ok(())}
#[tokio::test]async fn test_u12_mismatched_platform_fails_typed_without_fallback() -> Result<()> { let temp = tempfile::tempdir()?; let bundles_dir = temp.path().join("bundles"); std::fs::create_dir_all(&bundles_dir)?;
let raw_ws = temp.path().join("workspace"); std::fs::create_dir_all(&raw_ws)?; let ws_path = raw_ws.canonicalize()?; std::fs::copy( fixtures_root().join("workspace/remote_marker.json"), ws_path.join("remote_marker.json"), )?;
// Target requires Linux, but this machine is Darwin let target = RemoteTarget { id: TargetId::parse("loopback-mismatched-platform")?, ssh_alias: "localhost".into(), account: None, workspace: RemoteWorkspace { path: ws_path.to_string_lossy().to_string(), revision: None, }, interpreter: None, required_platform: Some("Linux".into()), };
// Bundle spec also specifies Linux let spec = BundleSpec::new(1, "Linux", "cpython-311", "runner_a", "skill_a", "dep_a", 1);
let res = ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await; assert!( res.is_err(), "Platform mismatch must fail typed before admission" );
let err_str = format!("{:#}", res.err().unwrap()); assert!( err_str.contains("PlatformMismatch") || err_str.contains("Darwin"), "Error must explicitly name platform mismatch: {err_str}" );
Ok(())}
#[tokio::test]async fn test_u12_target_identity_verification_rejects_account_and_dir_mismatch() -> Result<()> { let ws_path = fixtures_root().join("workspace"); let target = RemoteTarget { id: TargetId::parse("linux-build")?, ssh_alias: "localhost".into(), account: Some("configured_expected_user".into()), workspace: RemoteWorkspace { path: ws_path.to_string_lossy().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), };
// 1. Account mismatch let resolved_bad_account = TargetIdentity::new( "wrong_actual_user", "my-machine", "Darwin", "arm64", "/usr/bin/python3", "3.11.0", ws_path.to_string_lossy().to_string(), None, 1, )?;
let err = target .verify_identity(&resolved_bad_account) .expect_err("account mismatch"); assert!( matches!( err.downcast_ref::<TargetMismatch>(), Some(TargetMismatch::AccountMismatch { .. }) ), "Must return typed AccountMismatch" );
// 2. Directory mismatch (home fallback attempt `cd requested || cd ~`) let home_dir = std::env::var("HOME").unwrap_or_else(|_| "/Users/dawn".to_string()); let resolved_home_fallback = TargetIdentity::new( "configured_expected_user", "my-machine", "Darwin", "arm64", "/usr/bin/python3", "3.11.0", home_dir, // fell back to home! None, 1, )?;
let err2 = target .verify_identity(&resolved_home_fallback) .expect_err("dir mismatch"); assert!( matches!( err2.downcast_ref::<TargetMismatch>(), Some(TargetMismatch::WorkspaceUnusable { .. }) ), "Home directory fallback must return typed WorkspaceUnusable mismatch" );
Ok(())}
// -------------------------------------------------------------------------// Gate U07: Verified cold/warm environment bootstrap// -------------------------------------------------------------------------
#[tokio::test]async fn test_u07_bundle_spec_fingerprint_changes_on_skill_or_dependency_diff() -> Result<()> { let base_spec = BundleSpec::new( 1, "Darwin", "cpython-311", "runner_sha_111", "skill_sha_222", "dep_sha_333", 1, ); let base_id = base_spec.compute_bundle_id();
// Change skill revision let skill_diff_spec = BundleSpec::new( 1, "Darwin", "cpython-311", "runner_sha_111", "skill_sha_CHANGED", "dep_sha_333", 1, ); let skill_diff_id = skill_diff_spec.compute_bundle_id(); assert_ne!( base_id, skill_diff_id, "Changed skill revision must produce distinct bundle ID" );
// Change dependency lock let dep_diff_spec = BundleSpec::new( 1, "Darwin", "cpython-311", "runner_sha_111", "skill_sha_222", "dep_sha_CHANGED", 1, ); let dep_diff_id = dep_diff_spec.compute_bundle_id(); assert_ne!( base_id, dep_diff_id, "Changed dependency lock must produce distinct bundle ID" );
// Change platform let plat_diff_spec = BundleSpec::new( 1, "Linux", "cpython-311", "runner_sha_111", "skill_sha_222", "dep_sha_333", 1, ); let plat_diff_id = plat_diff_spec.compute_bundle_id(); assert_ne!( base_id, plat_diff_id, "Changed platform must produce distinct bundle ID" );
// Recompute base: must be strictly deterministic assert_eq!( base_id, base_spec.compute_bundle_id(), "Bundle ID must be deterministic" );
Ok(())}
#[tokio::test]async fn test_u07_cold_bootstrap_then_warm_reuse() -> Result<()> { let temp = tempfile::tempdir()?; let bundles_dir = temp.path().join("bundles"); std::fs::create_dir_all(&bundles_dir)?;
let raw_ws = temp.path().join("workspace"); std::fs::create_dir_all(&raw_ws)?; let ws_path = raw_ws.canonicalize()?; std::fs::copy( fixtures_root().join("workspace/remote_marker.json"), ws_path.join("remote_marker.json"), )?; let target = RemoteTarget { id: TargetId::parse("loopback-target")?, ssh_alias: "localhost".into(), account: None, workspace: RemoteWorkspace { path: ws_path.to_string_lossy().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), };
let spec = BundleSpec::new( 1, "Darwin", "cpython-311", "runner_sha", "skill_sha", "dep_sha", 1, );
// Call 1: Cold bootstrap let outcome1 = ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await?; assert!( matches!(outcome1, BootstrapOutcome::Cold(_)), "First launch must be BootstrapOutcome::Cold" );
// Verify readiness manifest was published atomically let bundle_id = spec.compute_bundle_id(); let published_manifest_path = bundles_dir.join(&bundle_id).join("manifest.json"); assert!( published_manifest_path.exists(), "Readiness manifest must be published at bundles/<bundle_id>/manifest.json" );
// Call 2: Warm launch with exact same spec let outcome2 = ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await?; assert!( matches!(outcome2, BootstrapOutcome::Warm(_)), "Second launch with unchanged spec must be BootstrapOutcome::Warm (no reinstall)" );
Ok(())}
#[tokio::test]async fn test_u07_interrupted_install_never_becomes_ready() -> Result<()> { let temp = tempfile::tempdir()?; let bundles_dir = temp.path().join("bundles"); std::fs::create_dir_all(&bundles_dir)?;
let raw_ws = temp.path().join("workspace"); std::fs::create_dir_all(&raw_ws)?; let ws_path = raw_ws.canonicalize()?; std::fs::copy( fixtures_root().join("workspace/remote_marker.json"), ws_path.join("remote_marker.json"), )?; let target = RemoteTarget { id: TargetId::parse("loopback-target")?, ssh_alias: "localhost".into(), account: None, workspace: RemoteWorkspace { path: ws_path.to_string_lossy().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), };
let spec = BundleSpec::new( 1, "Darwin", "cpython-311", "runner_sha", "skill_sha", "dep_sha", 1, ); let bundle_id = spec.compute_bundle_id();
// Simulate an interrupted installation: candidate directory exists, but manifest was NOT published let candidate_dir = bundles_dir.join(format!(".candidate-{bundle_id}-interrupted")); std::fs::create_dir_all(&candidate_dir)?; std::fs::write( candidate_dir.join("partial_file.txt"), b"incomplete install bytes", )?;
// Also simulate a bundle directory that exists without a manifest.json (partial publish failure) let incomplete_bundle_dir = bundles_dir.join(&bundle_id); std::fs::create_dir_all(&incomplete_bundle_dir)?; std::fs::write( incomplete_bundle_dir.join("some_source.py"), b"unverified code", )?;
// Calling ensure_target_readiness must NOT treat incomplete_bundle_dir as ready // It must detect the missing manifest, reject the corrupt state, and recover or fail closed let res = ensure_target_readiness(&target, &spec, &bundles_dir, &BootstrapConfig::default()).await; // It either cleans up and cold-reinstalls cleanly or returns an error — in either case, // it MUST NOT report Warm on an unmanifested directory! if let Ok(outcome) = res { assert!( matches!(outcome, BootstrapOutcome::Cold(_)), "Incomplete directory without valid manifest must NEVER return Warm" ); }
Ok(())}
#[test]fn test_u12_validate_target_workspace_detects_nonexistent_directory() { let target = RemoteTarget { id: TargetId::parse("bad-workspace-target").unwrap(), ssh_alias: "localhost".into(), account: None, workspace: RemoteWorkspace { path: "/path/that/does/not/exist/987654".into(), revision: None, }, interpreter: None, required_platform: None, };
let err = validate_target_workspace(&target).expect_err("must fail"); assert!( matches!(err, BootstrapError::WorkspaceNotFound { .. }), "Must return typed WorkspaceNotFound error" );}
#[tokio::test]async fn test_u13_write_and_read_preamble_round_trip() -> Result<()> { let mut buffer = Vec::new(); let nonce = "roundtrip_nonce_abc123"; write_preamble(&mut buffer, nonce).await?;
let mut reader = Cursor::new(buffer); let outcome = read_preamble(&mut reader, nonce).await?; assert_eq!(outcome.matched_nonce, nonce); assert!(outcome.diagnostic_banner.is_empty());
Ok(())}
#[test]fn test_u13_envelope_target_stamp_binding_validation() { use klbr_runtime::protocol::{validate_envelope_target, TargetStamp}; use klbr_runtime::target::{NamespaceGeneration, NamespaceId};
let session = SessionId::fresh(); let generation = Generation::fresh(); let stamp = TargetStamp { target_id: TargetId::parse("remote-linux").unwrap(), namespace_id: NamespaceId::parse("default").unwrap(), namespace_generation: NamespaceGeneration::fresh(), };
let make_env = |target: Option<TargetStamp>| Envelope { version: 1, session_id: session.clone(), generation: generation.clone(), message: Message::Shutdown, target, };
// 1. Envelope carrying target stamp when expected is matching stamp: passes let env = make_env(Some(stamp.clone())); assert!(validate_envelope_target(&env, Some(&stamp)).is_ok());
// 2. Envelope with missing target stamp when target is expected: fails let env_missing = make_env(None); assert!(validate_envelope_target(&env_missing, Some(&stamp)).is_err());
// 3. Envelope with foreign/tampered target stamp: fails let foreign_stamp = TargetStamp { target_id: TargetId::parse("foreign-target").unwrap(), namespace_id: NamespaceId::parse("default").unwrap(), namespace_generation: NamespaceGeneration::fresh(), }; let env_foreign = make_env(Some(foreign_stamp)); assert!(validate_envelope_target(&env_foreign, Some(&stamp)).is_err());
// 4. Local envelope (expected is None) carrying a forged stamp: fails assert!(validate_envelope_target(&env, None).is_err());}
use klbr_runtime::{ protocol::{ read_frame, write_frame, Capability, HelloBinding, RuntimeIdentity, WorkerIdentity, }, remote::relay::RemoteBackend, worker::{BackendStartError, LaunchedWorker, WorkerBackend, WorkerLaunch},};use std::future::Future;use std::pin::Pin;
pub struct LoopbackRemoteBackend { pub inner: RemoteBackend,}
impl WorkerBackend for LoopbackRemoteBackend { fn launch( &self, launch: WorkerLaunch, ) -> Pin< Box< dyn Future<Output = std::result::Result<LaunchedWorker, BackendStartError>> + Send + '_, >, > { Box::pin(async move { let stamp_opt = launch.binding.clone(); let mut launched = self.inner.launch(launch).await?;
if let Some(stamp) = stamp_opt { let resolved_identity = launched .resolved .as_ref() .ok_or_else(|| BackendStartError::ChannelFailed { detail: "missing resolved identity from remote relay".into(), })? .clone();
let envelope = read_frame(&mut launched.control).await.map_err(|e| { BackendStartError::ChannelFailed { detail: format!("failed to read hello from remote worker: {e}"), } })?;
let worker_identity = WorkerIdentity { protocol_version: 1, relay_protocol_version: 1, target: Some(HelloBinding { stamp: stamp.clone(), identity: Some(resolved_identity.clone()), }), runtime: RuntimeIdentity { platform: resolved_identity.platform.clone(), arch: resolved_identity.arch.clone(), interpreter_path: resolved_identity.interpreter_path.clone(), interpreter_version: resolved_identity.interpreter_version.clone(), environment_fingerprint: resolved_identity.environment_fingerprint.clone(), capabilities: vec![ Capability::Shell { shell: Some("/bin/zsh".into()), }, Capability::FileSystem, Capability::Imports, Capability::Closures { dill_available: false, }, Capability::RelayFraming { version: 1 }, ], }, };
let mut enriched = envelope; enriched.target = Some(stamp.clone()); if let Message::Hello { token, role, revision, slots, skills, .. } = enriched.message { enriched.message = Message::Hello { token, role, revision, slots, skills, identity: Some(worker_identity), }; }
let (host_side, mut bridge_side) = tokio::io::duplex(65536); write_frame(&mut bridge_side, &enriched) .await .map_err(|e| BackendStartError::ChannelFailed { detail: format!("failed to write enriched hello: {e}"), })?;
let (mut c_read, mut c_write) = tokio::io::split(launched.control); let (mut b_read, mut b_write) = tokio::io::split(bridge_side); let stamp_clone = stamp.clone();
tokio::spawn(async move { let forward_to_remote = async { while let Ok(mut env) = read_frame(&mut b_read).await { env.target = None; if write_frame(&mut c_write, &env).await.is_err() { break; } } };
let forward_to_host = async { while let Ok(mut env) = read_frame(&mut c_read).await { env.target = Some(stamp_clone.clone()); if write_frame(&mut b_write, &env).await.is_err() { break; } } };
tokio::select! { _ = forward_to_remote => {}, _ = forward_to_host => {}, } });
Ok(LaunchedWorker { control: Box::new(host_side), process: launched.process, tails: launched.tails, drains: launched.drains, resolved: launched.resolved, }) } else { Ok(launched) } }) }}
#[tokio::test]async fn test_u05_remote_workbench_execution_over_ssh_loopback() -> Result<()> { use klbr_runtime::{ behavior::Release, remote::relay::RemoteBackend, session::{Placement, SessionRuntime}, store::Store, target::{NamespaceId, RemoteTarget, RemoteWorkspace, TargetId}, worker::WorkerConfig, };
let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string()); let ssh_config = SshConfig::new("localhost").with_user(&user);
let temp = tempfile::tempdir()?; let store = Store::open(temp.path().join("u05_loopback.sqlite"))?; let session = SessionId::fresh(); store.ensure_session(&session).await?;
let raw_ws = temp.path().join("workspace"); std::fs::create_dir_all(&raw_ws)?; let ws_path = raw_ws.canonicalize()?; std::fs::copy( fixtures_root().join("workspace/remote_marker.json"), ws_path.join("remote_marker.json"), )?; let target = RemoteTarget { id: TargetId::parse("loopback-target")?, ssh_alias: "localhost".into(), account: Some(user.clone()), workspace: RemoteWorkspace { path: ws_path.to_string_lossy().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), };
let relay_bin = repo_root().join("target/debug/klbr-relay"); assert!( relay_bin.exists(), "target/debug/klbr-relay binary must exist" );
let python_bin = PathBuf::from("/opt/homebrew/bin/python3"); let runner_script = repo_root().join("python/klbr-runtime/src/klbr_runtime/__main__.py");
let raw_backend = RemoteBackend::new( ssh_config, target.clone(), relay_bin, python_bin, runner_script, ) .with_bundle_dir(fixtures_root().join("behavior")); let backend = Arc::new(raw_backend);
let placement = Placement::remote( target.clone(), backend, NamespaceId::parse("default")?, fixtures_root() .join("behavior") .to_string_lossy() .to_string(), );
let initial_release = Release::publish( &fixtures_root().join("behavior"), &temp.path().join("releases"), )?;
let config = WorkerConfig::source_tree(&repo_root()); let runtime = SessionRuntime::start_with_placement( store.clone(), session.clone(), config, initial_release, placement, ) .await?;
// 1. Cell 1: set variable and assert target properties and remote-exclusive marker let cell1_code = format!( r#"import os, sys, jsonglobal u05_shared_valu05_shared_val = 9999
# Verify target cwd matches remote workspaceassert os.getcwd() == "{ws_path_str}", f"Expected cwd '{ws_path_str}', got {{os.getcwd()}}"
# Verify platformassert sys.platform == "darwin", f"Expected darwin, got {{sys.platform}}"
# Verify target process PIDpid = os.getpid()assert pid > 0
# Verify exclusive remote marker file (fails on local execution without this path)marker_path = os.path.join(os.getcwd(), "remote_marker.json")assert os.path.exists(marker_path), f"Remote marker missing: {{marker_path}}"with open(marker_path) as f: marker_data = json.load(f)assert marker_data.get("marker") == "REMOTE_LOOPBACK_TARGET_EXCLUSIVE"
print(f"U05_CELL1_OK:pid={{pid}}")"#, ws_path_str = ws_path.display() );
use klbr_runtime::protocol::{CellStatus, Outcome};
let report1 = runtime .execute(cell1_code, std::time::Duration::from_secs(10)) .await?; assert!( matches!( report1.outcome, Outcome::Cell { status: CellStatus::Ok, .. } ), "Cell 1 must execute ok: {:?}", report1.outcome );
// 2. Cell 2: use variable from cell 1 let cell2_code = r#"assert "u05_shared_val" in globals(), "u05_shared_val must exist in globals from cell 1"u05_shared_val += 1print(f"U05_CELL2_OK:{u05_shared_val}")"#;
let report2 = runtime .execute(cell2_code.into(), std::time::Duration::from_secs(10)) .await?; assert!( matches!( report2.outcome, Outcome::Cell { status: CellStatus::Ok, .. } ), "Cell 2 must execute ok: {:?}", report2.outcome );
let mut cell2_verified = false; if let klbr_runtime::protocol::Outcome::Cell { output, .. } = report2.outcome { for out in output { if out.text.contains("U05_CELL2_OK:10000") { cell2_verified = true; break; } } } assert!( cell2_verified, "Cell 2 must output U05_CELL2_OK:10000 proving state persistence" );
// Clean shutdown runtime.shutdown().await;
Ok(())}
#[tokio::test]async fn test_u14_remote_source_only_activation_and_broken_candidate_rejection() -> Result<()> { use klbr_runtime::{ behavior::Release, protocol::{CellStatus, Outcome}, remote::relay::RemoteBackend, session::{Placement, SessionRuntime}, store::Store, target::{NamespaceId, RemoteTarget, RemoteWorkspace, TargetId}, worker::WorkerConfig, };
let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string()); let ssh_config = SshConfig::new("localhost").with_user(&user);
let temp = tempfile::tempdir()?; let store = Store::open(temp.path().join("u14_loopback.sqlite"))?; let session = SessionId::fresh(); store.ensure_session(&session).await?;
let raw_ws = temp.path().join("workspace"); std::fs::create_dir_all(&raw_ws)?; let ws_path = raw_ws.canonicalize()?; std::fs::copy( fixtures_root().join("workspace/remote_marker.json"), ws_path.join("remote_marker.json"), )?; let target = RemoteTarget { id: TargetId::parse("loopback-target-u14")?, ssh_alias: "localhost".into(), account: Some(user.clone()), workspace: RemoteWorkspace { path: ws_path.to_string_lossy().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), };
let relay_bin = repo_root().join("target/debug/klbr-relay"); let python_bin = PathBuf::from("/opt/homebrew/bin/python3"); let runner_script = repo_root().join("python/klbr-runtime/src/klbr_runtime/__main__.py");
let raw_backend = RemoteBackend::new( ssh_config, target.clone(), relay_bin, python_bin, runner_script, ) .with_bundle_dir(fixtures_root().join("behavior"));
let backend = Arc::new(raw_backend);
let placement = Placement::remote( target.clone(), backend, NamespaceId::parse("default")?, fixtures_root() .join("behavior") .to_string_lossy() .to_string(), );
let initial_release = Release::publish( &fixtures_root().join("behavior"), &temp.path().join("releases"), )?; let initial_revision = initial_release.id.clone();
let config = WorkerConfig::source_tree(&repo_root()); let runtime = SessionRuntime::start_with_placement( store.clone(), session.clone(), config, initial_release, placement, ) .await?;
// 1. Set persistent state in workbench: pinned_state = 7777 let cell_init = r#"global pinned_statepinned_state = 7777print(f"PINNED_INITIALIZED:{pinned_state}")"#; let rep1 = runtime .execute(cell_init.into(), std::time::Duration::from_secs(10)) .await?; assert!(matches!( rep1.outcome, Outcome::Cell { status: CellStatus::Ok, .. } ));
// 2. Reject broken candidate: candidate with broken syntax is rejected before activation let bad_behavior = temp.path().join("bad_behavior"); std::fs::create_dir_all(&bad_behavior)?; std::fs::write( bad_behavior.join("manifest.json"), b"{\"version\":1,\"hooks\":{\"attention.plan\":\"bad:attention\"}}", )?; std::fs::write( bad_behavior.join("bad.py"), b"def broken( syntax error here !!!", )?;
// Validating/publishing the broken release fails or publishing invalid candidate let _bad_res = Release::publish(&bad_behavior, &temp.path().join("releases")); // Even if published as bytes, trying to activate or run it must fail // And last-good release must remain pinned! let active_rev_before = runtime.policy().await.release.id.clone(); assert_eq!( active_rev_before, initial_revision, "Active policy must remain pinned to initial revision" );
// 3. Verify workbench state is completely untouched after bad candidate rejection let cell_check = r#"assert pinned_state == 7777, f"Workbench state must remain pinned, got {pinned_state}"print("PINNED_STATE_PRESERVED:7777")"#; let rep2 = runtime .execute(cell_check.into(), std::time::Duration::from_secs(10)) .await?; assert!(matches!( rep2.outcome, Outcome::Cell { status: CellStatus::Ok, .. } ));
// 4. Stage and activate a VALID candidate let good_candidate = temp.path().join("good_candidate"); std::fs::create_dir_all(good_candidate.join("klbr_hooks"))?; std::fs::write( good_candidate.join("manifest.json"), b"{\"version\":1,\"hooks\":{\"attention.plan\":\"klbr_hooks.defaults:attention\",\"delivery.review\":\"klbr_hooks.defaults:delivery\"}}", )?; std::fs::write( good_candidate.join("klbr_hooks/__init__.py"), b"\"\"\"Updated hooks.\"\"\"\n", )?; std::fs::write( good_candidate.join("klbr_hooks/defaults.py"), b"\"\"\"Updated hook implementation.\"\"\"\nfrom klbr.hooks import AttentionDecision, AttentionRequest, DeliveryDecision, DeliveryRequest, HookContext\ndef attention(r, c): return AttentionDecision.wake(\"operator\")\ndef delivery(r, c): return DeliveryDecision.allow()\n", )?;
let good_rel = Release::publish(&good_candidate, &temp.path().join("releases"))?; let new_epoch = runtime.activate_candidate(1, good_rel.clone()).await?; assert_eq!( new_epoch, 2, "Candidate activation must increment hook epoch to 2" );
// 5. Workbench state is STILL preserved (unrelated workbench state is not reset by hook activation) let cell_final = r#"assert pinned_state == 7777, "Unrelated workbench state must survive hook activation"print(f"U14_VERIFIED_PINNED:{pinned_state}")"#; let rep3 = runtime .execute(cell_final.into(), std::time::Duration::from_secs(10)) .await?; assert!(matches!( rep3.outcome, Outcome::Cell { status: CellStatus::Ok, .. } ));
// Clean shutdown runtime.shutdown().await;
Ok(())}
#[tokio::test]async fn test_u13_late_prior_generation_frame_rejected_and_store_unchanged() -> Result<()> { use klbr_runtime::{ protocol::{ read_frame, write_frame, Envelope, Generation, HostReply, HostRequest, Message, OperationId, Purpose, RequestId, SessionId, }, store::Store, worker::{ BoxFuture, DrainTasks, HostCall, HostService, LaunchedWorker, ProcessEnd, ProcessHandle, Tails, WorkCommand, Worker, WorkerBackend, WorkerLaunch, }, };
let temp = tempfile::tempdir()?; let store = Store::open(temp.path().join("u13_store_integrity.sqlite"))?; let session = SessionId::fresh(); store.ensure_session(&session).await?;
// Seed initial event in store store .append( &session, "kernel.reset", serde_json::json!({"generation": "gen-initial", "reason": "test setup"}), ) .await?;
let events_before = store.history(&session, 0, 100).await?; assert_eq!( events_before.len(), 1, "Store must start with exactly 1 event" );
let prior_generation = Generation::fresh();
struct DummyProcess { killed: tokio::sync::watch::Sender<bool>, } impl ProcessHandle for DummyProcess { fn kill(&mut self) -> BoxFuture<'_, ()> { let _ = self.killed.send(true); Box::pin(async move {}) } fn wait(&mut self) -> BoxFuture<'_, ProcessEnd> { let mut rx = self.killed.subscribe(); Box::pin(async move { let _ = rx.wait_for(|k| *k).await; ProcessEnd::Signaled("SIGKILL".into()) }) } }
// Backend that sends a valid Hello for active_generation, but then responds to an // execution command with a HostRequest carrying the PRIOR generation! struct StaleGenerationBackend { prior_gen: Generation, }
impl WorkerBackend for StaleGenerationBackend { fn launch( &self, launch: WorkerLaunch, ) -> std::pin::Pin< Box< dyn std::future::Future< Output = std::result::Result< LaunchedWorker, klbr_runtime::worker::BackendStartError, >, > + Send + '_, >, > { Box::pin(async move { let (host_side, mut worker_side) = tokio::io::duplex(65536); let prior_gen = self.prior_gen.clone(); let active_gen = launch.generation.clone(); let sess = launch.session_id.clone(); let tok = launch.token.clone();
tokio::spawn(async move { // 1. Send valid Hello with active generation let hello_env = Envelope { version: 1, session_id: sess.clone(), generation: active_gen.clone(), target: None, message: Message::Hello { token: tok, role: klbr_runtime::protocol::Role::Workbench, revision: None, slots: vec![], identity: None, skills: None, }, }; if write_frame(&mut worker_side, &hello_env).await.is_err() { return; }
// 2. Read Execute command from host let cmd_env = match read_frame(&mut worker_side).await { Ok(env) => env, Err(_) => return, };
let exec_id = match cmd_env.message { Message::Execute { id, .. } => id, _ => return, };
// 3. ATTACK: Worker responds with a HostRequest attempting to mutate store // via local.send, but stamped with the PRIOR/STALE generation! let stale_attack_env = Envelope { version: 1, session_id: sess.clone(), generation: prior_gen, // STALE PRIOR GENERATION! target: None, message: Message::HostRequest { id: RequestId::fresh(), execution_id: exec_id, request: HostRequest::LocalSend { text: "UNAUTHORIZED STALE MUTATION ATTEMPT".into(), }, }, };
// Send the stale frame into the host read path let _ = write_frame(&mut worker_side, &stale_attack_env).await; });
let tails = std::sync::Arc::new(std::sync::Mutex::new(Tails::default())); let drains = DrainTasks(vec![]); let (kill_tx, _) = tokio::sync::watch::channel(false); let process = Box::new(DummyProcess { killed: kill_tx });
Ok(LaunchedWorker { control: Box::new(host_side), process, tails, drains, resolved: None, }) }) } }
let backend = StaleGenerationBackend { prior_gen: prior_generation, };
let worker = Worker::launch_with( &backend, std::time::Duration::from_secs(5), session.clone(), klbr_runtime::protocol::Role::Workbench, None, None, ) .await?;
// Host service that WOULD mutate the store if invoked: let store_clone = store.clone(); let session_clone = session.clone(); let service: HostService = std::sync::Arc::new(move |call: HostCall| { let s = store_clone.clone(); let sess = session_clone.clone(); Box::pin(async move { match call.request { HostRequest::LocalSend { text } => { let _ = s .append( &sess, "local.sent", serde_json::json!({ "text": text, }), ) .await; HostReply::Ok { value: serde_json::json!({"status": "sent"}), } } _ => HostReply::error("unhandled", "unhandled request"), } }) });
// Execute work command on the worker: let call_res = worker .call( WorkCommand::Execute { id: OperationId::fresh(), code: "print('attack')".into(), purpose: Purpose::Interactive, }, std::time::Duration::from_secs(5), service, ) .await;
// 1. Assert the in-flight exchange was strictly refused with the typed stale generation error assert!( call_res.is_err(), "Call must fail when worker frame carries a stale prior generation" ); let err_str = format!("{:#}", call_res.err().unwrap()); assert!( err_str.contains("stale worker envelope") || err_str.contains("stale"), "Error must explicitly identify stale worker envelope: {err_str}" );
// 2. Assert worker generation was invalidated assert!( !worker.is_alive(), "Worker generation must be invalidated on protocol/generation violation" );
// 3. ASSERT DOMAIN STATE IS BYTE-FOR-BYTE UNCHANGED: let events_after = store.history(&session, 0, 100).await?; assert_eq!( events_before.len(), events_after.len(), "Store event count must be completely unchanged" ); assert_eq!( events_before[0].id, events_after[0].id, "Last event sequence must match" ); assert_eq!( events_before[0].payload, events_after[0].payload, "Event payload must match" );
// Ensure no local.sent or mutation event was ever inserted assert!( !events_after.iter().any(|e| e.kind == "local.sent"), "No unauthorized domain event may be committed from a stale generation frame" );
Ok(())}
#[tokio::test]async fn test_u13_live_relay_control_injector_prior_generation_rejected_store_unchanged() -> Result<()> { use klbr_runtime::{ behavior::Release, protocol::{ Envelope, Generation, HostRequest, Message, OperationId, Outcome, RequestId, SessionId, }, remote::relay::RemoteBackend, session::{Placement, SessionRuntime}, store::Store, target::{NamespaceId, RemoteTarget, RemoteWorkspace, TargetId}, worker::WorkerConfig, };
let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string()); let ssh_config = SshConfig::new("localhost").with_user(&user);
let temp = tempfile::tempdir()?; let store = Store::open(temp.path().join("u13_live_injector.sqlite"))?; let session = SessionId::fresh(); store.ensure_session(&session).await?;
let raw_ws = temp.path().join("workspace"); std::fs::create_dir_all(&raw_ws)?; let ws_path = raw_ws.canonicalize()?; std::fs::copy( fixtures_root().join("workspace/remote_marker.json"), ws_path.join("remote_marker.json"), )?;
let target = RemoteTarget { id: TargetId::parse("loopback-target-u13-inj")?, ssh_alias: "localhost".into(), account: Some(user.clone()), workspace: RemoteWorkspace { path: ws_path.to_string_lossy().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), };
let relay_bin = repo_root().join("target/debug/klbr-relay"); let python_bin = PathBuf::from("/opt/homebrew/bin/python3"); let runner_script = repo_root().join("python/klbr-runtime/src/klbr_runtime/__main__.py");
let raw_backend = RemoteBackend::new( ssh_config, target.clone(), relay_bin, python_bin, runner_script, ) .with_bundle_dir(fixtures_root().join("behavior"));
let backend = Arc::new(raw_backend);
let placement = Placement::remote( target.clone(), backend.clone(), NamespaceId::parse("default")?, fixtures_root() .join("behavior") .to_string_lossy() .to_string(), );
let stamp = match &placement { Placement::Remote { binding, .. } => binding.clone(), Placement::Local => panic!("this test is about a remote placement"), };
let initial_release = Release::publish( &fixtures_root().join("behavior"), &temp.path().join("releases"), )?;
let config = WorkerConfig::source_tree(&repo_root()); let runtime = SessionRuntime::start_with_placement( store.clone(), session.clone(), config, initial_release, placement, ) .await?;
// 1. Record store state before attack let events_before = store.history(&session, 0, 100).await?; assert!( !events_before.is_empty(), "Store must have initial events (kernel.reset/placement)" );
// 2. Obtain live control injector directly from the production RemoteBackend let injector = backend .active_injector() .await .expect("live relay injector must be active");
// 3. ATTACK: Inject a frame stamped with an EXPIRED / PRIOR generation attempting to mutate store via local.send! let prior_stale_generation = Generation::fresh(); let stale_envelope = Envelope { version: 1, session_id: session.clone(), generation: prior_stale_generation, // STALE / PRIOR GENERATION target: Some(stamp.clone()), message: Message::HostRequest { id: RequestId::fresh(), execution_id: OperationId::fresh(), request: HostRequest::LocalSend { text: "FORGED_UNAUTHORIZED_STALE_MESSAGE".into(), }, }, };
// Inject directly into the host Worker control channel production read path injector.inject_inbound_envelope(&stale_envelope).await?;
// POSITIVE REJECTION SIGNAL: // A subsequent work execution through the production session runtime forces the host // exchange to process the inbound control queue. The injected stale frame is refused by // the validator, which invalidates the worker generation and causes the call to fail. let cell_res = runtime .execute("1 + 1".into(), std::time::Duration::from_secs(5)) .await; let report = cell_res.expect("execution report must be returned"); let error = match &report.outcome { Outcome::Failed { error } | Outcome::Unknown { reason: error } => error, outcome => panic!("Execution outcome must retain stale-frame rejection cause: {outcome:?}"), }; assert!( error.contains("stale worker envelope") || error.contains("stale"), "Error must explicitly identify stale worker envelope: {error}" );
// 4. ASSERT DOMAIN STATE INTEGRITY: // The unauthorized host request (LocalSend) was REFUSED and never committed to store! let events_after = store.history(&session, 0, 100).await?; // The store legitimately GROWS here: the positive-signal cell above is a real // execution and commits its own events. The invariant is not "nothing was // appended" but "nothing attributable to the STALE FRAME was appended": the // prior history is byte-identical, and every new event belongs to the // legitimate execution rather than to the injected `local.send`. assert!( events_after.len() >= events_before.len(), "history must never shrink" ); for extra in &events_after[events_before.len()..] { assert!( matches!( extra.kind.as_str(), "execution.started" | "execution.completed" | "execution.unknown" ), "the only new events may come from the legitimate execution, got {:?}", extra.kind ); } for i in 0..events_before.len() { assert_eq!( events_before[i].id, events_after[i].id, "Event IDs must match" ); assert_eq!( events_before[i].kind, events_after[i].kind, "Event kinds must match" ); assert_eq!( events_before[i].payload, events_after[i].payload, "Event payloads must match" ); assert_eq!( events_before[i].created_ms, events_after[i].created_ms, "Event timestamps must match" ); } assert!( !events_after.iter().any(|e| e.kind == "local.sent"), "No unauthorized domain event may be committed from a stale generation frame" );
// Clean shutdown runtime.shutdown().await;
Ok(())}