Place a session's workbench on a verified remote target over a host-owned relay
165 unmodified linesExpand all166167168169170171172173174175176177178179165 unmodified linesExpand all{"id":"int-1311fd723011a73c8a4cfc7d6bbddadb","kind":"field_change","created_at":"2026-09-11T05:02:37.455564Z","actor":"dawn","issue_id":"klbr-8jn","extra":{"field":"status","new_value":"closed","old_value":"in_progress"}}{"id":"int-59e864a8032046a90e0932981aed7393","kind":"field_change","created_at":"2026-09-11T05:03:19.190974Z","actor":"dawn","issue_id":"klbr-bhe","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-0bbbc8f3c7f715b1e86de3febd4ec58c","kind":"field_change","created_at":"2026-09-11T05:12:24.761458Z","actor":"dawn","issue_id":"klbr-dt6","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-8a4129800b246671611156d29ef1b092","kind":"field_change","created_at":"2026-09-11T05:21:19.069621Z","actor":"dawn","issue_id":"klbr-2hy","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-62898bdee0d64b20955f6d48df6eace8","kind":"field_change","created_at":"2026-09-11T05:27:27.547566Z","actor":"dawn","issue_id":"klbr-7su","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-6083489daad8c8aee7c64a7c92cbe075","kind":"field_change","created_at":"2026-09-11T05:46:36.378101Z","actor":"dawn","issue_id":"klbr-m7x","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-a6d7ddc8fe915b891efc4b9546ce6d26","kind":"field_change","created_at":"2026-09-11T05:46:37.244297Z","actor":"dawn","issue_id":"klbr-5il","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-2cf4a64d13e58e9b4c4fbd8a42eaba4d","kind":"field_change","created_at":"2026-09-11T05:46:38.280755Z","actor":"dawn","issue_id":"klbr-1mz","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-b863e84f96e867e566f6e579734fc114","kind":"field_change","created_at":"2026-09-11T05:46:40.563313Z","actor":"dawn","issue_id":"klbr-kla","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-7c0d7a98fb95a8b32c6988429e9ee4b2","kind":"field_change","created_at":"2026-09-11T05:46:41.167779Z","actor":"dawn","issue_id":"klbr-c3e","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"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"}}442 unmodified linesExpand all4434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914923 unmodified lines49649749849950050150250350450550650750850951051151251351451 unmodified lines566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603528529530531532533534604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633422 unmodified linesExpand all442 unmodified linesExpand all Ok(()) }}
fn ensure_configured_matches_durable( configured: &klbr_runtime::session::Placement, durable: &klbr_runtime::target::SessionPlacement,) -> anyhow::Result<()> { let configured_target = match &configured.target { KernelTarget::Remote(remote) => Some(remote.id.to_string()), KernelTarget::Local => None, }; let durable_target = match &durable.kernel_target { KernelTarget::Remote(remote) => Some(remote.id.to_string()), KernelTarget::Local => None, }; anyhow::ensure!( configured_target == durable_target, "durable placement names target {:?}; the configured placement names {:?}", durable_target, configured_target ); Ok(())}
use klbr_runtime::target::KernelTarget;
/// Placement source for the session descriptor. Resolved once per session/// open; None means the local default. Durable remote placement must come/// from here - a restarted remote session must never silently fall back to/// local execution.pub trait PlacementSource: Send + Sync { fn placement( &self, session: &SessionId, ) -> futures::future::BoxFuture<'_, anyhow::Result<Option<klbr_runtime::session::Placement>>>;}
#[derive(Clone)]pub struct SessionRegistry { pub store: Store, pub worker_config: WorkerConfig, pub sessions: Arc<RwLock<HashMap<SessionId, Arc<SessionEntry>>>>, open_locks: Arc<Mutex<HashMap<SessionId, Arc<Mutex<()>>>>>, service_tasks: Arc<StdMutex<HashMap<ServiceTaskGroup, Vec<tokio::task::JoinHandle<()>>>>>, placement: Option<Arc<dyn PlacementSource>>, pub recovery_policy: crate::observation::RecoveryObservationPolicy,}
impl SessionRegistry {3 unmodified lines worker_config, sessions: Arc::new(RwLock::new(HashMap::new())), open_locks: Arc::new(Mutex::new(HashMap::new())), service_tasks: Arc::new(StdMutex::new(HashMap::new())), placement: None, recovery_policy: crate::observation::RecoveryObservationPolicy::default(), } }
/// Register the host-owned placement resolver used at session open. pub fn with_placement(mut self, placement: Arc<dyn PlacementSource>) -> Self { self.placement = Some(placement); self }
pub fn with_recovery_policy( mut self, policy: crate::observation::RecoveryObservationPolicy, ) -> Self {51 unmodified lines } }
// We are the single owner responsible for starting the SessionRuntime // Durable placement first: a persisted REMOTE placement is still the // contract. A restart without a configured backend for it is a typed // refusal, never a silent local fallback. let durable = self .store .session_placement(id) .await .context("failed to read persisted session placement")?; let runtime = match (&self.placement, durable.clone()) { (Some(source), _) => { let placement = source .placement(id) .await .context("placement resolution failed")?; match placement { Some(placement) => { if let Some(durable) = &durable { ensure_configured_matches_durable(&placement, durable) .context("configured placement contradicts the persisted placement")?; } SessionRuntime::start_with_placement( self.store.clone(), id.clone(), self.worker_config.clone(), initial_release.clone(), placement, ) .await } None => match durable.map(|d| d.kernel_target) { Some(KernelTarget::Remote(_)) => bail!( "session {id} is durably placed on a remote target but this service has no backend configured for it; refusing silent local fallback" ), _ => { let runtime = SessionRuntime::start( self.store.clone(), id.clone(), self.worker_config.clone(), initial_release.clone(), ) .await SessionRuntime::start( self.store.clone(), id.clone(), self.worker_config.clone(), initial_release.clone(), ) .await } }, } } (None, Some(durable)) if matches!(durable.kernel_target, KernelTarget::Remote(_)) => { bail!( "session {id} is durably placed on a remote target but no placement source is configured; refusing silent local fallback" ) } _ => { SessionRuntime::start( self.store.clone(), id.clone(), self.worker_config.clone(), initial_release.clone(), ) .await } } .context("failed to start session runtime")?;
let now_ms = chrono::Utc::now().timestamp_millis(); let initial_state = match self.store.session_state(id).await? {422 unmodified linesExpand all123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727//! M1 Gate U05: Entire workbench executes on selected remote target.//!//! Required observations://! 1. Target process identity, platform ("Darwin"), and cwd are verified.//! 2. A variable from cell one is used by cell two across ordinary turns in the same persistent namespace.//! 3. The local-only marker or path cannot satisfy the target assertion (verified via exclusive remote marker).//! 4. No peer or model was created merely to execute.//! 5. Both calls execute through the daemon's real turn path (normal provider loop via WebSocket SendInput + Ack).//!//! Verification layer: real loopback SSH transport over localhost (Darwin arm64 login shell zsh).
use std::{path::PathBuf, sync::Arc, time::Duration};
use anyhow::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 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()}
fn fixtures_root() -> PathBuf { repo_root().join("tests/fixtures/remote-loopback")}
struct LoopbackPlacementSource { placement: Placement,}
impl PlacementSource for LoopbackPlacementSource { fn placement( &self, _session: &SessionId, ) -> futures::future::BoxFuture<'_, anyhow::Result<Option<Placement>>> { let p = self.placement.clone(); Box::pin(async move { Ok(Some(p)) }) }}
struct TestDaemon { pub addr: std::net::SocketAddr, pub store: Store, pub registry: Arc<SessionRegistry>, #[allow(dead_code)] pub v2_event_tx: broadcast::Sender<V2ServerMsg>, pub shutdown_tx: watch::Sender<bool>, pub handle: tokio::task::JoinHandle<()>,}
impl TestDaemon { pub 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_test_daemon_with_placement( temp: &tempfile::TempDir, llm: Option<LlmClient>, initial_release: Option<Release>, placement_source: Arc<dyn PlacementSource>,) -> Result<TestDaemon> { let listener = TcpListener::bind("127.0.0.1:0").await?; let addr = listener.local_addr()?;
let db_path = temp.path().join("daemon_test.sqlite"); let store = Store::open(&db_path)?; let worker_config = WorkerConfig::source_tree(&repo_root()); let reg = SessionRegistry::new(store.clone(), worker_config).with_placement(placement_source); let registry = Arc::new(reg);
let mem_path = temp.path().join("mem.sqlite"); let mem_store = MemoryStore::open(mem_path.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 reg_clone = registry.clone(); let v2_tx_clone = v2_event_tx.clone();
let initial_rel = initial_release.or_else(|| { Release::publish( &fixtures_root().join("behavior"), &temp.path().join("releases"), ) .ok() });
let handle = tokio::spawn(async move { let _ = serve_listener( listener, interrupt_tx, output_tx, v2_tx_clone, snapshot, mem_store, 50, "test_soul".to_string(), reg_clone, llm, initial_rel, true, // v2 mode shutdown_rx, ) .await; });
Ok(TestDaemon { addr, store, registry, v2_event_tx, shutdown_tx, handle, })}
async fn spawn_u05_llm_server(remote_ws_path: String) -> 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 (mut socket, _) = match listener.accept().await { Ok(conn) => conn, Err(_) => break, };
let ws_path = remote_ws_path.clone(); tokio::spawn(async move { let mut buf = vec![0u8; 32768]; let n = socket.read(&mut buf).await.unwrap_or(0); let req_str = String::from_utf8_lossy(&buf[..n]).to_string(); let body_str = if let Some(idx) = req_str.find("\r\n\r\n") { &req_str[idx + 4..] } else if let Some(idx) = req_str.find("\n\n") { &req_str[idx + 2..] } else { &req_str };
let json_val: serde_json::Value = serde_json::from_str(body_str).unwrap_or_default(); let messages = json_val.get("messages").and_then(|m| m.as_array());
let last_role = messages .and_then(|msgs| msgs.last()) .and_then(|m| m.get("role")) .and_then(|r| r.as_str());
let body = if last_role == Some("tool") { let chunk = serde_json::json!({ "choices": [{"delta": {"role": "assistant", "content": "Tool call execution complete."}}] }); format!( "data: {}\n\ndata: [DONE]\n\n", serde_json::to_string(&chunk).unwrap() ) } else { let last_user = messages .and_then(|msgs| { msgs.iter() .rev() .find(|m| m.get("role").and_then(|r| r.as_str()) == Some("user")) }) .and_then(|m| m.get("content").and_then(|c| c.as_str())) .unwrap_or("");
if last_user.contains("Cell 2") || last_user.contains("cell-2") { // Cell 2: uses variable from Cell 1 let code = r#"assert "shared_round_counter_u05" in globals(), "shared_round_counter_u05 must exist from Cell 1"shared_round_counter_u05 += 1print(f"U05_CELL2_SUCCESS:{shared_round_counter_u05}")"#; let args = serde_json::json!({ "code": code }); let chunk = serde_json::json!({ "choices": [{ "delta": { "tool_calls": [{ "index": 0, "id": "call_u05_cell_2", "type": "function", "function": { "name": "python", "arguments": serde_json::to_string(&args).unwrap(), } }] } }] }); format!( "data: {}\n\ndata: [DONE]\n\n", serde_json::to_string(&chunk).unwrap() ) } else { // Cell 1: defines variable and asserts target process identity, platform, cwd, and remote marker let code = format!( r#"import os, sys, jsonglobal shared_round_counter_u05shared_round_counter_u05 = 42
# 1. Verify remote target cwd matches configured remote workspaceassert os.getcwd() == "{ws_path}", f"Expected remote cwd '{ws_path}', got '{{os.getcwd()}}'"
# 2. Verify remote target platform is Darwinassert sys.platform == "darwin", f"Expected darwin platform, got '{{sys.platform}}'"
# 3. Verify target process identitypid = os.getpid()assert pid > 0, "Target process PID must be positive"
# 4. Exclusive remote marker verification (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 '{{marker_path}}' not found"with open(marker_path) as f: marker_data = json.load(f)assert marker_data.get("marker") == "REMOTE_LOOPBACK_TARGET_EXCLUSIVE"
print(f"U05_CELL1_SUCCESS: pid={{pid}}, platform={{sys.platform}}, cwd={{os.getcwd()}}")"# ); let args = serde_json::json!({ "code": code }); let chunk = serde_json::json!({ "choices": [{ "delta": { "tool_calls": [{ "index": 0, "id": "call_u05_cell_1", "type": "function", "function": { "name": "python", "arguments": serde_json::to_string(&args).unwrap(), } }] } }] }); format!( "data: {}\n\ndata: [DONE]\n\n", serde_json::to_string(&chunk).unwrap() ) } };
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)}
use klbr_runtime::{ protocol::{ read_frame, write_frame, Capability, HelloBinding, Message, RuntimeIdentity, WorkerIdentity, }, 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, .. } = enriched.message { enriched.message = Message::Hello { token, role, revision, slots, 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 { loop { match read_frame(&mut b_read).await { Ok(mut env) => { env.target = None; if write_frame(&mut c_write, &env).await.is_err() { break; } } Err(_) => break, } } };
let forward_to_host = async { loop { match read_frame(&mut c_read).await { Ok(mut env) => { env.target = Some(stamp_clone.clone()); if write_frame(&mut b_write, &env).await.is_err() { break; } } 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_whole_workbench_executes_on_selected_remote_target() -> Result<()> { // 0. Verify genuine SSH to localhost is operational let user = std::env::var("USER").unwrap_or_else(|_| "dawn".to_string()); let ssh_config = SshConfig::new("localhost").with_user(&user); let mut probe = SshTransport::spawn(&ssh_config, "echo SSH_U05_PROBE_OK")?; let mut probe_out = Vec::new(); probe.stdout.read_to_end(&mut probe_out).await?; let probe_status = probe.wait().await?; assert!( probe_status.success(), "Loopback SSH to localhost must succeed" ); assert!( String::from_utf8_lossy(&probe_out).contains("SSH_U05_PROBE_OK"), "SSH probe must return echo receipt" );
let temp = tempfile::tempdir()?; 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 ws_path_str = ws_path.to_string_lossy().to_string();
// 1. Build RemoteTarget descriptor pointing to localhost loopback with Darwin requirement let target = RemoteTarget { id: TargetId::parse("remote-workbench-loopback")?, ssh_alias: "localhost".into(), account: Some(user.clone()), workspace: RemoteWorkspace { path: ws_path_str.clone(), 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 placement_source = Arc::new(LoopbackPlacementSource { placement });
// 2. Spawn mock LLM server handling both cell turns let mock_addr = spawn_u05_llm_server(ws_path_str.clone()).await?; let llm = LlmClient::new(ModelsConfig { llm: ModelConfig { url: format!("http://{}", mock_addr), model: "m1-gate-mock".into(), ..Default::default() }, ..Default::default() });
// 3. Boot real daemon with remote placement source let daemon = start_test_daemon_with_placement(&temp, Some(llm), None, placement_source).await?; let ws_url = format!("ws://{}", daemon.addr); let (ws_stream, _) = connect_async(&ws_url).await?; let (mut write, mut read) = ws_stream.split();
let session_id = "u05_remote_session"; let s_id = parse_session_id(session_id);
let live_messages: Arc<tokio::sync::Mutex<Vec<V2ServerMsg>>> = Arc::new(tokio::sync::Mutex::new(Vec::new())); let live_msgs_task = live_messages.clone();
let read_handle = tokio::spawn(async move { while let Some(msg_res) = read.next().await { if let Ok(WsMessage::Text(txt)) = msg_res { if let Ok(server_msg) = serde_json::from_str::<V2ServerMsg>(&txt) { live_msgs_task.lock().await.push(server_msg); } } } });
let entry_probe = daemon .registry .get_or_open( &s_id, &Release::publish( &fixtures_root().join("behavior"), &temp.path().join("releases"), )?, ) .await; if let Err(e) = &entry_probe { eprintln!("DAEMON GET_OR_OPEN ERROR: {e:#}"); } assert!(entry_probe.is_ok(), "get_or_open must succeed");
// Subscribe to session let sub = V2ClientMsg::Subscribe { session_id: session_id.to_string(), after_seq: Some(0), }; write .send(WsMessage::Text(serde_json::to_string(&sub)?)) .await?;
// Wait for initial sync let deadline = tokio::time::Instant::now() + Duration::from_secs(5); let mut initial_sync_completed = false; while tokio::time::Instant::now() < deadline { let msgs = live_messages.lock().await; if msgs .iter() .any(|m| matches!(m, V2ServerMsg::SyncCompleted { .. })) { initial_sync_completed = true; break; } drop(msgs); tokio::time::sleep(Duration::from_millis(50)).await; } assert!(initial_sync_completed, "Initial sync must complete");
// 4. TURN 1: Send normal operator stimulus through real admission interface let send1 = V2ClientMsg::SendInput { session_id: session_id.to_string(), source: "operator".to_string(), text: "Execute Cell 1 on remote workbench: verify target identity, platform, cwd" .to_string(), client_request_id: Some("req-u05-1".to_string()), }; write .send(WsMessage::Text(serde_json::to_string(&send1)?)) .await?;
// Wait for Ack 1 let deadline = tokio::time::Instant::now() + Duration::from_secs(5); let mut ack1 = false; while tokio::time::Instant::now() < deadline { let msgs = live_messages.lock().await; if msgs.iter().any(|m| matches!(m, V2ServerMsg::Ack { client_request_id: Some(ref r), .. } if r == "req-u05-1")) { ack1 = true; break; } drop(msgs); tokio::time::sleep(Duration::from_millis(50)).await; } assert!(ack1, "Expected Ack for SendInput 1");
// Wait for Turn 1 execution completed event in store history let deadline = tokio::time::Instant::now() + Duration::from_secs(15); let mut turn1_output = String::new(); while tokio::time::Instant::now() < deadline { let history = daemon.store.history(&s_id, 0, 50).await.unwrap_or_default(); for ev in &history { if ev.kind == "execution.completed" { if let Some(outputs) = ev .payload .get("outcome") .and_then(|o| o.get("output")) .and_then(|o| o.as_array()) { for out in outputs { if let Some(txt) = out.get("text").and_then(|t| t.as_str()) { if txt.contains("U05_CELL1_SUCCESS") { turn1_output = txt.to_string(); break; } } } } } } if !turn1_output.is_empty() { break; } tokio::time::sleep(Duration::from_millis(50)).await; } if turn1_output.is_empty() { let history = daemon.store.history(&s_id, 0, 50).await.unwrap_or_default(); for ev in &history { eprintln!("U05 STORE EV: kind={} payload={:?}", ev.kind, ev.payload); } } assert!( !turn1_output.is_empty(), "Turn 1 must execute cell on remote target and produce verified output" ); assert!( turn1_output.contains("platform=darwin"), "Turn 1 output must verify Darwin platform: {turn1_output}" );
// 5. TURN 2: Send second turn using the variable defined in cell 1 across persistent namespace let send2 = V2ClientMsg::SendInput { session_id: session_id.to_string(), source: "operator".to_string(), text: "Execute Cell 2 on remote workbench using variable from cell 1".to_string(), client_request_id: Some("req-u05-2".to_string()), }; write .send(WsMessage::Text(serde_json::to_string(&send2)?)) .await?;
// Wait for Ack 2 let deadline = tokio::time::Instant::now() + Duration::from_secs(5); let mut ack2 = false; while tokio::time::Instant::now() < deadline { let msgs = live_messages.lock().await; if msgs.iter().any(|m| matches!(m, V2ServerMsg::Ack { client_request_id: Some(ref r), .. } if r == "req-u05-2")) { ack2 = true; break; } drop(msgs); tokio::time::sleep(Duration::from_millis(50)).await; } assert!(ack2, "Expected Ack for SendInput 2");
// Wait for Turn 2 execution completed event in store history let deadline = tokio::time::Instant::now() + Duration::from_secs(15); let mut turn2_output = String::new(); while tokio::time::Instant::now() < deadline { let history = daemon.store.history(&s_id, 0, 50).await.unwrap_or_default(); for ev in &history { if ev.kind == "execution.completed" { if let Some(outputs) = ev .payload .get("outcome") .and_then(|o| o.get("output")) .and_then(|o| o.as_array()) { for out in outputs { if let Some(txt) = out.get("text").and_then(|t| t.as_str()) { if txt.contains("U05_CELL2_SUCCESS:43") { turn2_output = txt.to_string(); break; } } } } } } if !turn2_output.is_empty() { break; } tokio::time::sleep(Duration::from_millis(50)).await; } assert!( !turn2_output.is_empty(), "Turn 2 must successfully read and increment variable from Cell 1 across turns (U05_CELL2_SUCCESS:43)" );
// 6. Assertions on target binding and no-peer execution // Check that no auxiliary peer or model was created merely to execute let all_sessions = daemon.store.all_sessions().await?; assert_eq!( all_sessions.len(), 1, "No auxiliary peer session may be created merely to execute" );
// Clean shutdown drop(write); daemon.stop_and_join().await?; let _ = read_handle.await;
Ok(())}12345678910111213141516171819202122232425262728293031-- Migration 007: durable session workbench placement, keyed by namespace-- generation. Every session (including pre-existing ones) gets an explicit-- local placement row; absence is never a guess.BEGIN IMMEDIATE;
CREATE TABLE session_placements ( session_id TEXT NOT NULL REFERENCES sessions(id), namespace_generation TEXT NOT NULL, placement TEXT NOT NULL CHECK(json_valid(placement)), created_ms INTEGER NOT NULL CHECK(created_ms >= 0), PRIMARY KEY (session_id, namespace_generation)) STRICT;
CREATE INDEX idx_session_placements_session ON session_placements(session_id, created_ms);
-- Existing sessions upgrade to an explicit Local binding: unambiguously-- local, not NULL-means-guess. The generation marker identifies the-- pre-M1 binding; a relocate always mints a fresh generation instead.INSERT INTO session_placements (session_id, namespace_generation, placement, created_ms)SELECT s.id, 'm0-legacy', '{"kernel_target":{"placement":"local"},"namespace_id":"main","namespace_generation":"m0-legacy"}', CAST(strftime('%s','now') AS INTEGER) * 1000FROM sessions sWHERE NOT EXISTS ( SELECT 1 FROM session_placements p WHERE p.session_id = s.id);
PRAGMA user_version = 7;COMMIT;12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455-- Migration 008: extend the events kind vocabulary with 'kernel.placement'-- so a placement change is a committed, immutable transition event, not an-- in-place session mutation. SQLite cannot ALTER a CHECK constraint, so the-- events table is rebuilt with identical columns/indexes/triggers plus the-- new kind; all rows and their sequence numbers are preserved.---- PRAGMA ORDERING MATTERS: foreign_keys may only be toggled OUTSIDE any-- transaction (the pragma is a no-op inside one), and this rebuild drops a-- table that executions, effects, inbox and the migration tables reference.-- The toggle therefore brackets the whole rebuild, and the rebuild must-- pass an orphan guard before it commits: a rebuild that silently orphans-- a reference is the one failure mode worse than refusing to upgrade.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', 'local.sent','delivery.held','kernel.reset','session.waited', 'session.woke','hooks.activated','kernel.placement' )), 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;
DROP TABLE events;ALTER TABLE events_v2 RENAME TO events;
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;
-- FATAL orphan guard: pragma_foreign_key_check must be EMPTY. The insert-- below only fires when there are rows to insert, and the CHECK rejects-- every row, so any orphan aborts the migration before COMMIT.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 = 8;COMMIT;PRAGMA foreign_keys = ON;5 unmodified lines6789101112131415165 unmodified lines
pub mod behavior;pub mod hooks;pub mod protocol;pub mod remote;pub mod session;pub mod store;pub mod target;pub mod worker;
pub mod work;