Run and inspect a real remote Python namespace without pretending about loss
16 unmodified lines1718192021222316 unmodified lines*.db.beads-credential-key
__pycache__/
# Python build metadata regenerated by wheel builds (U33 packaging smoke)*.egg-info/175 unmodified linesExpand all176177178179180181182183184175 unmodified linesExpand all{"id":"int-a957c6ddaa1db4030e9b70a17773797d","kind":"field_change","created_at":"2026-09-11T06:00:37.917619Z","actor":"dawn","issue_id":"klbr-1mz","extra":{"field":"status","new_value":"closed","old_value":"in_progress"}}{"id":"int-d87a805b64c280c43e8b13512c1252ac","kind":"field_change","created_at":"2026-09-11T06:00:46.541673Z","actor":"dawn","issue_id":"klbr-c3e","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-0286715a372fd79de32dabc3af1d96cc","kind":"field_change","created_at":"2026-09-11T06:49:01.024429Z","actor":"dawn","issue_id":"klbr-5il","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-e52d8f63e72d7b184af6c8a519f55a71","kind":"field_change","created_at":"2026-09-11T06:49:08.84523Z","actor":"dawn","issue_id":"klbr-kla","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-7f0e1be37b66868b55b2f54619c6e927","kind":"field_change","created_at":"2026-09-11T16:29:35.063433Z","actor":"dawn","issue_id":"klbr-9fd","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-e200a1874eda590078568dd696161b1f","kind":"field_change","created_at":"2026-09-11T16:29:47.402971Z","actor":"dawn","issue_id":"klbr-hc0","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-5e17da4e85e0bfa5ef6322f033a07a11","kind":"field_change","created_at":"2026-09-11T16:29:50.281978Z","actor":"dawn","issue_id":"klbr-inv","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-5499c483d6408f447fbe317fc4ab2f21","kind":"field_change","created_at":"2026-09-11T16:29:51.35083Z","actor":"dawn","issue_id":"klbr-aza","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-3c87c92b0b6f7aa74ad8b716a9e58dc2","kind":"field_change","created_at":"2026-09-11T16:29:52.407978Z","actor":"dawn","issue_id":"klbr-5uv","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594//! m2 callback gate: host policy callbacks must not serialize the remote workbench.//!//! this drives the real daemon websocket, real loopback ssh transport, and the//! actual python workbench. the callback hook blocks, while ping/input ingress,//! candidate validation, and hook activation continue independently.
use anyhow::{ensure, Result};use futures::{SinkExt, StreamExt};use klbr_core::{ memory::MemoryStore, models::{LlmClient, ModelConfig, ModelsConfig}, AgentEvent, MetricsSnapshot,};use klbr_daemon::{ daemon::{parse_session_id, serve_listener}, session_registry::{PlacementSource, SessionRegistry},};use klbr_ipc::v2::{V2ClientMsg, V2ServerMsg};use klbr_runtime::{ behavior::Release, protocol::SessionId, remote::{ relay::RemoteBackend, ssh::{SshConfig, SshTransport}, }, session::Placement, store::Store, target::{NamespaceId, RemoteTarget, RemoteWorkspace, TargetId}, worker::WorkerConfig,};use std::{ fs, path::{Path, PathBuf}, sync::Arc, time::Duration,};use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::TcpListener, sync::{broadcast, mpsc, watch, RwLock},};use tokio_tungstenite::{connect_async, tungstenite::protocol::Message as WsMessage};
fn repo_root() -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() .expect("repo root") .to_path_buf()}
struct LoopbackPlacementSource { placement: Placement,}
impl PlacementSource for LoopbackPlacementSource { fn placement( &self, _session: &SessionId, ) -> futures::future::BoxFuture<'_, anyhow::Result<Option<Placement>>> { let placement = self.placement.clone(); Box::pin(async move { Ok(Some(placement)) }) }}
struct TestDaemon { addr: std::net::SocketAddr, store: Store, registry: Arc<SessionRegistry>, shutdown_tx: watch::Sender<bool>, handle: tokio::task::JoinHandle<()>,}
impl TestDaemon { async fn stop_and_join(self) -> Result<()> { let _ = self.shutdown_tx.send(true); self.registry.shutdown_all().await; let _ = self.handle.await; Ok(()) }}
async fn start_daemon( temp: &tempfile::TempDir, llm: LlmClient, initial_release: Release, placement_source: Arc<dyn PlacementSource>,) -> Result<TestDaemon> { let listener = TcpListener::bind("127.0.0.1:0").await?; let addr = listener.local_addr()?; let store = Store::open(temp.path().join("daemon.sqlite"))?; let registry = Arc::new( SessionRegistry::new(store.clone(), WorkerConfig::source_tree(&repo_root())) .with_placement(placement_source), ); let mem_store = MemoryStore::open(temp.path().join("memory.sqlite").to_str().unwrap(), 64)?; let snapshot = Arc::new(RwLock::new(None)) as MetricsSnapshot; let (interrupt_tx, _) = mpsc::channel(32); let (output_tx, _) = broadcast::channel::<AgentEvent>(1024); let (v2_event_tx, _) = broadcast::channel::<V2ServerMsg>(1024); let (shutdown_tx, shutdown_rx) = watch::channel(false); let registry_task = registry.clone(); let v2_task = v2_event_tx.clone(); let handle = tokio::spawn(async move { let _ = serve_listener( listener, interrupt_tx, output_tx, v2_task, snapshot, mem_store, 50, "m2-callback-gate".to_string(), registry_task, Some(llm), Some(initial_release), true, shutdown_rx, ) .await; }); Ok(TestDaemon { addr, store, registry, shutdown_tx, handle, })}
fn py_string(path: &Path) -> String { serde_json::to_string(&path.display().to_string()).expect("path is valid json")}
fn write_behavior(root: &Path, module: &str, source: &str) -> Result<()> { fs::create_dir_all(root)?; fs::write( root.join("manifest.json"), format!( r##"{{"version":1,"hooks":{{"attention.plan":"{module}:attention","delivery.review":"{module}:delivery"}}}}"## ), )?; let module_path = root.join(format!("{}.py", module.replace('.', "/"))); if let Some(parent) = module_path.parent() { fs::create_dir_all(parent)?; } fs::write(module_path, source)?; Ok(())}
async fn spawn_mock_llm() -> Result<std::net::SocketAddr> { let listener = TcpListener::bind("127.0.0.1:0").await?; let addr = listener.local_addr()?; tokio::spawn(async move { loop { let Ok((mut socket, _)) = listener.accept().await else { break; }; tokio::spawn(async move { let mut buf = vec![0_u8; 131_072]; let n = socket.read(&mut buf).await.unwrap_or(0); let request = String::from_utf8_lossy(&buf[..n]); let body = request .split_once("\r\n\r\n") .map(|(_, body)| body) .unwrap_or(request.as_ref()); let value: serde_json::Value = serde_json::from_str(body).unwrap_or_default(); let last_role = value .get("messages") .and_then(|messages| messages.as_array()) .and_then(|messages| messages.last()) .and_then(|message| message.get("role")) .and_then(|role| role.as_str()); let body = if last_role == Some("tool") { let chunk = serde_json::json!({ "choices": [{"delta": {"role": "assistant", "content": "callback gate complete"}}] }); format!("data: {}\n\ndata: [DONE]\n\n", chunk) } else { let code = r#"from klbr import localimport threadingimport time
global u11_callback_stateif "u11_callback_state" not in globals(): u11_callback_state = 4242
callback_finished = threading.Event()global callback_tickscallback_ticks = 0def run_separate_computation(): global callback_ticks while not callback_finished.is_set(): callback_ticks += sum(value * value for value in range(256)) time.sleep(0)
computation_thread = threading.Thread(target=run_separate_computation)computation_thread.start()receipt = await local.send("callback lane test")assert u11_callback_state == 4242callback_finished.set()computation_thread.join(timeout=2)print(f"M2_CALLBACK_COMPUTATION_TICKS:{callback_ticks}")"#; let args = serde_json::json!({"code": code}); let chunk = serde_json::json!({ "choices": [{"delta": {"tool_calls": [{ "index": 0, "id": "m2_callback_tool_call", "type": "function", "function": {"name": "python", "arguments": args.to_string()} }]}}] }); format!("data: {}\n\ndata: [DONE]\n\n", chunk) }; let response = format!( "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\nconnection: close\r\n\r\n{body}" ); let _ = socket.write_all(response.as_bytes()).await; }); } }); Ok(addr)}
async fn wait_until<F>(timeout: Duration, mut condition: F) -> Result<()>where F: FnMut() -> bool,{ let deadline = tokio::time::Instant::now() + timeout; while tokio::time::Instant::now() < deadline { if condition() { return Ok(()); } tokio::time::sleep(Duration::from_millis(25)).await; } anyhow::bail!("condition did not become true within {timeout:?}")}
async fn wait_message<F>( messages: &Arc<tokio::sync::Mutex<Vec<V2ServerMsg>>>, timeout: Duration, mut condition: F,) -> Result<()>where F: FnMut(&[V2ServerMsg]) -> bool,{ let deadline = tokio::time::Instant::now() + timeout; while tokio::time::Instant::now() < deadline { if condition(&messages.lock().await) { return Ok(()); } tokio::time::sleep(Duration::from_millis(25)).await; } anyhow::bail!("websocket condition did not become true within {timeout:?}")}
async fn wait_for_execution_completed( store: &Store, session_id: &SessionId, timeout: Duration,) -> Result<()> { let deadline = tokio::time::Instant::now() + timeout; while tokio::time::Instant::now() < deadline { if store .history(session_id, 0, 256) .await? .iter() .any(|event| event.kind == "execution.completed") { return Ok(()); } tokio::time::sleep(Duration::from_millis(25)).await; } anyhow::bail!("execution did not complete within {timeout:?}")}
#[tokio::test]async fn u11_remote_callback_lane_and_u14_role_activation_gate() -> Result<()> { let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string()); let ssh = SshConfig::new("localhost").with_user(&user); let mut probe = SshTransport::spawn(&ssh, "printf M2_CALLBACK_SSH_OK")?; let mut probe_output = Vec::new(); probe.stdout.read_to_end(&mut probe_output).await?; ensure!(probe.wait().await?.success(), "loopback ssh must succeed"); ensure!( String::from_utf8_lossy(&probe_output).contains("M2_CALLBACK_SSH_OK"), "loopback ssh probe receipt missing" );
let temp = tempfile::tempdir()?; let workspace = temp.path().join("remote-workspace"); fs::create_dir_all(&workspace)?; let initial_source = temp.path().join("initial-source"); let initial_marker = temp.path().join("initial-hook-entered"); let candidate_marker = temp.path().join("candidate-hook-entered"); let gate = temp.path().join("release-old-callback"); let initial = format!( r#"from pathlib import Pathimport timefrom klbr.hooks import AttentionDecision, DeliveryDecision
def attention(request, context): return AttentionDecision.wake("operator")
def delivery(request, context): Path({initial_marker}).write_text("initial hook entered") while not Path({gate}).exists(): time.sleep(0.01) return DeliveryDecision.allow()"#, initial_marker = py_string(&initial_marker), gate = py_string(&gate), ); write_behavior(&initial_source, "klbr_hooks.initial_hooks", &initial)?; let initial_release = Release::publish(&initial_source, &temp.path().join("releases"))?;
let candidate_source = temp.path().join("candidate-source"); let candidate = format!( r#"from pathlib import Pathfrom klbr.hooks import AttentionDecision, DeliveryDecision
def attention(request, context): return AttentionDecision.wake("operator")
def delivery(request, context): Path({candidate_marker}).write_text("candidate hook entered") return DeliveryDecision.allow()"#, candidate_marker = py_string(&candidate_marker), ); write_behavior(&candidate_source, "klbr_hooks.candidate_hooks", &candidate)?; let candidate_release = Release::publish(&candidate_source, &temp.path().join("releases"))?;
let broken_source = temp.path().join("broken-source"); write_behavior( &broken_source, "klbr_hooks.broken_hooks", "def attention(request, context):\n return (\n", )?; let broken_release = Release::publish(&broken_source, &temp.path().join("releases"))?;
let target = RemoteTarget { id: TargetId::parse("m2-callback-loopback")?, ssh_alias: "localhost".into(), account: Some(user.clone()), workspace: RemoteWorkspace { path: workspace.canonicalize()?.display().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), }; let placement = Placement::remote( target.clone(), Arc::new( RemoteBackend::new( ssh, target.clone(), repo_root().join("target/debug/klbr-relay"), PathBuf::from("/opt/homebrew/bin/python3"), repo_root().join("python/klbr-runtime/src/klbr_runtime/__main__.py"), ) .with_bundle_dir(repo_root().join("tests/fixtures/remote-loopback/behavior")), ), NamespaceId::parse("m2-callback")?, repo_root() .join("tests/fixtures/remote-loopback/behavior") .display() .to_string(), )?; let placement_source = Arc::new(LoopbackPlacementSource { placement }); let llm_addr = spawn_mock_llm().await?; let llm = LlmClient::new(ModelsConfig { llm: ModelConfig { url: format!("http://{llm_addr}"), model: "m2-callback-mock".into(), ..Default::default() }, ..Default::default() }); let daemon = start_daemon(&temp, llm, initial_release.clone(), placement_source).await?; let session_text = "m2_callback_session"; let session_id = parse_session_id(session_text); let entry = daemon .registry .get_or_open(&session_id, &initial_release) .await?; let runtime = entry.runtime.clone(); let old_policy = runtime.policy().await; let old_hook_worker_generation = old_policy.worker.generation().as_str().to_string(); let before_identity = runtime.target_identity().await;
let (stream, _) = connect_async(format!("ws://{}", daemon.addr)).await?; let (mut write, mut read) = stream.split(); let messages = Arc::new(tokio::sync::Mutex::new(Vec::new())); let messages_task = messages.clone(); let reader = tokio::spawn(async move { while let Some(Ok(WsMessage::Text(text))) = read.next().await { if let Ok(message) = serde_json::from_str::<V2ServerMsg>(&text) { messages_task.lock().await.push(message); } } }); write .send(WsMessage::Text(serde_json::to_string( &V2ClientMsg::Subscribe { session_id: session_text.into(), after_seq: Some(0), }, )?)) .await?; wait_message(&messages, Duration::from_secs(5), |messages| { messages .iter() .any(|message| matches!(message, V2ServerMsg::SyncCompleted { .. })) }) .await?;
write .send(WsMessage::Text(serde_json::to_string( &V2ClientMsg::SendInput { session_id: session_text.into(), source: "operator".into(), text: "run the blocked callback cell".into(), client_request_id: Some("m2-first".into()), }, )?)) .await?; wait_message(&messages, Duration::from_secs(5), |messages| { messages.iter().any(|message| { matches!(message, V2ServerMsg::Ack { client_request_id: Some(id), .. } if id == "m2-first") }) }) .await?; // the host-side hook marker is the positive receipt that the remote cell // reached local.send; it avoids another target filesystem evidence path. wait_until(Duration::from_secs(5), || initial_marker.exists()).await?;
// a blocked host callback does not prevent a second ingress request or ping. write .send(WsMessage::Text(serde_json::to_string( &V2ClientMsg::Ping { timestamp: Some(11), }, )?)) .await?; write .send(WsMessage::Text(serde_json::to_string( &V2ClientMsg::SendInput { session_id: session_text.into(), source: "operator".into(), text: "queue another input while callback is blocked".into(), client_request_id: Some("m2-second".into()), }, )?)) .await?; wait_message(&messages, Duration::from_secs(5), |messages| { let pong = messages .iter() .any(|message| matches!(message, V2ServerMsg::Pong { timestamp: Some(11) })); let ack = messages.iter().any(|message| { matches!(message, V2ServerMsg::Ack { client_request_id: Some(id), .. } if id == "m2-second") }); pong && ack }) .await?;
let history_while_blocked = daemon.store.history(&session_id, 0, 256).await?; let started_while_blocked: Vec<_> = history_while_blocked .iter() .filter(|event| event.kind == "execution.started") .collect(); ensure!( started_while_blocked.len() == 1, "callback waiting must not admit a second foreground execution" ); let first_execution = started_while_blocked[0] .payload .get("execution_id") .and_then(|value| value.as_str()) .expect("execution.started operation") .to_string();
// Reading runtime state here and completing the durable candidate activation // below while the callback is blocked is the concrete lock-order receipt: // no store transaction or session lock crosses the callback await. let identity_during_callback = runtime.target_identity().await; ensure!( identity_during_callback.generation == before_identity.generation, "workbench state remains available while callback waits" );
// U14: a malformed candidate is refused while the old policy remains active. ensure!( runtime.activate_candidate(1, broken_release).await.is_err(), "broken candidate must be rejected" ); ensure!( runtime.policy().await.release.id == initial_release.id, "broken candidate changed the active release" );
// Candidate validation and hook-worker replacement are independent of the // blocked old callback and do not replace the remote workbench generation. let new_epoch = runtime .activate_candidate(1, candidate_release.clone()) .await?; ensure!(new_epoch == 2, "good candidate must advance epoch to 2"); wait_until(Duration::from_secs(5), || candidate_marker.exists()).await?; let new_policy = runtime.policy().await; ensure!( new_policy.release.id == candidate_release.id, "candidate is active" ); ensure!( new_policy.worker.generation().as_str() != old_hook_worker_generation, "hook activation must replace the hook worker role" ); ensure!( runtime.target_identity().await.generation == before_identity.generation, "hook activation must not reset the unrelated workbench generation" );
// Release the old callback. Its immutable context must retain the old // policy revision even though a new policy is now active. fs::write(&gate, "release")?; wait_for_execution_completed(&daemon.store, &session_id, Duration::from_secs(15)).await?; let completed = daemon .store .history(&session_id, 0, 256) .await? .into_iter() .find(|event| event.kind == "execution.completed") .expect("execution.completed event"); let output = completed .payload .get("outcome") .and_then(|outcome| outcome.get("output")) .and_then(|output| output.as_array()) .into_iter() .flatten() .filter_map(|item| item.get("text").and_then(|text| text.as_str())) .collect::<String>(); let completed_ticks = output .split("M2_CALLBACK_COMPUTATION_TICKS:") .nth(1) .and_then(|ticks| ticks.trim().parse::<u64>().ok()) .expect("computation tick output"); ensure!( completed_ticks > 1_000, "separate computation must do real work while callback waits" );
let context = old_policy .last_callback_context() .await .expect("old callback must retain its context"); ensure!( context.session_id == session_id, "callback session attribution" ); ensure!( context.execution_id.as_str() == first_execution, "callback execution attribution" ); ensure!( context.generation.as_str() == before_identity.generation, "callback generation attribution" ); ensure!( context.policy_revision == initial_release.id, "callback policy attribution" ); match context.target { klbr_runtime::remote::callback::CallbackTarget::Remote { stamp } => { ensure!( stamp.target_id.as_str() == "m2-callback-loopback", "callback target attribution" ); } other => anyhow::bail!("remote callback carried wrong target: {other:?}"), } let stats = old_policy.callback_stats(); ensure!( stats.submitted >= 1 && stats.serviced >= 1, "callback broker receipt missing" );
drop(write); reader.abort(); daemon.stop_and_join().await?; Ok(())}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493//! m2 gate u17: the normal remote path must not invoke the agent-facing//! `ssh_control` fallback. this drives real code through the host-owned//! `RemoteBackend` and persistent remote worker, while a failing executable//! named `ssh_control` is first on the target process PATH.//!//! the trap is not the transport: `SshTransport` and `RemoteBackend` remain//! production transport code. it only detects an illicit agent-facing fallback.
use anyhow::{Context, Result};use klbr_runtime::{ behavior::Release, protocol::Outcome, remote::{relay::RemoteBackend, ssh::SshConfig}, session::{Placement, SessionRuntime}, store::Store, target::{NamespaceId, RemoteTarget, RemoteWorkspace, TargetId}, worker::WorkerConfig,};use std::{ fs, os::unix::fs::PermissionsExt, path::{Path, PathBuf}, process::Command, sync::Arc, time::Duration,};
fn repo_root() -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() .expect("repo root") .to_path_buf()}
fn source_output(report: &klbr_runtime::session::ExecutionReport) -> String { match &report.outcome { Outcome::Cell { output, .. } => output .iter() .map(|entry| entry.text.as_str()) .collect::<String>(), other => panic!("remote workflow returned non-cell outcome: {other:?}"), }}
fn quote_python(value: &Path) -> String { serde_json::to_string(&value.display().to_string()).expect("path is valid json")}
fn quote_shell(value: &Path) -> String { format!("'{}'", value.display().to_string().replace('\'', "'\\''"))}
#[tokio::test]async fn test_u17_normal_remote_inspect_edit_build_test_has_zero_ssh_control_invocations() -> Result<()> { let root = repo_root(); let temp = tempfile::tempdir()?; let workspace = temp.path().join("remote-workspace"); let trap_bin = workspace.join("agent-path"); let trap = trap_bin.join("ssh_control"); let invocations = workspace.join("ssh_control.invocations"); fs::create_dir_all(&trap_bin)?; fs::write( workspace.join("remote_marker.json"), r#"{"marker":"U17_LOOPBACK_TARGET"}"#, )?;
// the trap fails loudly if the forbidden route is reached. fs::write( &trap, format!( "#!/bin/sh\nprintf '%s\\n' invoked >> {}\nprintf '%s\\n' 'ssh_control fallback trap invoked' >&2\nexit 97\n", quote_shell(&invocations) ), )?; fs::set_permissions(&trap, fs::Permissions::from_mode(0o755))?;
let release = Release::publish( &root.join("tests/fixtures/remote-loopback/behavior"), &temp.path().join("releases"), )?; let user = std::env::var("USER").unwrap_or_else(|_| "dawn".into()); let ssh = SshConfig::new("localhost").with_user(&user); let target = RemoteTarget { id: TargetId::parse("u17-loopback-target")?, ssh_alias: "localhost".into(), account: Some(user), workspace: RemoteWorkspace { path: workspace.canonicalize()?.display().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), }; let backend = Arc::new( RemoteBackend::new( ssh, target.clone(), root.join("target/debug/klbr-relay"), PathBuf::from("/opt/homebrew/bin/python3"), root.join("python/klbr-runtime/src/klbr_runtime/__main__.py"), ) .with_bundle_dir(root.join("tests/fixtures/remote-loopback/behavior")), ); let placement = Placement::remote( target, backend, NamespaceId::parse("u17")?, root.join("tests/fixtures/remote-loopback/behavior") .display() .to_string(), )?; let db = temp.path().join("u17.sqlite"); let store = Store::open(&db)?; let runtime = SessionRuntime::start_with_placement( store, "u17-session".parse().map_err(anyhow::Error::msg)?, WorkerConfig::source_tree(&root), release, placement, ) .await?;
let trap_path = quote_python(&trap); let workspace_path = quote_python(&workspace.canonicalize()?); let first = runtime .execute( format!( r#"import json, os, pathlib, shutil, systrap = pathlib.Path({trap_path})workspace = pathlib.Path({workspace_path})os.environ["PATH"] = str(trap.parent) + os.pathsep + os.environ["PATH"]assert pathlib.Path.cwd().resolve() == workspaceassert shutil.which("ssh_control") == str(trap)assert sys.platform == "darwin"assert json.loads((workspace / "remote_marker.json").read_text())["marker"] == "U17_LOOPBACK_TARGET"(workspace / "u17_workflow.py").write_text( "from pathlib import Path\n" "Path('u17.remote').write_text('edited through remote kernel')\n" "print('U17_WORKFLOW_TEST_OK')\n")print("U17_INSPECT_OK", os.getpid(), pathlib.Path.cwd())"#, trap_path = trap_path, workspace_path = workspace_path, ), Duration::from_secs(15), ) .await?; let first_output = source_output(&first); assert!(first_output.contains("U17_INSPECT_OK"));
let second = runtime .execute( r#"import pathlib, subprocess, syssource = pathlib.Path("u17_workflow.py")compile_result = subprocess.run( [sys.executable, "-m", "py_compile", str(source)], capture_output=True, text=True,)assert compile_result.returncode == 0, compile_result.stderrrun_result = subprocess.run( [sys.executable, str(source)], capture_output=True, text=True,)assert run_result.returncode == 0, run_result.stderrassert "U17_WORKFLOW_TEST_OK" in run_result.stdoutassert pathlib.Path("u17.remote").read_text() == "edited through remote kernel"print("U17_BUILD_TEST_OK")"# .into(), Duration::from_secs(15), ) .await?; assert!(source_output(&second).contains("U17_BUILD_TEST_OK"));
let remote_marker = workspace.join("u17.remote"); assert_eq!( fs::read_to_string(&remote_marker)?, "edited through remote kernel" ); let count = invocations .try_exists() .context("check ssh_control trap receipt")? .then(|| fs::read_to_string(&invocations)) .transpose()? .unwrap_or_default(); assert!(count.is_empty(), "ssh_control trap was invoked: {count}"); runtime.shutdown().await; Ok(())}
#[tokio::test]async fn test_u33_built_wheel_imports_outside_checkout_and_remote_target_runs_python_runner() -> Result<()> { let root = repo_root(); let temp = tempfile::tempdir()?; let dist = temp.path().join("dist"); let install = temp.path().join("outside-install"); let outside = temp.path().join("outside-checkout"); fs::create_dir_all(&outside)?;
let build = Command::new("uv") .args(["build", "--out-dir"]) .arg(&dist) .current_dir(root.join("python/klbr-runtime")) .output() .context("build klbr-runtime wheel")?; assert!( build.status.success(), "uv build failed: {}", String::from_utf8_lossy(&build.stderr) ); let wheel = fs::read_dir(&dist)? .filter_map(|entry| entry.ok().map(|entry| entry.path())) .find(|path| path.extension().is_some_and(|ext| ext == "whl")) .context("uv build produced no wheel")?; let install_result = Command::new("uv") .args(["pip", "install", "--target"]) .arg(&install) .arg("--no-compile") .arg(&wheel) .current_dir(&outside) .output() .context("install built wheel outside checkout")?; assert!( install_result.status.success(), "wheel install failed: {}", String::from_utf8_lossy(&install_result.stderr) );
let probe = outside.join("wheel_probe.py"); fs::write( &probe, r#"import importlib.resourcesimport pathlibimport sys
install = pathlib.Path(sys.argv[1]).resolve()checkout = pathlib.Path(sys.argv[2]).resolve()sys.path.insert(0, str(install))import klbrimport klbr_runtime
assert pathlib.Path(klbr.__file__).resolve().is_relative_to(install)assert pathlib.Path(klbr_runtime.__file__).resolve().is_relative_to(install)assert not pathlib.Path(klbr.__file__).resolve().is_relative_to(checkout)assert not pathlib.Path(klbr_runtime.__file__).resolve().is_relative_to(checkout)assert importlib.resources.files("klbr").joinpath("py.typed").is_file()template = importlib.resources.files("klbr.templates").joinpath("maintenance-attempt.md")assert template.is_file() and template.read_text().strip()print("U33_WHEEL_IMPORT_OK")"#, )?; let probe_result = Command::new(std::env::var_os("KLBR_PYTHON").unwrap_or_else(|| "python3".into())) .args(["-I"]) .arg(&probe) .arg(&install) .arg(&root) .current_dir(&outside) .env_remove("PYTHONPATH") .output() .context("import wheel from outside checkout")?; assert!( probe_result.status.success(), "wheel probe failed: {}", String::from_utf8_lossy(&probe_result.stderr) ); assert!(String::from_utf8_lossy(&probe_result.stdout).contains("U33_WHEEL_IMPORT_OK"));
let workspace = temp.path().join("remote-workspace"); fs::create_dir_all(&workspace)?; let release = Release::publish( &root.join("tests/fixtures/remote-loopback/behavior"), &temp.path().join("releases"), )?; let user = std::env::var("USER").unwrap_or_else(|_| "dawn".into()); let target = RemoteTarget { id: TargetId::parse("u33-wheel-target")?, ssh_alias: "localhost".into(), account: Some(user.clone()), workspace: RemoteWorkspace { path: workspace.canonicalize()?.display().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), }; let python = PathBuf::from( std::env::var_os("KLBR_PYTHON").unwrap_or_else(|| "/opt/homebrew/bin/python3".into()), ); let runner = install.join("klbr_runtime/__main__.py"); assert!( runner.is_file(), "wheel omitted target runner: {}", runner.display() ); let backend = Arc::new( RemoteBackend::new( SshConfig::new("localhost").with_user(&user), target.clone(), root.join("target/debug/klbr-relay"), python, runner.clone(), ) .with_bundle_dir(root.join("tests/fixtures/remote-loopback/behavior")), ); let placement = Placement::remote( target, backend, NamespaceId::parse("u33")?, root.join("tests/fixtures/remote-loopback/behavior") .display() .to_string(), )?; let runtime = SessionRuntime::start_with_placement( Store::open(temp.path().join("u33.sqlite"))?, "u33-session".parse().map_err(anyhow::Error::msg)?, WorkerConfig::source_tree(&root), release, placement, ) .await?; let runner_json = quote_python(&runner); let report = runtime .execute( format!( r#"import pathlib, shlex, subprocess, sysrunner = pathlib.Path({runner_json}).resolve()assert pathlib.Path(sys.argv[0]).resolve() == runnerchild = subprocess.run( ["ps", "-ww", "-p", str(__import__("os").getpid()), "-o", "command="], capture_output=True, text=True,)parent = subprocess.run( ["ps", "-ww", "-p", str(__import__("os").getppid()), "-o", "command="], capture_output=True, text=True,)assert child.returncode == 0 and runner.name in child.stdoutassert parent.returncode == 0 and "klbr-relay" in parent.stdoutcommands = child.stdout + parent.stdoutlines = [line for line in commands.splitlines() if line.strip()]executables = [pathlib.Path(shlex.split(line)[0]).name for line in lines]assert len(executables) == 2assert executables[0].lower().startswith("python") and executables[1] == "klbr-relay"assert set(executables).isdisjoint({{"node", "bun", "deno"}})print("U33_TARGET_PYTHON_RUNNER_OK", child.stdout.strip(), parent.stdout.strip())"#, runner_json = runner_json, ), Duration::from_secs(15), ) .await?; let output = source_output(&report); assert!(output.contains("U33_TARGET_PYTHON_RUNNER_OK")); runtime.shutdown().await; Ok(())}
#[tokio::test]async fn test_u08_u09_real_remote_sdk_dispatch() -> Result<()> { let root = repo_root(); let temp = tempfile::tempdir()?; let workspace = temp.path().join("remote-workspace"); fs::create_dir_all(&workspace)?; let local_shadow = temp.path().join("host-only").join("u08_sdk.marker"); let release = Release::publish( &root.join("tests/fixtures/remote-loopback/behavior"), &temp.path().join("releases"), )?; let user = std::env::var("USER").unwrap_or_else(|_| "dawn".into()); let target = RemoteTarget { id: TargetId::parse("u08-loopback-target")?, ssh_alias: "localhost".into(), account: Some(user.clone()), workspace: RemoteWorkspace { path: workspace.canonicalize()?.display().to_string(), revision: None, }, interpreter: None, required_platform: Some("Darwin".into()), }; let backend = Arc::new( RemoteBackend::new( SshConfig::new("localhost").with_user(&user), target.clone(), root.join("target/debug/klbr-relay"), PathBuf::from( std::env::var_os("KLBR_PYTHON") .unwrap_or_else(|| "/opt/homebrew/bin/python3".into()), ), root.join("python/klbr-runtime/src/klbr_runtime/__main__.py"), ) .with_bundle_dir(root.join("tests/fixtures/remote-loopback/behavior")), ); let placement = Placement::remote( target, backend, NamespaceId::parse("u08")?, root.join("tests/fixtures/remote-loopback/behavior") .display() .to_string(), )?; let runtime = SessionRuntime::start_with_placement( Store::open(temp.path().join("u08.sqlite"))?, "u08-session".parse().map_err(anyhow::Error::msg)?, WorkerConfig::source_tree(&root), release, placement, ) .await?; let workspace_json = quote_python(&workspace.canonicalize()?); let report = runtime .execute( format!( r#"import osfrom klbr.remote import RemoteExecutionError, Unsupported, connect
async with connect("u08-loopback-target", workspace={workspace_json}) as remote: # These fields come from the relay's measured target probe. assert remote.identity.platform.lower() == "darwin" assert remote.identity.arch assert remote.identity.cwd == {workspace_json} assert remote.identity.account assert remote.capabilities["shell"].usable assert remote.capabilities["file_system"].usable assert remote.capabilities["imports"].usable assert remote.capabilities["skill_manifest"].revision == "remote-dispatch-v1"
marker = remote.pathlib.Path("u08_sdk.marker") await marker.write_text("U08_REMOTE_OK") assert await marker.exists() assert await marker.read_text() == "U08_REMOTE_OK"
key = "KLBR_U08_REMOTE_ONLY_ENV" os.environ.pop(key, None) await remote.os.environ.set(key, "remote") assert await remote.os.environ.get(key) == "remote" assert os.environ.get(key) is None
await remote.control.execute("import math") assert await remote.control.eval("{{'numbers': [1, 2, 3], 'factorial': math.factorial(6)}}") == {{"numbers": [1, 2, 3], "factorial": 720}} try: await remote.control.eval("1 / 0") except RemoteExecutionError as error: assert "ZeroDivisionError" in str(error) else: raise AssertionError("remote evaluation error was silently accepted")
def shipped_value(): return "should not run" try: await remote.control.call( shipped_value, trusted=True, dependencies=["klbr_module_that_is_not_installed"], ) except Unsupported: pass else: raise AssertionError("unadvertised closure dependency was not typed Unsupported")print("U08_U09_SDK_REAL_OK")"#, workspace_json = workspace_json, ), Duration::from_secs(60), ) .await?; assert!( source_output(&report).contains("U08_U09_SDK_REAL_OK"), "remote report: {report:?}" ); assert_eq!( fs::read_to_string(workspace.join("u08_sdk.marker"))?, "U08_REMOTE_OK" ); assert!( !local_shadow.exists(), "remote SDK write escaped into the host shadow path" ); assert!(!local_shadow.parent().unwrap().exists()); assert!( std::env::var_os("KLBR_U08_REMOTE_ONLY_ENV").is_none(), "remote environment mutation escaped into the host process" ); runtime.shutdown().await; Ok(())}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263-- Migration 009: add the M2 remote repair event and preserve uncertain-- executions as a distinct terminal state. SQLite cannot ALTER either CHECK-- constraint, so both tables are rebuilt under one foreign-key-disabled-- transaction. The pragma MUST remain outside BEGIN.PRAGMA foreign_keys = OFF;BEGIN IMMEDIATE;
CREATE TABLE events_v2 ( seq INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL REFERENCES sessions(id), kind TEXT NOT NULL CHECK(kind IN ( 'input','scratchpad','execution.started','execution.completed','execution.unknown', 'local.sent','delivery.held','kernel.reset','session.waited', 'session.woke','hooks.activated','kernel.placement','remote.repair' )), source_key TEXT, payload TEXT NOT NULL CHECK(json_valid(payload)), created_ms INTEGER NOT NULL CHECK(created_ms >= 0), UNIQUE(session_id, source_key)) STRICT;
INSERT INTO events_v2 (seq, session_id, kind, source_key, payload, created_ms)SELECT seq, session_id, kind, source_key, payload, created_ms FROM events;
CREATE TABLE executions_v2 ( id TEXT PRIMARY KEY, session_id TEXT NOT NULL REFERENCES sessions(id), generation TEXT NOT NULL, hook_revision TEXT NOT NULL REFERENCES releases(id), state TEXT NOT NULL CHECK(state IN ('running','ok','error','yielded','interrupted','unknown')), started_event INTEGER NOT NULL UNIQUE REFERENCES events(seq), completed_event INTEGER UNIQUE REFERENCES events(seq)) STRICT;
INSERT INTO executions_v2 (id, session_id, generation, hook_revision, state, started_event, completed_event)SELECT id, session_id, generation, hook_revision, state, started_event, completed_event FROM executions;
DROP TABLE executions;DROP TABLE events;ALTER TABLE events_v2 RENAME TO events;ALTER TABLE executions_v2 RENAME TO executions;
CREATE INDEX events_session_order ON events(session_id, seq);CREATE TRIGGER events_no_update BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'historical events are immutable');END;CREATE TRIGGER events_no_delete BEFORE DELETE ON events BEGIN SELECT RAISE(ABORT, 'historical events are immutable');END;CREATE UNIQUE INDEX one_running_execution ON executions(session_id) WHERE completed_event IS NULL;
-- FATAL orphan guard: any row from pragma_foreign_key_check aborts this-- migration before COMMIT. This catches existing FK children (inbox,-- executions, and effects), not only an empty fresh database.CREATE TEMP TABLE _orphan_guard (reason TEXT CHECK (reason IS NULL));INSERT INTO _orphan_guard (reason)SELECT 'orphaned ' || "table" || ' row ' || rowid || ' via fk ' || fkid || ' -> parent ' || parentFROM pragma_foreign_key_check;DROP TABLE _orphan_guard;
PRAGMA user_version = 9;COMMIT;PRAGMA foreign_keys = ON;12345678930 unmodified lines40414243444546474849505152515354555657585960616263646566676869707172737475767778798081828384858687888990919228 unmodified lines12112212312412512612712812913013113213313413513613713813914014114289909192939414314414514614714814915015115215315415515615715815916016116216316416516616716816917010210310417117217317417517666 unmodified lines//! Policies return proposals. Validation and failure disposition belong here.use crate::{ behavior::Release, protocol::*, remote::callback::{CallbackBroker, CallbackContext, CallbackRequest}, worker::{no_host_operations, WorkCommand, Worker, WorkerConfig},};use anyhow::{bail, ensure, Context, Result};use serde::{Deserialize, Serialize};30 unmodified lines
pub struct Policy { pub release: Release, pub worker: Worker, callbacks: CallbackBroker,}impl Policy { pub async fn start( config: &WorkerConfig, session: SessionId, release: Release, ) -> Result<Arc<Self>> { let worker = Worker::spawn(config, session, Role::Hooks, Some(&release)).await?; Ok(Arc::new(Self { release, worker })) let revision = release.id.clone(); let hook_worker = worker.clone(); let callbacks = CallbackBroker::new( 32, Arc::new(move |callback| { let hook_worker = hook_worker.clone(); let revision = revision.clone(); Box::pin(async move { ensure!( callback.context.policy_revision == revision, "callback policy revision is not active" ); let outcome = hook_worker .call( WorkCommand::Invoke { id: OperationId::fresh(), slot: callback.slot, input: callback.input, }, Duration::from_secs(2), no_host_operations(), ) .await?; match outcome { Outcome::Hook { value } => Ok(value), Outcome::Failed { error } => bail!("hook failed: {error}"), _ => bail!("invalid policy result"), } }) }), )?; Ok(Arc::new(Self { release, worker, callbacks, })) }
pub async fn validate_synthetic(&self) -> Result<()> { self.attention(1, "operator", 0)28 unmodified lines }
/// Unavailable required review means hold, never accidental approval. pub async fn review(&self, effect: &EffectId, text: &str) -> ReviewedDelivery { let value = self .invoke( HookSlot::Delivery, json!({"effect_id": effect, "destination":"operator", "text":text}), ) .await; Self::decode_review(value) }
/// Reviews an effect through the independent callback lane. The context is /// immutable for the originating execution and policy release. pub async fn review_with_context( &self, context: CallbackContext, effect: &EffectId, text: &str, ) -> ReviewedDelivery { let result = async { let value = self .invoke( HookSlot::Delivery, json!({"effect_id": effect, "destination":"operator", "text":text}), ) .await?; let request = CallbackRequest::new( context, HookSlot::Delivery, json!({"effect_id": effect, "destination":"operator", "text":text}), )?; self.callbacks.submit(request, Duration::from_secs(2)).await } .await; Self::decode_review(result) }
pub async fn last_callback_context(&self) -> Option<CallbackContext> { self.callbacks.last_context().await }
pub fn callback_stats(&self) -> crate::remote::callback::CallbackStats { self.callbacks.stats() }
fn decode_review(result: Result<Value>) -> ReviewedDelivery { let result = result.and_then(|value| { let decision: DeliveryDecision = serde_json::from_value(value)?; if let DeliveryDecision::Hold { reason } = &decision { ensure!( !reason.trim().is_empty() && reason.len() <= 2048, "invalid hold reason" ); } Ok::<_, anyhow::Error>(decision) } .await; Ok(decision) }); match result { Ok(decision) => ReviewedDelivery { decision, fault: None,66 unmodified lines