//! 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 sys
from klbr import workspace
ws = 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, local
holding = 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 local
r2 = 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 behavior
broken = behavior.load(r"{broken_path}")
caught = None
try:
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 behavior
preloaded_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 behavior
caught_tamper = None
try:
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 orientation
try:
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(())
}