ive harnessed the harness
Something went wrong. Try again.
32 kB · 899 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900//! Actual Rust -> Python integration tests. No provider or Discord credentials.//! These were authored but NOT RUN in the artifact-building environment.use anyhow::{anyhow, Result};use klbr_runtime::{ behavior::Release, protocol::*, session::{ExecutionReport, Placement, SessionRuntime}, store::Store, target::{KernelTarget, NamespaceId}, worker::WorkerConfig,};use serde_json::json;use std::{fs, path::PathBuf, sync::Arc, time::Duration};use tokio::time::{sleep, timeout};
fn root() -> PathBuf { PathBuf::from(env!("CARGO_MANIFEST_DIR")) .parent() .unwrap() .into()}fn output(report: &ExecutionReport) -> String { match &report.outcome { Outcome::Cell { output, .. } => output.iter().map(|o| o.text.as_str()).collect(), other => panic!("expected cell, got {other:?}"), }}fn ok(report: &ExecutionReport) { assert!( matches!( &report.outcome, Outcome::Cell { status: CellStatus::Ok, .. } ), "{report:?}" );}struct Rig { temp: tempfile::TempDir, session: Arc<SessionRuntime>,}impl Rig { async fn new() -> Result<Self> { let temp = tempfile::tempdir()?; let initial = Release::publish(&root().join("behavior"), &temp.path().join("releases"))?; let store = Store::open(temp.path().join("runtime.sqlite"))?; let session = SessionRuntime::start( store, SessionId::fresh(), WorkerConfig::source_tree(&root()), initial, ) .await?; Ok(Self { temp, session }) } fn changed(&self, name: &str, body: &str) -> Result<Release> { let source = self.temp.path().join(name); fs::create_dir_all(source.join("klbr_hooks"))?; fs::copy( root().join("behavior/manifest.json"), source.join("manifest.json"), )?; fs::copy( root().join("behavior/klbr_hooks/__init__.py"), source.join("klbr_hooks/__init__.py"), )?; let mut defaults = fs::read_to_string(root().join("behavior/klbr_hooks/defaults.py"))?; defaults.push_str(body); fs::write(source.join("klbr_hooks/defaults.py"), defaults)?; Release::publish(&source, &self.temp.path().join("releases")) } async fn run(&self, code: &str) -> Result<ExecutionReport> { self.session .execute(code.into(), Duration::from_secs(10)) .await }}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn persistent_cells_explicit_speech_and_terminal_wait() -> Result<()> { let rig = Rig::new().await?; ok(&rig.run("answer = 42\nprint('not speech')\nanswer").await?); let report = rig .run("from klbr import local\nr = await local.send(str(answer))\nr.status") .await?; ok(&report); assert!(output(&report).contains("confirmed")); let history = rig.session.store.history(&rig.session.id, 0, 100).await?; assert_eq!(history.iter().filter(|e| e.kind == "local.sent").count(), 1); let waiting=rig.run("from klbr import runtime\nawait runtime.wait(reason='idle')\nawait local.send('incorrect')").await?; assert!(matches!( waiting.outcome, Outcome::Cell { status: CellStatus::Yielded, .. } )); assert_eq!( rig.session .store .history(&rig.session.id, 0, 100) .await? .iter() .filter(|e| e.kind == "local.sent") .count(), 1 ); rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn bad_guard_holds_instead_of_approving_and_hook_change_keeps_globals() -> Result<()> { let rig = Rig::new().await?; ok(&rig .run("kept = object()\noriginal_identity = id(kept)") .await?); let broken = rig.changed( "bad-return", "\nasync def delivery(req, ctx):\n return True\n", )?; rig.session .activate_hooks(rig.session.hook_epoch().await, broken) .await?; let report=rig.run("from klbr import local\nassert id(kept) == original_identity\nr = await local.send('must be held')\nr.status").await?; ok(&report); assert!(output(&report).contains("held")); assert_eq!( rig.session .store .history(&rig.session.id, 0, 100) .await? .iter() .filter(|e| e.kind == "local.sent") .count(), 0 ); rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn invalid_candidate_does_not_activate_or_destroy_workbench() -> Result<()> { let rig = Rig::new().await?; ok(&rig.run("kept = 42").await?); let epoch = rig.session.hook_epoch().await; let broken = rig.changed("syntax-error", "\nthis is not valid python !\n")?; assert!(rig.session.activate_hooks(epoch, broken).await.is_err()); assert_eq!(rig.session.hook_epoch().await, epoch); assert!(output(&rig.run("kept").await?).contains("42")); rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn host_deadline_replaces_stuck_workbench_without_replaying_code() -> Result<()> { let rig = Rig::new().await?; let report = rig .session .execute("while True: pass".into(), Duration::from_millis(200)) .await?; assert!(matches!(report.outcome, Outcome::Failed { .. })); let next = rig.run("40 + 2").await?; ok(&next); assert_ne!(report.generation, next.generation); assert!(output(&next).contains("42")); rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn workspace_commands_are_host_owned_bounded_and_truthful() -> Result<()> { let rig = Rig::new().await?; let report = rig .run( r#"import sysfrom klbr import workspacews = await workspace.current()small = await ws.run((sys.executable, "-c", "print('abcdefghijk')"), max_output_bytes=5)slow = await ws.run((sys.executable, "-c", "import time; time.sleep(5)"), timeout=0.05)(small.stdout, small.stdout_truncated, slow.completed, slow.timed_out)"#, ) .await?; ok(&report); let text = output(&report); assert!(text.contains("b'abcde'"), "{text}"); assert!(text.contains("True, False, True"), "{text}"); rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn discarded_workspace_command_cannot_leave_a_process_group_behind() -> Result<()> { let rig = Rig::new().await?; let marker = rig.temp.path().join("orphan-wrote"); let child = format!( "import subprocess,sys,time; subprocess.Popen([sys.executable, '-c', {:?}]); time.sleep(30)", format!("import time,pathlib; time.sleep(1); pathlib.Path({:?}).write_text('orphan')", marker) ); let code = format!( "import sys\nfrom klbr import workspace\nws = await workspace.current()\nawait ws.run((sys.executable, '-c', {:?}), timeout=30)", child ); let report = rig .session .execute(code, Duration::from_millis(200)) .await?; assert!(matches!(report.outcome, Outcome::Failed { .. })); sleep(Duration::from_millis(1300)).await; assert!( !marker.exists(), "workspace command descendant survived cancellation" ); rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn blocked_hook_does_not_block_admission_and_fails_closed() -> Result<()> { let rig = Rig::new().await?; let marker = rig.temp.path().join("hook-entered"); let body=format!("\nasync def delivery(req, ctx):\n from pathlib import Path\n Path({:?}).touch()\n while True: pass\n",marker.to_str().unwrap()); let hanging = rig.changed("hang", &body)?; rig.session .activate_hooks(rig.session.hook_epoch().await, hanging) .await?; let session = rig.session.clone(); let execution = tokio::spawn(async move { session .execute( "from klbr import local\nr = await local.send('held after timeout')\nr.status" .into(), Duration::from_secs(10), ) .await }); timeout(Duration::from_secs(5), async { while !marker.exists() { sleep(Duration::from_millis(5)).await; } }) .await?; timeout( Duration::from_secs(1), rig.session.store.admit( &rig.session.id, "during-hook".into(), json!({"text":"new input"}), ), ) .await??; let report = timeout(Duration::from_secs(8), execution).await???; ok(&report); assert!(output(&report).contains("held")); assert_eq!( rig.session.store.pending(&rig.session.id, 10).await?.len(), 1 ); rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn in_flight_cell_pins_old_guard_and_next_cell_uses_new_guard() -> Result<()> { let rig = Rig::new().await?; let boundary = rig.temp.path().join("continue-cell"); let code=format!("import asyncio\nfrom pathlib import Path\nfrom klbr import local\nwhile not Path({:?}).exists():\n await asyncio.sleep(0.01)\nr = await local.send('old pinned guard')\nr.status",boundary.to_str().unwrap()); let session = rig.session.clone(); let task = tokio::spawn(async move { session.execute(code, Duration::from_secs(20)).await }); timeout(Duration::from_secs(5), async { loop { if rig .session .store .history(&rig.session.id, 0, 100) .await? .iter() .any(|e| e.kind == "execution.started") { return Ok::<_, anyhow::Error>(()); } sleep(Duration::from_millis(10)).await; } }) .await??; let revision = rig.changed( "hold", "\nasync def delivery(req, ctx):\n return DeliveryDecision.hold('new guard')\n", )?; rig.session .activate_hooks(rig.session.hook_epoch().await, revision.clone()) .await?; fs::write(boundary, "continue")?; let old = task.await??; ok(&old); assert!(output(&old).contains("confirmed")); assert_ne!(old.hook_revision, revision.id); let next = rig.run("(await local.send('new guard')).status").await?; ok(&next); assert!(output(&next).contains("held")); assert_eq!(next.hook_revision, revision.id); rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn operator_stop_remains_available_while_python_is_cpu_bound() -> Result<()> { let rig = Rig::new().await?; let session = rig.session.clone(); let task = tokio::spawn(async move { session .execute("while True: pass".into(), Duration::from_secs(30)) .await }); timeout(Duration::from_secs(5), async { loop { if rig .session .store .history(&rig.session.id, 0, 100) .await? .iter() .any(|e| e.kind == "execution.started") { break; } sleep(Duration::from_millis(10)).await; } Ok::<_, anyhow::Error>(()) }) .await??; timeout(Duration::from_secs(2), rig.session.interrupt()).await?; let report = timeout(Duration::from_secs(5), task).await???; if !matches!(report.outcome, Outcome::Failed { .. }) { return Err(anyhow!("stopped code should not claim completion")); } rig.session.shutdown().await; Ok(())}
#[tokio::test]async fn shutdown_does_not_silently_respawn_the_kernel() -> Result<()> { let rig = Rig::new().await?; rig.session.shutdown().await; assert!(rig.run("42").await.is_err()); Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn workbench_cell_behavior_activate_updates_next_cell_pins_current_and_handles_rejections() -> Result<()> { let rig = Rig::new().await?; let initial_epoch = rig.session.hook_epoch().await; assert_eq!(initial_epoch, 1);
// 1. Initial cell verifies baseline delivery works and is confirmed let base_rep = rig .run("from klbr import local\nr0 = await local.send('baseline')\nr0.status") .await?; ok(&base_rep); assert!(output(&base_rep).contains("confirmed"));
// 2. Prepare custom release that holds delivery let holding = rig.changed( "holding-behavior", "\nasync def delivery(req, ctx):\n from klbr.hooks import DeliveryDecision\n return DeliveryDecision.hold('cell_activated_hold')\n", )?; let holding_path = holding.path.to_str().unwrap().replace('\\', "/");
// 3. Cell 1 activates holding behavior via host RPC. // PINNING LAW: Local send in Cell 1 STILL uses the epoch 1 policy (allow -> confirmed), // because Cell 1 pinned the active policy at cell start. let cell_1_code = format!( r#"from klbr import behavior, localholding = behavior.load(r"{holding_path}")active = await holding.activate(expected_epoch={initial_epoch})r1 = await local.send('cell 1 in-flight')(active.epoch, r1.status)"# ); let rep_1 = rig.run(&cell_1_code).await?; ok(&rep_1); let out_1 = output(&rep_1); assert!( out_1.contains("2"), "active epoch must advance to 2: {out_1}" ); assert!( out_1.contains("confirmed"), "cell 1 in-flight send must remain pinned to old policy (confirmed): {out_1}" ); assert_eq!(rig.session.hook_epoch().await, 2);
// 4. Cell 2 starts after activation. It pins the NEW policy (holds delivery). let cell_2_code = r#"from klbr import localr2 = await local.send('cell 2 new policy')(r2.status, r2.reason)"#; let rep_2 = rig.run(cell_2_code).await?; ok(&rep_2); let out_2 = output(&rep_2); assert!( out_2.contains("held"), "cell 2 must use newly activated policy (held): {out_2}" ); assert!( out_2.contains("cell_activated_hold"), "cell 2 reason must match new policy: {out_2}" );
// 5. Rejection Test A: Candidate with syntax error / import failure let broken_dir = rig.temp.path().join("broken-candidate"); fs::create_dir_all(broken_dir.join("klbr_hooks"))?; fs::copy( root().join("behavior/manifest.json"), broken_dir.join("manifest.json"), )?; fs::copy( root().join("behavior/klbr_hooks/__init__.py"), broken_dir.join("klbr_hooks/__init__.py"), )?; fs::write( broken_dir.join("klbr_hooks/defaults.py"), "this is not valid python syntax !!!", )?; let broken_rel = Release::publish(&broken_dir, &rig.temp.path().join("releases"))?; let broken_path = broken_rel.path.to_str().unwrap().replace('\\', "/");
let cell_3_code = format!( r#"from klbr import behaviorbroken = behavior.load(r"{broken_path}")caught = Nonetry: await broken.activate(expected_epoch=2)except Exception as e: caught = type(e).__name__caught"# ); let rep_3 = rig.run(&cell_3_code).await?; ok(&rep_3); let out_3 = output(&rep_3); assert!( out_3.contains("HostError") || out_3.contains("ValidationError") || out_3.contains("CASConflictError"), "broken candidate activation must be rejected: {out_3}" ); // In-memory epoch unchanged assert_eq!(rig.session.hook_epoch().await, 2);
// Verify policy in memory is still the epoch 2 holding policy let rep_4 = rig .run("from klbr import local\nr4 = await local.send('post-broken')\nr4.status") .await?; ok(&rep_4); assert!(output(&rep_4).contains("held"));
// 6. Rejection Test B: Tampered source on disk let valid_cand = rig.changed( "tamper-target", "\nasync def delivery(req, ctx):\n from klbr.hooks import DeliveryDecision\n return DeliveryDecision.allow()\n", )?; let tamper_path = valid_cand.path.to_str().unwrap().replace('\\', "/");
// Pre-load the valid release object in cell before disk is modified let preload_code = format!( r#"from klbr import behaviorpreloaded_rel = behavior.load(r"{tamper_path}")preloaded_rel.id"# ); let rep_preload = rig.run(&preload_code).await?; ok(&rep_preload);
// Tamper with the published files on disk let hook_file = valid_cand.path.join("klbr_hooks/defaults.py"); let mut original_code = fs::read_to_string(&hook_file)?; original_code.push_str("\n# tampered bytes after verification\n"); fs::write(&hook_file, original_code)?;
// Calling activate on the preloaded release object verifies disk bytes and raises ValidationError let cell_5_code = r#"from klbr import behaviorcaught_tamper = Nonetry: await preloaded_rel.activate(expected_epoch=2)except Exception as e: caught_tamper = type(e).__name__caught_tamper"#; let rep_5 = rig.run(cell_5_code).await?; ok(&rep_5); let out_5 = output(&rep_5); assert!( out_5.contains("ValidationError"), "tampered release activate() must raise ValidationError: {out_5}" ); // In-memory epoch unchanged assert_eq!(rig.session.hook_epoch().await, 2);
// 7. Database consistency verification: // Verify DB session row has epoch 2 and revision matches holding release let history = rig.session.store.history(&rig.session.id, 0, 100).await?; let activated_events: Vec<_> = history .iter() .filter(|e| e.kind == "hooks.activated") .collect(); // Initial start (epoch 1) + holding activation (epoch 2) = exactly 2 activation events assert_eq!( activated_events.len(), 2, "only 2 successful activations must be committed to event history" );
rig.session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn relocation_is_a_new_generation_and_never_claims_globals_moved() -> Result<()> { let temp = tempfile::tempdir()?; let initial = Release::publish(&root().join("behavior"), &temp.path().join("releases"))?; let store = Store::open(temp.path().join("runtime.sqlite"))?; let session = SessionRuntime::start( store, SessionId::fresh(), WorkerConfig::source_tree(&root()), initial, ) .await?; ok(&session .execute("kept = 42".into(), Duration::from_secs(10)) .await?); let old_generation = session .execute("kept".into(), Duration::from_secs(10)) .await? .generation .clone();
// the SAME placement class is still a deliberate placement change let new_generation = session .relocate(klbr_runtime::session::Placement::local()) .await?; assert_ne!(&new_generation, &old_generation);
// the transition is persisted and explicit: old and new generation both named let history = session.store.history(&session.id, 0, 100).await?; let placement_events: Vec<_> = history .iter() .filter(|e| e.kind == "kernel.placement") .collect(); assert_eq!(placement_events.len(), 1); let event = &placement_events[0].payload; assert_eq!( event["previous_generation"], json!(old_generation.as_str()), "the transition must name the generation it ended" ); assert_eq!(event["generation"], json!(new_generation.as_str()));
// the DURABLE placement record moved to the new namespace generation too let descriptor = session .store .session_placement(&session.id) .await? .expect("durable placement row"); assert_eq!( descriptor.namespace_generation.as_str(), new_generation.as_str(), "the persisted placement must be bound to the new namespace generation" ); assert!( matches!( descriptor.kernel_target, klbr_runtime::target::KernelTarget::Local ), "relocating to a local placement records a local descriptor" ); assert!( event["reason"] .as_str() .unwrap_or("") .contains("did not migrate"), "the transition must say the namespace did not move: {event}" );
// the new namespace does NOT contain the old globals: no silent heap move let report = session .execute("kept".into(), Duration::from_secs(10)) .await?; assert!( !matches!( report.outcome, Outcome::Cell { status: CellStatus::Ok, .. } ), "globals must not have migrated with the placement change: {report:?}" ); session.shutdown().await; Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn placement_change_is_refused_while_a_foreground_cell_runs() -> Result<()> { let temp = tempfile::tempdir()?; let initial = Release::publish(&root().join("behavior"), &temp.path().join("releases"))?; let store = Store::open(temp.path().join("runtime.sqlite"))?; let session = SessionRuntime::start( store, SessionId::fresh(), WorkerConfig::source_tree(&root()), initial, ) .await?; let busy = session.clone(); let cell_session = busy.clone(); let cell = tokio::spawn(async move { cell_session .execute("while True: pass".into(), Duration::from_secs(20)) .await }); timeout(Duration::from_secs(5), async { loop { if busy .store .history(&busy.id, 0, 100) .await .unwrap_or_default() .iter() .any(|e| e.kind == "execution.started") { break; } sleep(Duration::from_millis(10)).await; } Ok::<_, anyhow::Error>(()) }) .await??; let refused = session .relocate(klbr_runtime::session::Placement::local()) .await; assert!( refused.is_err(), "placement change must refuse during a foreground cell" ); session.interrupt().await; let _ = cell.await?; session.shutdown().await; Ok(())}
#[tokio::test]async fn a_remote_placement_names_its_target_and_carries_a_binding() -> Result<()> { // A remote placement cannot exist without a backend and a namespace binding: the fields are one // variant, so the inconsistent placements this test used to construct are unrepresentable, and // the refusal happens while parsing the target's configuration instead. let temp = tempfile::tempdir()?; let target: klbr_runtime::target::RemoteTarget = serde_json::from_value(json!({ "id": "linux-build", "ssh_alias": "linux-build", "workspace": {"path": "/srv/work/tangled"} })) .map_err(|e| anyhow!("{e}"))?; let backend = Arc::new(klbr_runtime::remote::relay::RemoteBackend::new( klbr_runtime::remote::ssh::SshConfig::new("linux-build"), target.clone(), temp.path().join("klbr-relay"), temp.path().join("python3"), temp.path().join("__main__.py"), )); let placement = Placement::remote( target.clone(), backend, NamespaceId::parse("default").map_err(|e| anyhow!("{e}"))?, "/srv/work/tangled/.klbr/behavior".to_string(), ); match placement.kernel_target() { KernelTarget::Remote(named) => assert_eq!(named.id, target.id), KernelTarget::Local => panic!("a remote placement must name a remote target"), } Ok(())}
#[tokio::test]async fn test_orientation_unconfigured_host_returns_typed_error() -> Result<()> { let rig = Rig::new().await?;
let report = rig .run( r#"from klbr import orientationtry: await orientation.adopt( candidate_id="cand-unconfigured", text="some text", target="some target", ) assert False, "must fail when orientation host is unconfigured"except Exception as e: assert "unconfigured_orientation_host" in str(e), f"unexpected error: {e}""#, ) .await?; ok(&report); Ok(())}
#[tokio::test]async fn runtime_inspection_reports_host_pins_and_explicit_budget_scope() -> Result<()> { let rig = Rig::new().await?; let selected = rig .session .store .activation(&rig.session.id) .await? .unwrap(); let report = rig .session .execute_with_purpose( "from klbr import runtime\nimport json\nprint(json.dumps(await runtime.inspect()))" .into(), Duration::from_secs(5), Purpose::Maintenance, ) .await?; ok(&report); let manifest: serde_json::Value = serde_json::from_str(output(&report).trim())?; assert_eq!(manifest["session_id"], rig.session.id.as_str()); assert_eq!(manifest["execution_id"], report.id.as_str()); assert_eq!(manifest["purpose"], "maintenance"); assert_eq!(manifest["pinned_behavior_revision"], report.hook_revision); assert_eq!(manifest["selected_behavior_revision"], selected.revision); assert_eq!(manifest["selected_behavior_epoch"], selected.epoch); assert_eq!( manifest["target"]["workspace"], root().to_string_lossy().as_ref() ); assert_eq!(manifest["target"]["generation"], report.generation.as_str()); assert_eq!( manifest["behavior_source"]["path"], selected.path.to_string_lossy().as_ref() ); assert_eq!(manifest["behavior_source"]["writable"], false); assert_eq!(manifest["cell_budget"]["limit_ms"], 5000); assert!(manifest["cell_budget"]["remaining_ms"].as_u64().unwrap() <= 5000);
// P05: the roots and the applied revision are properties of THIS session and // must be true of it, not of whatever the daemon was started from. The SDK-side // test can only prove the values are passed through; accuracy is proven here // against the real session. assert_eq!( manifest["source_root"]["path"], rig.session.workspace().display().to_string() ); assert_eq!(manifest["source_root"]["exists"], true); assert_eq!( manifest["draft_root"]["path"], rig.session .workspace() .join(".klbr") .join("drafts") .display() .to_string() ); assert_eq!( manifest["draft_root"]["target"], manifest["target"]["target"].clone() ); // Applied is the revision the worker actually runs under, and it must not be // reported as the selection: a selected revision that has not activated is not // in force. assert_eq!(manifest["applied_behavior_revision"], report.hook_revision); assert_eq!( manifest["applied_behavior_revision"], manifest["pinned_behavior_revision"] ); // The draft root is now a reported fact, so it may not still be claimed missing. assert!(manifest["unavailable"].get("draft_root").is_none()); assert!(manifest["unavailable"]["cumulative_model_budget"].is_string()); assert!(manifest.get("environment").is_none()); assert!(manifest.get("api_key").is_none()); rig.session.shutdown().await; Ok(())}
/// The header a transcript runs under is a durable fact, and the host reports it rather than/// making the model hash a context it cannot reproduce byte for byte.#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn inspect_names_the_pinned_head_and_the_change_observed_since_it() -> Result<()> { use klbr_runtime::prompt_header::{ change_between, join_sources, HeaderPin, HeaderRevision, HeaderSnapshot, HeaderSource, PinReason, RevisionActor, HEADER_PIN_EVENT, HEADER_REVISION_EVENT, };
let rig = Rig::new().await?; let pinned_sources = vec![ HeaderSource::new("soul", "you are a participant\n"), HeaderSource::new("contract", "contract v1\n"), ]; let pin = HeaderPin::of(&pinned_sources, PinReason::SessionStart, 1_700_000_000_000); rig.session .store .append(&rig.session.id, HEADER_PIN_EVENT, pin.payload()) .await?;
let current = vec![ HeaderSource::new("soul", "you are a participant\n"), HeaderSource::new("contract", "contract v2\n"), ]; let revision = HeaderRevision::new( &join_sources(¤t), current.clone(), change_between(&HeaderSnapshot::new(pinned_sources.clone()), ¤t), RevisionActor::host("edited outside this transcript"), 1_700_000_000_500, ); rig.session .store .append(&rig.session.id, HEADER_REVISION_EVENT, revision.payload()) .await?;
let report = rig .session .execute_with_purpose( "from klbr import runtime\nimport json\nprint(json.dumps({'inspect': (await runtime.inspect())['instructions'], 'text': (await runtime.instructions())['digest']}))" .into(), Duration::from_secs(5), Purpose::Maintenance, ) .await?; ok(&report); let answer: serde_json::Value = serde_json::from_str(output(&report).trim())?; let manifest = &answer["inspect"];
assert_eq!(manifest["head_digest"], pin.digest); assert_eq!(manifest["head_reason"], "session_start"); assert_eq!(manifest["head_pinned_at_ms"], pin.pinned_at_ms); assert_eq!(manifest["revisions_since_head"], 1); assert_eq!(manifest["known_digest"], revision.digest); // The two host answers agree: the text the transcript can reconstruct is the text the // revision recorded, and neither is a guess about the live file contents. assert_eq!(answer["text"], revision.digest); assert_ne!(manifest["head_digest"], manifest["known_digest"]); Ok(())}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]async fn workspace_diffs_are_per_cell_and_durable_even_after_an_error() -> Result<()> { let rig = Rig::new().await?; let initial = rig.run(&format!( "from pathlib import Path\nfrom klbr.workspace import Workspace\nw = Workspace(Path({:?}))\nf = await w.open_file('index.html')\nawait f.write('wrong title\\nretained body\\n')\nraise ValueError('intentional test failure')", rig.temp.path().to_string_lossy() )).await?; let Outcome::Cell { status: CellStatus::Error, trace: Some(trace), .. } = initial.outcome else { panic!("expected a failed cell with actual edit receipts") }; assert_eq!(trace.changes[0].added, 2); let fixed = rig .run("await f.replace('wrong title', 'right title')") .await?; ok(&fixed); let Outcome::Cell { trace: Some(trace), .. } = fixed.outcome else { panic!("expected a cell trace") }; assert_eq!((trace.changes[0].added, trace.changes[0].removed), (1, 1)); assert!(trace.changes[0].diff.contains("+right title")); let history = rig.session.store.history(&rig.session.id, 0, 100).await?; let completions: Vec<_> = history .iter() .filter(|event| event.kind == "execution.completed") .collect(); assert_eq!(completions.len(), 2); assert_eq!( completions[0] .payload .pointer("/outcome/trace/changes/0/path"), Some(&json!("index.html")) ); assert_eq!( completions[1] .payload .pointer("/outcome/trace/changes/0/removed"), Some(&json!(1)) ); rig.session.shutdown().await; Ok(())}