Keep an unresolved objective open and separate service quiescence from Stop
156 unmodified linesExpand all157158159160161162163164165166167168169156 unmodified linesExpand all{"id":"int-5c67c21d5895676f0f42b6fb1049d33c","kind":"field_change","created_at":"2026-09-07T02:15:05.496401Z","actor":"dawn","issue_id":"klbr-b4c","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-83902ba4c73f1107771f39cf7a1d984d","kind":"field_change","created_at":"2026-09-07T02:16:07.437249Z","actor":"dawn","issue_id":"klbr-2fd","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-88b9ce6690a1797c586e134495d61228","kind":"field_change","created_at":"2026-09-07T02:18:11.455525Z","actor":"dawn","issue_id":"klbr-yhk","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-28793dfc4423680ca57bdf806fe45a67","kind":"field_change","created_at":"2026-09-07T02:49:35.687968Z","actor":"dawn","issue_id":"klbr-a0s","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-5020a927786ae19e59342369680df521","kind":"field_change","created_at":"2026-09-11T04:43:55.940105Z","actor":"dawn","issue_id":"klbr-7su","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-a93783516a54fee0d7d7c76b1b519b4c","kind":"field_change","created_at":"2026-09-11T04:44:12.06691Z","actor":"dawn","issue_id":"klbr-8jn","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-b46d7862e239b890503f021d42f66302","kind":"field_change","created_at":"2026-09-11T04:44:46.977606Z","actor":"dawn","issue_id":"klbr-bhe","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-1b43c96f73f74d0b36b293e1d4ca00f9","kind":"field_change","created_at":"2026-09-11T04:44:48.004271Z","actor":"dawn","issue_id":"klbr-2hy","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"id":"int-689af5ecabc025bd9cda70a92b55ddaa","kind":"field_change","created_at":"2026-09-11T04:45:06.689282Z","actor":"dawn","issue_id":"klbr-dt6","extra":{"field":"status","new_value":"in_progress","old_value":"open"}}{"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"}}788 unmodified linesExpand all78979079179279379379479579679779881 unmodified lines880881882883883884885886887888889884885886887888889227 unmodified linesExpand all111711181119112011251121112211231124112578 unmodified lines12041205120612071212121312141215121612171208120912101211121212131214121512161217121812191220122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128469 unmodified lines13541355135613571340134113421343134413451346134713481349135013511352135313581359136013611362136313641365136613671368136913701371137213731374137513761360137713781379138013811382138313841385136813861387138813891390139113921393139413951396139713981399140014011402140314041405140614071376140814091410141114121413141414151416141714181419142014211422142314241425142636 unmodified lines14631464146514661467146814691470147157 unmodified lines1529153015311532148715331534153515361537153815391540154115421543154415451546154715481549155015 unmodified lines156615671568156915701571157215731574157515131576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159837 unmodified lines1636163716381639164016411642164316441645164616475 unmodified lines16531654165516561577165716581659166016611582166216631664166516661 unmodified line16681669167016711592167216731674167516762 unmodified lines167916801681168216831684168516861687168816891690806 unmodified linesExpand all788 unmodified linesExpand all Some(s) if !s.is_empty() => s, _ => crate::maintenance::install_standing_responsibility(klbr_core::config::DEFAULT_SOUL), };
let op_for_attach = op_id.clone(); tokio::spawn(async move { let runner_task = tokio::spawn(async move { let runner_future = async { loop { if !entry_clone.can_admit().await { break;81 unmodified lines
let initial_epoch = entry_clone.runtime.hook_epoch().await; let initial_policy = entry_clone.runtime.policy().await; let initial_rev = initial_policy.release.id.clone(); let turn_start_event_id = reg_clone .store() .history(&s_id, 0, 1024) .await .ok() .and_then(|evs| evs.last().map(|e| e.id)) .unwrap_or(0); let turn_start_event_id = crate::disposition::session_head_event_id(reg_clone.store(), &s_id).await;
let messages = match &work_item { Some(w) if matches!(w.cause, RunCause::MaintenanceReady { .. }) => { let ctx = crate::maintenance::MaintenanceAttemptContext::new(227 unmodified linesExpand all .await; break; }
if failed_error.is_none() { { let policy = reg_clone.recovery_policy(); let rev = entry_clone.runtime.policy().await.release.id.clone(); let workspace = entry_clone.runtime.workspace().display().to_string();
78 unmodified lines let final_epoch = entry_clone.runtime.hook_epoch().await; let final_policy = entry_clone.runtime.policy().await; let final_rev = final_policy.release.id.clone();
// Query events committed to this session during this turn let events_during_turn = reg_clone .store() .history(&s_id, turn_start_event_id, 1024) .await .unwrap_or_default(); // Query events committed to this session during this turn using proper pagination let events_during_turn = crate::disposition::read_events_after( reg_clone.store(), &s_id, turn_start_event_id, ) .await;
// 1. Explicit typed assessment from HostRequest::MaintenanceAssess let explicit_assessment = events_during_turn.iter().rev().find_map(|ev| { if ev.payload.get("type").and_then(|v| v.as_str()) == Some("maintenance.assessed") && ev.payload.get("work_item_id").and_then(|v| v.as_str()) == Some(w.id.as_str()) { let outcome_str = ev.payload.get("outcome").and_then(|v| v.as_str()).unwrap_or("no-change"); let reason = ev.payload.get("reason").and_then(|v| v.as_str()).unwrap_or_default().to_string(); let missing_refs: Vec<String> = ev.payload.get("missing_refs") .and_then(|v| serde_json::from_value(v.clone()).ok()) .unwrap_or_default(); let revisit = ev.payload.get("revisit_condition").and_then(|v| v.as_str()).unwrap_or("condition specified by actor").to_string(); let candidate_rev = ev.payload.get("candidate_revision").and_then(|v| v.as_str()).map(ToString::to_string); let check_receipts: Vec<String> = ev.payload.get("check_receipts") .and_then(|v| serde_json::from_value(v.clone()).ok()) .unwrap_or_default();
Some(match outcome_str { "needs-evidence" => crate::disposition::MaintenanceDisposition::NeedsEvidence { missing_refs, revisit_condition: revisit, }, "no-change" => crate::disposition::MaintenanceDisposition::NoChange { reason, evidence_refs: w.provenance.source_refs.clone(), }, "candidate-failed" => crate::disposition::MaintenanceDisposition::CandidateFailed { error: reason, candidate_revision: candidate_rev, }, "deferred" => crate::disposition::MaintenanceDisposition::Deferred { reason, trigger_ms: None, }, "activated" => crate::disposition::MaintenanceDisposition::Activated { work_item_id: w.id.clone(), candidate_revision: candidate_rev.unwrap_or_else(|| final_rev.clone()), epoch: final_epoch, check_receipts, }, _ => crate::disposition::MaintenanceDisposition::NoChange { reason, evidence_refs: w.provenance.source_refs.clone(), }, }) // Causal binding: if attempt_id is present, ensure match with current attempt if let Some(att) = ev.payload.get("attempt_id").and_then(|v| v.as_str()) { if att != op_id.as_str() { tracing::warn!(session = %s_id, expected = %op_id, actual = %att, "ignoring assessment bound to different attempt"); return None; } } let outcome_str = ev.payload.get("outcome").and_then(|v| v.as_str()).unwrap_or(""); match klbr_runtime::protocol::MaintenanceAssessment::parse(outcome_str, &ev.payload) { Ok(parsed) => { Some(match parsed { klbr_runtime::protocol::MaintenanceAssessment::NeedsEvidence { missing_refs, revisit_condition, .. } => { crate::disposition::MaintenanceDisposition::NeedsEvidence { missing_refs, revisit_condition: revisit_condition.unwrap_or_else(|| "condition specified by actor".to_string()), } } klbr_runtime::protocol::MaintenanceAssessment::NoChange { reason, evidence_refs } => { let refs = if evidence_refs.is_empty() { w.provenance.source_refs.clone() } else { evidence_refs }; crate::disposition::MaintenanceDisposition::NoChange { reason, evidence_refs: refs, } } klbr_runtime::protocol::MaintenanceAssessment::CandidateFailed { error, candidate_revision } => { crate::disposition::MaintenanceDisposition::CandidateFailed { error, candidate_revision, } } klbr_runtime::protocol::MaintenanceAssessment::Deferred { reason, trigger_ms } => { crate::disposition::MaintenanceDisposition::Deferred { reason, trigger_ms, } } klbr_runtime::protocol::MaintenanceAssessment::Activated { candidate_revision, check_receipts } => { crate::disposition::MaintenanceDisposition::Activated { work_item_id: w.id.clone(), candidate_revision, epoch: final_epoch, check_receipts, } } }) } Err(e) => { tracing::warn!( session = %s_id, err = %e, outcome = %outcome_str, "unknown or invalid assessment outcome rejected; not defaulted to no-change" ); None } } } else { None } });69 unmodified lines now_turn_end, ) .await;
let _ = reg_clone .store() .append( &s_id, "scratchpad", serde_json::json!({ "type": "maintenance.attempt_completed", "item_id": w.id, "purpose": w.purpose, "disposition": disposition, }), ) .await;
match retire_res { Ok(r) if r.is_retired() => { let _ = reg_clone .store() .append( &s_id, "scratchpad", serde_json::json!({ "type": "maintenance.attempt_completed", "item_id": w.id, "purpose": w.purpose, "disposition": disposition, }), ) .await; tracing::info!( session = %s_id, work_id = %w.id, ?disposition, "maintenance work item attempt retired" "maintenance work item attempt retired successfully" ); } Ok(r) => { tracing::warn!( session = %s_id, work_id = %w.id, attempt = %op_id, ?r, "maintenance disposition not retired" "maintenance attempt retirement failed or wrong owner; omitting attempt_completed event" ); let _ = reg_clone .store() .append( &s_id, "scratchpad", serde_json::json!({ "type": "maintenance.attempt_failed", "item_id": w.id, "attempt_id": op_id.to_string(), "reason": format!("retirement returned {:?}", r), }), ) .await; } Err(e) => { tracing::error!( session = %s_id, work_id = %w.id, attempt = %op_id, err = %e, "error applying maintenance disposition" "error applying maintenance disposition; omitting attempt_completed event" ); let _ = reg_clone .store() .append( &s_id, "scratchpad", serde_json::json!({ "type": "maintenance.attempt_failed", "item_id": w.id, "attempt_id": op_id.to_string(), "reason": format!("apply error: {}", e), }), ) .await; } } } else { let retire_res = reg_clone36 unmodified lines
let _ = futures::future::Abortable::new(runner_future, abort_reg).await; entry_clone.retire_runner(&op_id).await; }); entry.attach_runner_task(&op_for_attach, runner_task).await;
Ok(())}
57 unmodified lines let boot_reg = registry.clone(); let boot_llm = llm.clone(); let boot_release = initial_release.clone(); let boot_v2_tx = v2_event_tx.clone(); tokio::spawn(async move { let recovery_task = tokio::spawn(async move { let boot_ws = boot_reg.worker_config().cwd.display().to_string(); let boot_rev = boot_release .as_ref() .map(|r| r.id.clone()) .unwrap_or_else(|| "boot".into()); let _ = crate::observation::drain_all_unobserved_sessions( boot_reg.store(), &boot_ws, &boot_rev, boot_reg.recovery_policy(), ) .await;
let now = match klbr_runtime::store::now_ms() { Ok(n) => n, Err(_) => chrono::Utc::now().timestamp_millis(), };15 unmodified lines } } } }); registry.track_service_task( crate::session_registry::ServiceTaskGroup::Recovery, recovery_task, ); }
tokio::spawn(async move { let driver_task = tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_millis(50)); loop { tokio::select! { _ = driver_shutdown.changed() => break, _ = interval.tick() => { let driver_ws = driver_reg.worker_config().cwd.display().to_string(); let driver_rev = driver_release .as_ref() .map(|r| r.id.clone()) .unwrap_or_else(|| "driver".into()); let _ = crate::observation::drain_all_unobserved_sessions( driver_reg.store(), &driver_ws, &driver_rev, driver_reg.recovery_policy(), ) .await;
let now = match klbr_runtime::store::now_ms() { Ok(n) => n, Err(_) => chrono::Utc::now().timestamp_millis(), };37 unmodified lines } } } }); registry.track_service_task( crate::session_registry::ServiceTaskGroup::Driver, driver_task, );
loop { let accepted = tokio::select! { _ = shutdown.changed() => break,5 unmodified lines let v2_bcast = v2_event_tx.clone(); let snapshot = snapshot.clone(); let memory = memory.clone(); let default_soul = default_soul.clone(); let registry = registry.clone(); let client_reg = registry.clone(); let shutdown = shutdown.clone(); let llm = llm.clone(); let initial_release = initial_release.clone();
tokio::spawn(async move { let client_task = tokio::spawn(async move { if let Err(e) = handle( stream, tx, bcast,1 unmodified line snapshot, memory, history_window, default_soul, registry, client_reg, llm, initial_release, v2_mode, shutdown,2 unmodified lines { tracing::error!(err = %e, "client error"); } }); registry.track_service_task( crate::session_registry::ServiceTaskGroup::Client, client_task, ); }
registry.shutdown_all().await; Ok(())806 unmodified linesExpand all29 unmodified lines3031323334353637383927 unmodified lines67686970717273747576777811 unmodified lines90919293949596979826 unmodified lines12512612712812913013113213313413513613713813914014114223 unmodified lines16616716816915315417017117215617317417516016116216317617717817918018118218318416918518618718818919013 unmodified lines20420520620719119220820921021121221321421519819921621721821920120220320420520620720822022122222322422522622722822923023123223323423523621723723823924024124224322024424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231323031431531631731831932032132225 unmodified lines34834935035126426535235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638756 unmodified lines44444544644733544833733833934034134234344945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261329 unmodified lines Deferred { reason: String, trigger_ms: Option<i64>, }, /// Attempt ended without establishing resolution; objective remains open. Unresolved { reason: String }, /// Candidate failed validation or checks; failure recorded and follow-up enabled. CandidateFailed { error: String, candidate_revision: Option<String>,27 unmodified lines pub fn is_deferred(&self) -> bool { matches!(self, Self::Deferred { .. }) }
pub fn is_unresolved(&self) -> bool { matches!(self, Self::Unresolved { .. }) }
/// Canonical serialized string representation for storage in work_items.disposition. pub fn canonical_disposition_string(&self) -> String { match self { Self::Activated {11 unmodified lines ) } } Self::NoChange { reason, .. } => format!("no-change: {reason}"), Self::Unresolved { reason } => format!("unresolved: {reason}"), Self::NeedsEvidence { missing_refs, revisit_condition, } => {26 unmodified lines last_error: Option<String>,) -> MaintenanceDisposition { // 1. Explicit supported assessment from the actor outranks passive heuristics if let Some(explicit) = explicit_assessment { // Validate an Activated assessment against actual execution facts if explicit.is_activated() && (!activated_candidate_for_work || final_epoch <= initial_epoch) { return MaintenanceDisposition::CandidateFailed { error: "assessment claimed activated, but candidate was not activated for this work item" .to_string(), candidate_revision: None, }; } return explicit; }
// 2. Missing evidence blocks resolution; objective stays unresolved & resumable23 unmodified lines }
// 5. Unrelated activation observed (epoch increased, but NOT for this work item) if final_epoch > initial_epoch && !activated_candidate_for_work { return MaintenanceDisposition::NoChange { reason: "unrelated hook activation observed during attempt; objective unchanged" return MaintenanceDisposition::Unresolved { reason: "unrelated hook activation observed during attempt; objective unchanged and unresolved" .to_string(), evidence_refs: work_item.provenance.source_refs.clone(), }; }
// 6. Normal attempt ended without change MaintenanceDisposition::NoChange { reason: "maintenance attempt finished without candidate activation".to_string(), evidence_refs: work_item.provenance.source_refs.clone(), // 6. Normal attempt ended without candidate activation or explicit assessment MaintenanceDisposition::Unresolved { reason: "maintenance attempt finished without candidate activation or explicit assessment" .to_string(), }}
/// Applies the evaluated disposition to the persistent Store.////// NeedsEvidence leaves the work item unresolved and resumable./// NeedsEvidence and Unresolved retire the attempt while leaving the objective open/// and not immediately runnable without evidence or a due revisit./// Activated/NoChange complete the work item with an explicit receipt./// CandidateFailed marks it failed with failure diagnosis retained.pub async fn apply_maintenance_disposition( store: &Store,13 unmodified lines MaintenanceDisposition::NeedsEvidence { missing_refs, revisit_condition, } => { // Objective stays unresolved and resumable: // 1. Defer the claim so the current attempt is retired // Objective stays unresolved and not immediately runnable without new evidence: // Defer the claim so the current attempt is retired with the blocked revisit gate. // Do NOT immediately call reopen_work_item back to ready. let reason = format!( "needs-evidence: missing={}; condition={}", missing_refs.join(","), revisit_condition ); let res = store .defer_work_item(&work_item.id, attempt_id, &reason) let gate = klbr_runtime::store::RevisitGate::blocked_on(revisit_condition); store .defer_work_item_with_revisit(&work_item.id, attempt_id, &reason, &gate) .await .context("failed to defer work item needing evidence")?;
// 2. Reopen the work item so it returns to pending & ready let _ = store .reopen_work_item(&work_item.id, &reason, missing_refs) .await;
Ok(res) .context("failed to defer work item needing evidence") } MaintenanceDisposition::Unresolved { reason } => { let unres_str = format!("unresolved: {reason}"); let gate = klbr_runtime::store::RevisitGate::blocked_on(reason); store .defer_work_item_with_revisit(&work_item.id, attempt_id, &unres_str, &gate) .await .context("failed to defer unresolved work item") } MaintenanceDisposition::CandidateFailed { error, .. } => { let err_str = format!("candidate-failed: {error}"); store .fail_work_item(&work_item.id, attempt_id, &err_str, now_ms) .await .context("failed to retire work item as failed candidate") } MaintenanceDisposition::Deferred { reason, .. } => { MaintenanceDisposition::Deferred { reason, trigger_ms } => { let def_str = format!("deferred: {reason}"); let gate = match trigger_ms { Some(ms) => klbr_runtime::store::RevisitGate::at(*ms), None => klbr_runtime::store::RevisitGate::blocked_on(reason), }; store .defer_work_item(&work_item.id, attempt_id, &def_str) .defer_work_item_with_revisit(&work_item.id, attempt_id, &def_str, &gate) .await .context("failed to defer work item") } }}
/// Traverses session event history through proper pagination to discover the true/// latest committed event ID (high-watermark) at turn start, avoiding the first-page cap.pub async fn session_head_event_id( store: &Store, session_id: &klbr_runtime::protocol::SessionId,) -> i64 { let mut cursor = 0i64; loop { match store.history(session_id, cursor, 1024).await { Ok(events) if !events.is_empty() => { let count = events.len(); if let Some(last) = events.last() { cursor = last.id; } if count < 1024 { break; } } _ => break, } } cursor}
/// Reads all committed events after the high-watermark, paginating through all available/// pages so attempts with large event volume or history never drop events.pub async fn read_events_after( store: &Store, session_id: &klbr_runtime::protocol::SessionId, after: i64,) -> Vec<klbr_runtime::store::Event> { let mut all_events = Vec::new(); let mut cursor = after; loop { match store.history(session_id, cursor, 1024).await { Ok(events) if !events.is_empty() => { let count = events.len(); if let Some(last) = events.last() { cursor = last.id; } all_events.extend(events); if count < 1024 { break; } } _ => break, } } all_events}
/// Checks whether a genuine hook activation during the attempt was specifically bound/// to the target work item id through the recorded event link.pub fn activation_bound_to_work(events: &[klbr_runtime::store::Event], work_item_id: &str) -> bool { events.iter().any(|ev| { ev.kind == "hooks.activated" && ev.payload.get("work_item_id").and_then(|v| v.as_str()) == Some(work_item_id) })}
#[cfg(test)]mod tests { use super::*; use klbr_runtime::work::{ConsequenceMetadata, Impact, Provenance, RunCause}; use klbr_runtime::{ protocol::SessionId, work::{ConsequenceMetadata, Impact, Provenance, RunCause, WorkStatus}, }; use serde_json::json;
fn sample_work_item(id: &str) -> WorkItem { WorkItem::new( id,25 unmodified lines None, );
assert!( disp.is_no_change(), "Unrelated activation must result in NoChange, not Activated" disp.is_unresolved(), "Unrelated activation must result in Unresolved, not Activated or NoChange" ); assert!(!disp.is_activated()); assert!(!disp.is_no_change()); }
#[test] fn test_absent_assessment_evaluates_to_unresolved() { let work = sample_work_item("work-ended-quietly"); let disp = evaluate_attempt_disposition( &work, 1, 1, "rev-1", "rev-1", false, vec![], vec![], None, None, );
assert!( disp.is_unresolved(), "Ended turn with no assessment must evaluate to Unresolved, not NoChange" ); assert!(!disp.is_activated()); assert!(!disp.is_no_change()); }
#[test] fn test_linked_activation_marks_activated() { let work = sample_work_item("work-linked"); let disp = evaluate_attempt_disposition( &work,56 unmodified lines );
assert_eq!(disp, explicit); }}
/// Checks whether a genuine hook activation during the attempt was specifically bound/// to the target work item id through the recorded event link.pub fn activation_bound_to_work(events: &[klbr_runtime::store::Event], work_item_id: &str) -> bool { events.iter().any(|ev| { ev.kind == "hooks.activated" && ev.payload.get("work_item_id").and_then(|v| v.as_str()) == Some(work_item_id) }) #[tokio::test] async fn test_more_than_2048_history_events_and_conflicting_older_assessment() -> Result<()> { let temp = tempfile::tempdir()?; let store = Store::open(temp.path().join("paginated_history.sqlite"))?; let session = SessionId::fresh(); store.ensure_session(&session).await?;
let work = sample_work_item("work-paginated").with_session(session.to_string()); store.insert_work_item(&work).await?;
// 1. Plant an older conflicting assessment at event ~1 store .append( &session, "scratchpad", json!({ "type": "maintenance.assessed", "work_item_id": work.id, "outcome": "candidate-failed", "reason": "older obsolete failure from distant history", }), ) .await?;
// 2. Append more than 2,048 historical events to exceed two 1,024 pages for i in 0..2100 { store .append( &session, "scratchpad", json!({ "type": "heartbeat", "seq": i, }), ) .await?; }
// 3. Obtain turn start high-watermark using proper pagination let turn_start_event_id = session_head_event_id(&store, &session).await; assert!( turn_start_event_id >= 2101, "High watermark must be at head of history (>2100), got {turn_start_event_id}" );
// 4. Now simulate current turn: commit an explicit valid no-change assessment let source_event = store .append( &session, "scratchpad", json!({ "type": "evidence.source", "verified": true, }), ) .await?;
store .append( &session, "scratchpad", json!({ "type": "maintenance.assessed", "work_item_id": work.id, "outcome": "no-change", "reason": "current turn explicit evidence proves system correct", "missing_refs": [], "check_receipts": [], }), ) .await?;
// 5. Query events during turn using read_events_after let events_during_turn = read_events_after(&store, &session, turn_start_event_id).await; assert_eq!(events_during_turn.len(), 2);
// The older conflicting assessment from step 1 MUST NOT be present in this turn assert!(!events_during_turn.iter().any(|ev| { ev.payload.get("reason").and_then(|r| r.as_str()) == Some("older obsolete failure from distant history") }));
// The current assessment MUST be found let found = events_during_turn .iter() .rev() .find(|ev| { ev.payload.get("type").and_then(|t| t.as_str()) == Some("maintenance.assessed") && ev.payload.get("work_item_id").and_then(|w| w.as_str()) == Some(work.id.as_str()) }) .expect("current turn assessment must be found");
assert_eq!( found.payload.get("outcome").and_then(|o| o.as_str()), Some("no-change") );
// 6. Evaluate and apply disposition; positive control persists Completed let explicit = MaintenanceDisposition::NoChange { reason: "current turn explicit evidence proves system correct".into(), evidence_refs: vec![format!("event:{source_event}")], }; let disp = evaluate_attempt_disposition( &work, 1, 1, "rev-1", "rev-1", false, vec![], vec![], Some(explicit), None, ); let attempt = OperationId::fresh(); assert!(store .claim_work_item(&work.id, &attempt, 1000) .await? .is_claimed()); let retire = apply_maintenance_disposition(&store, &work, &attempt, &disp, 2000).await?; assert!(retire.is_retired());
let persisted = store.get_work_item(&work.id).await?.expect("work"); assert!(matches!(persisted.status, WorkStatus::Completed { .. }));
Ok(()) }
#[tokio::test] async fn test_coherent_retirement_failure_when_attempt_mismatched() -> Result<()> { let temp = tempfile::tempdir()?; let store = Store::open(temp.path().join("coherent_retire.sqlite"))?; let session = SessionId::fresh(); store.ensure_session(&session).await?;
let work = sample_work_item("work-coherent").with_session(session.to_string()); store.insert_work_item(&work).await?;
// Claimed by attempt A let attempt_a = OperationId::fresh(); assert!(store .claim_work_item(&work.id, &attempt_a, 1000) .await? .is_claimed());
// Attempt B tries to apply disposition let attempt_b = OperationId::fresh(); let disp = MaintenanceDisposition::Unresolved { reason: "no change attempted".into(), }; let res = apply_maintenance_disposition(&store, &work, &attempt_b, &disp, 2000).await?; assert!( matches!(res, RetireResult::WrongOwner { .. }), "Attempt B must fail retirement with WrongOwner" ); assert!(!res.is_retired());
// Work item remains claimed by attempt A, not retired or completed let persisted = store.get_work_item(&work.id).await?.expect("work"); assert!(matches!(persisted.status, WorkStatus::Claimed { .. }));
Ok(()) }}2 unmodified lines345678910111259 unmodified lines727374757476777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511610 unmodified lines1271281291301021311321331341351361371381391402 unmodified lines14314414514611314714814915015115215315415515615715815916016116216316416516616712916816917017117217317417517617717817918018118218319 unmodified lines2032042052061622072082092102112122132142152168 unmodified lines2252262272282292302312322332342357 unmodified lines24324424524619424724824925025125225325425525625725825926026126226320020120220326420426526626726826927027121227227327427527627727827928028128228328428528621421521628728828929029129222029329422229529629729829930030130230322530430530630730830931031131231331431531631731831932032132232332432532632732811 unmodified lines3403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913922533933942553953963973983995 unmodified lines40540640740840941041141241341441541627241741841942042142227842342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350419 unmodified lines52452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455538 unmodified lines5945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336342 unmodified lines//! Separates raw evidence capture from filtering policy.//! Error occurrences derive stable identities from committed event sequences,//! preserving causal trace references and deduplicating reliably across redeliveries.
use std::collections::HashMap;
use anyhow::Result;use klbr_runtime::{protocol::SessionId, store::Store};use serde::{Deserialize, Serialize};
59 unmodified lines }}
/// Categorizes raw diagnostic output so policy distinguishes execution/// failures from warnings and skipped peer calls./// failures from warnings, uncertain diagnostics, and skipped peer calls.#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]pub enum DiagnosticClassification { ExecutionFailure, RecoveredError, StderrWarning, SkippedPeerCall, Uncertain, NormalOutput,}
/// Checks whether a diagnostic string indicates a benign warning rather than a failure.pub fn is_warning_diagnostic(text: &str) -> bool { let lower = text.to_ascii_lowercase(); for line in lower.lines().map(str::trim).filter(|l| !l.is_empty()) { if line.starts_with("warning:") || line.starts_with("warn:") || line.starts_with("[warning]") || line.contains("warning:") || line.contains("userwarning:") || line.contains("deprecationwarning:") || line.contains("syntaxwarning:") || line.contains("runtimewarning:") || line.contains("futurewarning:") { return true; } if let Some((prefix, _)) = line.split_once(':') { let p = prefix.trim(); if p.ends_with("warning") { return true; } } } false}
/// Classifies a committed event to distinguish real failures, warnings, and skipped peer calls.pub fn classify_diagnostic( kind: &str, payload: &serde_json::Value,10 unmodified lines .unwrap_or_default() .to_string(); let sig = extract_error_signature(&err); let effective_sig = if sig.is_empty() { "ExecutionError".to_string() let first_line = err .lines() .map(str::trim) .find(|l| !l.is_empty()) .unwrap_or("ExecutionError"); truncate_signature(first_line, 96) } else { sig }; return (2 unmodified lines Some(err), ); }
// Check if cell output had stderr // Cell completed with state != "error" (e.g. ok). // Inspect whether stderr output contained an actual recovered error, // a benign warning, or an uncertain diagnostic. if let Some(outputs) = outcome .and_then(|o| o.get("output")) .and_then(|o| o.as_array()) { for out in outputs { let stream = out.get("stream").and_then(|s| s.as_str()); if stream == Some("stderr") { let text = out.get("text").and_then(|t| t.as_str()).unwrap_or_default(); if text.trim().is_empty() { continue; } let sig = extract_error_signature(text); if !sig.is_empty() { return ( DiagnosticClassification::RecoveredError, Some(sig), Some(text.to_string()), ); } else if !text.trim().is_empty() { } else if is_warning_diagnostic(text) { return ( DiagnosticClassification::StderrWarning, None, Some(text.to_string()), ); } else { return ( DiagnosticClassification::Uncertain, None, Some(text.to_string()), ); } } } }19 unmodified lines .unwrap_or_default() .to_string(); let sig = extract_error_signature(&err); let effective_sig = if sig.is_empty() { "ToolError".to_string() let first_line = err .lines() .map(str::trim) .find(|l| !l.is_empty()) .unwrap_or("ToolError"); truncate_signature(first_line, 96) } else { sig }; return (8 unmodified lines _ => (DiagnosticClassification::NormalOutput, None, None), }}
/// Extracts a structured error signature only when an explicit error/exception pattern exists./// Returns an empty string if the text is benign or does not match an error pattern,/// preventing benign warnings or normal output from being classified as failure signatures.pub fn extract_error_signature(text: &str) -> String { let text = text.trim(); if text.is_empty() { return String::new();7 unmodified lines
for line in text.lines().map(str::trim).filter(|l| !l.is_empty()) { if let Some((prefix, _)) = line.split_once(':') { let p = prefix.trim(); if p.ends_with("Error") || p.ends_with("Exception") || p.eq_ignore_ascii_case("error") { if (p.ends_with("Error") || p.ends_with("Exception") || p.eq_ignore_ascii_case("error") || p.eq_ignore_ascii_case("fatal") || p.eq_ignore_ascii_case("panic")) && !p.ends_with("Warning") { return truncate_signature(line, 96); } } let lower = line.to_ascii_lowercase(); if lower.starts_with("error:") || lower.starts_with("fatal:") || lower.starts_with("panic:") { return truncate_signature(line, 96); } }
if let Some(first_line) = text.lines().map(str::trim).find(|l| !l.is_empty()) { truncate_signature(first_line, 96) } else { String::new() String::new() }}
fn truncate_signature(s: &str, max_len: usize) -> String { let collapsed: String = s.split_whitespace().collect::<Vec<_>>().join(" "); collapsed.chars().take(max_len).collect()}
/// Replays committed events for a session from the ordered observer checkpoint.#[derive(Debug, Clone)]pub struct ReplayPageResult { pub created_work_ids: Vec<String>, pub processed_count: usize, pub has_more: bool, pub highest_seq: i64,}
pub const DEFAULT_PAGE_SIZE: usize = 256;
/// Marker stamped on observations when an originating execution record cannot be resolved.pub const UNRESOLVED_REVISION: &str = "unresolved_revision";
/// Replays a single bounded page of unobserved events for a session from the ordered checkpoint.////// Guaranteed to discover unobserved occurrences after a crash or restart/// between event commit and observation callback.pub async fn replay_session_observations(/// Resolves originating revision and workspace from the linked source execution/// rather than stamping current environment values onto historical records.pub async fn replay_session_observations_page( store: &Store, session_id: &SessionId, workspace: &str, component_revision: &str, fallback_revision: &str, policy: &RecoveryObservationPolicy,) -> Result<Vec<String>> { page_size: usize,) -> Result<ReplayPageResult> { let partition = format!("observer:recovered:{session_id}"); let cursor = store.get_observer_cursor(&partition).await?; let limit = if page_size == 0 { DEFAULT_PAGE_SIZE } else { page_size }; let events = store.history(session_id, cursor, 256).await?; let events = store.history(session_id, cursor, limit).await?;
let mut created_work_ids = Vec::new(); let mut highest_seq = cursor; let mut execution_rev_cache: HashMap<String, String> = HashMap::new();
let fetched_count = events.len(); let has_more = fetched_count >= limit;
for event in &events { // Track execution.started to learn execution -> revision mappings if event.kind == "execution.started" { if let (Some(exec_id), Some(rev)) = ( event.payload.get("execution_id").and_then(|v| v.as_str()), event.payload.get("revision").and_then(|v| v.as_str()), ) { execution_rev_cache.insert(exec_id.to_string(), rev.to_string()); } } }
for event in events { if event.id <= cursor { continue; }11 unmodified lines "cell.execution" } else { "tool.execution" };
// Resolve originating revision from linked source execution let exec_id_opt = event.payload.get("execution_id").and_then(|v| v.as_str()); let mut originating_rev = exec_id_opt.and_then(|id| execution_rev_cache.get(id).cloned());
// If not in cache, query authoritative execution record in store if originating_rev.is_none() { if let Some(exec_id) = exec_id_opt { originating_rev = store.get_execution_revision(exec_id).await?; } }
// If still not found, check payload directly if originating_rev.is_none() { originating_rev = event .payload .get("revision") .and_then(|v| v.as_str()) .map(ToString::to_string); }
let effective_rev = match originating_rev { Some(rev) => rev, None if exec_id_opt.is_some() => { let exec_id = exec_id_opt.unwrap(); tracing::warn!( execution_id = exec_id, event_seq = event.id, "originating revision could not be resolved from executions or payload; preserving observation with unresolved provenance" ); UNRESOLVED_REVISION.to_string() } None => fallback_revision.to_string(), };
// Resolve workspace from payload if present, else fallback let originating_ws = event .payload .get("workspace") .or_else(|| event.payload.get("workspace_path")) .and_then(|v| v.as_str()) .map(ToString::to_string); let effective_ws = originating_ws.unwrap_or_else(|| workspace.to_string());
let wid = crate::daemon::observe_recovered_error( store, session_id, op, component_revision, &effective_rev, &sig, workspace, &effective_ws, event.created_ms, event.id, policy, )5 unmodified lines // Failure event without signature: advance cursor so it is not rescanned every turn let _ = store.advance_observer_cursor(&partition, event.id).await; } } DiagnosticClassification::Uncertain => { // Preserve uncertain observation with its original evidence in SQLite, // advancing cursor so processing continues without inventing a failure. let _ = store.advance_observer_cursor(&partition, event.id).await; } DiagnosticClassification::StderrWarning | DiagnosticClassification::SkippedPeerCall | DiagnosticClassification::NormalOutput => { // Non-failure: we can advance cursor so we don't re-examine clean events // Non-failure: advance cursor past clean events let _ = store.advance_observer_cursor(&partition, event.id).await; } } }
Ok(created_work_ids) Ok(ReplayPageResult { created_work_ids, processed_count: fetched_count, has_more, highest_seq, })}
/// Replays committed events for a session from the ordered observer checkpoint,/// draining bounded pages while backlog remains.////// Guaranteed to discover unobserved occurrences after a crash or restart/// between event commit and observation callback, independent of model turns.pub async fn replay_session_observations( store: &Store, session_id: &SessionId, workspace: &str, fallback_revision: &str, policy: &RecoveryObservationPolicy,) -> Result<Vec<String>> { let mut all_created = Vec::new(); loop { let page = replay_session_observations_page( store, session_id, workspace, fallback_revision, policy, DEFAULT_PAGE_SIZE, ) .await?; all_created.extend(page.created_work_ids); if !page.has_more { break; } } Ok(all_created)}
/// Drains unobserved events across all sessions in the store./// Suitable for invocation during daemon startup recovery and periodic driver loops./// Drains unobserved events across all sessions in the store./// Suitable for invocation during daemon startup recovery and periodic driver loops./// Resilient across sessions: an unexpected failure in one session does not abort draining for others.pub async fn drain_all_unobserved_sessions( store: &Store, fallback_workspace: &str, fallback_revision: &str, policy: &RecoveryObservationPolicy,) -> Result<Vec<String>> { let sessions = match store.sessions_with_unobserved_events().await { Ok(s) => s, Err(_) => store.all_sessions().await.unwrap_or_default(), }; let mut all_created = Vec::new(); for sid in sessions { match replay_session_observations( store, &sid, fallback_workspace, fallback_revision, policy, ) .await { Ok(created) => all_created.extend(created), Err(e) => { tracing::warn!( session = %sid, error = %e, "failed to replay observations for session; continuing drain for subsequent sessions" ); } } } Ok(all_created)}
#[cfg(test)]mod tests { use super::*;
19 unmodified lines ); }
#[test] fn test_extract_error_signature_returns_empty_for_benign_warning() { assert_eq!( extract_error_signature("UserWarning: pkg_resources is deprecated as an API"), "" ); assert_eq!( extract_error_signature("warning: unused import: `std::path::PathBuf`"), "" ); assert_eq!( extract_error_signature("DeprecationWarning: Call to deprecated method"), "" ); }
#[test] fn test_is_warning_diagnostic() { assert!(is_warning_diagnostic("UserWarning: feature deprecated")); assert!(is_warning_diagnostic("warning: unused variable")); assert!(is_warning_diagnostic("[WARNING] low memory threshold")); assert!(!is_warning_diagnostic("RuntimeError: connection refused")); }
#[test] fn test_policy_evaluation_with_grouping_window() { let policy = RecoveryObservationPolicy { min_recurrence: 2, grouping_window_ms: 10_000,38 unmodified lines let (class, sig, _) = classify_diagnostic("scratchpad", &err_payload); assert_eq!(class, DiagnosticClassification::ExecutionFailure); assert_eq!(sig.as_deref(), Some("ValueError: bad argument")); }
#[test] fn test_diagnostic_classification_distinguishes_benign_warning_from_error() { let warning_payload = serde_json::json!({ "state": "ok", "outcome": { "output": [ { "stream": "stderr", "text": "UserWarning: deprecated API call", } ] } }); let (class, sig, text) = classify_diagnostic("execution.completed", &warning_payload); assert_eq!(class, DiagnosticClassification::StderrWarning); assert_eq!(sig, None); assert!(text.unwrap().contains("UserWarning"));
// Cell status error with warning in stderr is still ExecutionFailure let error_payload = serde_json::json!({ "state": "error", "outcome": { "error": "ZeroDivisionError: division by zero", "output": [ { "stream": "stderr", "text": "UserWarning: some warning before crash", } ] } }); let (class, sig, _) = classify_diagnostic("execution.completed", &error_payload); assert_eq!(class, DiagnosticClassification::ExecutionFailure); assert_eq!(sig.as_deref(), Some("ZeroDivisionError: division by zero")); }}