//! Observation replay contracts:
//! - a real failing cell produces resolvable durable evidence;
//! - redelivery does not multiply one occurrence.
use std::{path::PathBuf, time::Duration};
use anyhow::Result;
use klbr_daemon::{
daemon::observe_recovered_error,
observation::{replay_session_observations, RecoveryObservationPolicy},
};
use klbr_runtime::{
behavior::Release,
protocol::{CellStatus, Outcome, SessionId},
session::SessionRuntime,
store::Store,
worker::WorkerConfig,
};
use serde_json::json;
fn root() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.parent()
.expect("workspace root")
.to_path_buf()
}
/// 1. A real failing cell followed by successful recovery produces an occurrence WITH resolvable evidence.
#[tokio::test]
async fn test_real_failing_cell_recovery_produces_occurrence_with_resolvable_evidence() -> Result<()>
{
let temp = tempfile::tempdir()?;
let store = Store::open(temp.path().join("runtime.sqlite"))?;
let id = SessionId::fresh();
let release = Release::publish(&root().join("behavior"), &temp.path().join("releases"))?;
let runtime = SessionRuntime::start(
store.clone(),
id.clone(),
WorkerConfig::source_tree(&root()),
release.clone(),
)
.await?;
// Cell 1: deliberate failure in Python workbench
let rep1 = runtime
.execute(
"raise ValueError('calibrated workbench fault: buffer overflow')
"
.into(),
Duration::from_secs(10),
)
.await?;
assert!(matches!(
rep1.outcome,
Outcome::Cell {
status: CellStatus::Error,
..
}
));
// Cell 2: successful recovery
let rep2 = runtime
.execute(
"recovered_result = 42
recovered_result
"
.into(),
Duration::from_secs(10),
)
.await?;
assert!(matches!(
rep2.outcome,
Outcome::Cell {
status: CellStatus::Ok,
..
}
));
// Observation replay processes committed events from Cell 1
let policy = RecoveryObservationPolicy {
min_recurrence: 1, // trigger work on 1st recurrence for direct verification
..Default::default()
};
let created = replay_session_observations(
&store,
&id,
&runtime.workspace().display().to_string(),
&release.id,
&policy,
)
.await?;
runtime.shutdown().await;
assert_eq!(created.len(), 1, "Work item created for recovered failure");
let work_id = &created[0];
let work = store.get_work_item(work_id).await?.expect("work exists");
// Must carry resolvable evidence ref back to the committed event
assert!(
!work.provenance.source_refs.is_empty(),
"Work item must contain evidence refs"
);
let event_ref = &work.provenance.source_refs[0];
assert!(
event_ref.starts_with("event:"),
"Evidence ref must point to committed event: {event_ref}"
);
// Resolve evidence through the store to confirm the error trace is retrievable
let seq: i64 = event_ref.strip_prefix("event:").unwrap().parse()?;
let event = store.get_event(&id, seq).await?.expect("event exists");
let payload_str = serde_json::to_string(&event.payload)?;
assert!(
payload_str.contains("ValueError")
&& payload_str.contains("calibrated workbench fault: buffer overflow"),
"Resolved event payload must preserve original error signature and traceback"
);
Ok(())
}
/// 2. Repeated source inspection with redelivery at a later wall time stays ONE occurrence.
#[tokio::test]
async fn test_repeated_source_inspection_with_redelivery_at_later_wall_time_stays_one_occurrence(
) -> Result<()> {
let temp = tempfile::tempdir()?;
let store = Store::open(temp.path().join("dedup.sqlite"))?;
let id = SessionId::fresh();
store.ensure_session(&id).await?;
let event_id = store
.append(
&id,
"scratchpad",
json!({
"type": "tool_result",
"status": "error",
"error": "TypeError: duplicate delivery probe",
"call_id": "call-1",
}),
)
.await?;
let policy = RecoveryObservationPolicy {
min_recurrence: 2,
..Default::default()
};
// First inspection at T = 10_000
let res1 = observe_recovered_error(
&store,
&id,
"tool.search",
"rev-1",
"TypeError: duplicate delivery probe",
"/fixture/ws",
10_000,
event_id,
&policy,
)
.await?;
assert_eq!(res1, None, "1st delivery groups only");
// Redelivery at a much later wall time (T = 500_000) for the SAME source event
let res2 = observe_recovered_error(
&store,
&id,
"tool.search",
"rev-1",
"TypeError: duplicate delivery probe",
"/fixture/ws",
500_000,
event_id,
&policy,
)
.await?;
assert_eq!(
res2, None,
"Redelivery of same committed event must be deduplicated, NOT create a second occurrence"
);
// Verify case occurrence count in store is still 1, NOT 2
use klbr_core::action_identity::compute_action_identity;
let action_id = compute_action_identity(
"tool.search",
"rev-1",
"TypeError: duplicate delivery probe",
"/fixture/ws",
);
let case = store
.get_observer_case(&action_id)
.await?
.expect("case exists");
assert_eq!(
case.count, 1,
"Occurrence count must stay 1 across redeliveries"
);
Ok(())
}