Bind maintenance identity, evidence, and outcomes to committed records
152 unmodified linesExpand all153154155156157158159160152 unmodified linesExpand all{"id":"int-8118f86d3327d53f3079164428a13a8b","kind":"field_change","created_at":"2026-09-06T21:56:33.064471Z","actor":"dawn","issue_id":"klbr-ddx","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-c6feeac0bad16f805a1fe9faef28c111","kind":"field_change","created_at":"2026-09-06T21:59:30.618455Z","actor":"dawn","issue_id":"klbr-0uk","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-9ca99ecb4f31ccfb8495a14cf25fe239","kind":"field_change","created_at":"2026-09-06T22:14:24.914459Z","actor":"dawn","issue_id":"klbr-cw4","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"id":"int-166a41e09442a5853a26da136991a81e","kind":"field_change","created_at":"2026-09-06T22:15:44.586942Z","actor":"dawn","issue_id":"klbr-4zb","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Closed"}}{"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"}}445 unmodified linesExpand all4464474484494504504514524534541 unmodified line4564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904614624914924934944954964974984684995005015025032 unmodified lines5065075085095105115125135145155165175185195205215222 unmodified lines5255265275284895295305315325336 unmodified lines540541542543504544545506546547548549510550551552553554555556557558559518519560561562563564565311 unmodified linesExpand all877878879880881882883884885886887888889890891892893227 unmodified linesExpand all112111221123112410741125112611271128112910791080108110821083108410851086108710881089109010911092109310941095109611301131113211331134113511361137113811391140114161 unmodified lines1203120412051206116612071208116811691170117111721173117411751176117712091210121111791180118111821183118411851186118711881189121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286119211931194119511961197119812871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137012001201120212031204120512061371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413187 unmodified linesExpand all160116021603160416051606160716081609699 unmodified linesExpand all23092310231123122110231323142315231623176 unmodified lines232423252326232721252328232923302331233280 unmodified lines445 unmodified linesExpand all component_revision: &str, signature: &str, workspace: &str, now_ms: i64, seq: usize, source_event_seq: i64, policy: &crate::observation::RecoveryObservationPolicy,) -> Result<Option<String>> { use klbr_core::capability_check::compute_action_identity; use klbr_runtime::work::{1 unmodified line RunCause, WorkItem, };
let action_id = compute_action_identity(operation_id, component_revision, signature, workspace);
// If source_event_seq is 0, attempt to find matching committed event in session history let resolved_seq = if source_event_seq > 0 { source_event_seq } else { let found = store .history(session_id, 0, 50) .await .ok() .and_then(|events| { events.into_iter().rev().find_map(|ev| { let text = ev.payload.to_string(); if text.contains(signature) { Some(ev.id) } else { None } }) }); match found { Some(seq) => seq, None => { anyhow::bail!( "cannot observe occurrence without committed source event: no matching event found for signature '{signature}'" ); } } };
// Occurrence ID is deterministic based on session, action, and committed event seq let occurrence_id = OccurrenceId::from(format!( "recovered:{}:{}:{}:{}", session_id, action_id, seq, now_ms "recovered:{}:{}:{}", session_id, action_id, resolved_seq ));
let existing = store.get_observer_case(&action_id).await.ok().flatten(); let current_count = existing.as_ref().map_or(0, |c| c.count); let last_seen_ms = existing.as_ref().map(|c| c.last_seen_ms);
let action = policy.evaluate(current_count); let action = policy.evaluate_with_timing(current_count, last_seen_ms, now_ms); if action == crate::observation::RecoveryAction::Reject { return Ok(None); }
2 unmodified lines .as_ref() .and_then(|c| c.work_item_id.as_ref()) .is_none();
let evidence_ref = format!("event:{resolved_seq}"); let mut all_evidence = existing .as_ref() .map(|c| c.evidence_refs.clone()) .unwrap_or_default(); if !all_evidence.contains(&evidence_ref) { all_evidence.push(evidence_ref.clone()); }
let case = ObserverCaseRecord { id: action_id.clone(), session_id: session_id.to_string(), operation: operation_id.to_string(),2 unmodified lines workspace: workspace.to_string(), count: 1, first_seen_ms: now_ms, last_seen_ms: now_ms, evidence_refs: vec![], evidence_refs: all_evidence.clone(), work_item_id: if should_create_work { Some(format!("work-{}", action_id)) } else { existing.and_then(|c| c.work_item_id)6 unmodified lines format!("work-{}", action_id), format!("investigate recurring failure: {}", operation_id), RunCause::MaintenanceReady { item_id: action_id.clone(), purpose: "investigate".into(), purpose: format!("investigate failure in {}: {}", operation_id, signature), }, ConsequenceMetadata::new(Impact::Low), ConsequenceMetadata::new(Impact::Normal), Provenance::agent("observer.recovered", now_ms), now_ms, ) .with_session(session_id.to_string()), .with_session(session_id.to_string()) .with_evidence_refs(all_evidence), ) } else { None };
let partition = format!("observer:recovered:{session_id}"); let result = store .commit_observation_case_with_cursor( "observer:recovered", now_ms, &partition, resolved_seq, &occurrence_id, case, work_item, )311 unmodified linesExpand all } }
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 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 !recovered_errors.is_empty() && failed_error.is_none() { 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();
for (seq_idx, (sig, _text)) in recovered_errors .iter() .take(policy.max_per_turn) .enumerate() { let _ = observe_recovered_error( reg_clone.store(), &s_id, "cell.execution", &rev, sig, &workspace, now_turn_end, seq_idx, policy, ) .await; } let _ = crate::observation::replay_session_observations( reg_clone.store(), &s_id, &workspace, &rev, policy, ) .await; }
if let Some(y) = yielded { if let Some(ref w) = work_item {61 unmodified lines break; } } else { if let Some(ref w) = work_item { let disposition = if matches!(w.cause, RunCause::MaintenanceReady { .. }) { if matches!(w.cause, RunCause::MaintenanceReady { .. }) { let final_epoch = entry_clone.runtime.hook_epoch().await; if final_epoch > initial_epoch { "repaired" } else if !recovered_errors.is_empty() { "rejected" } else { "attempt_ended" } } else { "completed" }; let final_policy = entry_clone.runtime.policy().await; let final_rev = final_policy.release.id.clone();
let retire_res = reg_clone .store() .complete_work_item(&w.id, &op_id, disposition, now_turn_end) .await; match retire_res { Ok(r) if r.is_retired() => { tracing::info!( session = %s_id, work_id = %w.id, disposition, "work item retired as completed" // 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();
// 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(), }, }) } else { None } });
// 2. Candidate activation must be explicitly bound to THIS work item let activated_for_this_work = crate::disposition::activation_bound_to_work( &events_during_turn, &w.id, );
// 3. Collect executed check receipts from events committed during this turn let mut check_receipts: Vec<String> = Vec::new(); for ev in &events_during_turn { if let Some(check_id) = ev .payload .get("check_id") .or_else(|| ev.payload.get("receipt_id")) .and_then(|v| v.as_str()) { check_receipts.push(check_id.to_string()); } } Ok(r) => { tracing::warn!( session = %s_id, work_id = %w.id, ?r, "complete_work_item not retired" );
// 4. Check for missing evidence references in store let mut missing_evidence: Vec<String> = Vec::new(); for r in &w.provenance.source_refs { if let Some(rest) = r.strip_prefix("event:") { if let Ok(seq) = rest.parse::<i64>() { if reg_clone .store() .get_event(&s_id, seq) .await .ok() .flatten() .is_none() { missing_evidence.push(r.clone()); } } } else if let Some(case_id) = r.strip_prefix("case:") { if reg_clone .store() .get_observer_case(case_id) .await .ok() .flatten() .is_none() { missing_evidence.push(r.clone()); } } }
let disposition = crate::disposition::evaluate_attempt_disposition( w, initial_epoch, final_epoch, &initial_rev, &final_rev, activated_for_this_work, check_receipts, missing_evidence, explicit_assessment, failed_error.clone(), );
let retire_res = crate::disposition::apply_maintenance_disposition( reg_clone.store(), w, &op_id, &disposition, 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() => { tracing::info!( session = %s_id, work_id = %w.id, ?disposition, "maintenance work item attempt retired" ); } Ok(r) => { tracing::warn!( session = %s_id, work_id = %w.id, ?r, "maintenance disposition not retired" ); } Err(e) => { tracing::error!( session = %s_id, work_id = %w.id, err = %e, "error completing work item" ); Err(e) => { tracing::error!( session = %s_id, work_id = %w.id, err = %e, "error applying maintenance disposition" ); } } } else { let retire_res = reg_clone .store() .complete_work_item(&w.id, &op_id, "completed", now_turn_end) .await; match retire_res { Ok(r) if r.is_retired() => { tracing::info!( session = %s_id, work_id = %w.id, "work item retired as completed" ); } Ok(r) => { tracing::warn!( session = %s_id, work_id = %w.id, ?r, "complete_work_item not retired" ); } Err(e) => { tracing::error!( session = %s_id, work_id = %w.id, err = %e, "error completing work item" ); } } } } let _ = entry_clone .transition_to_ready_with_store(reg_clone.store())187 unmodified linesExpand all } }); }
registry.shutdown_all().await; Ok(())}
async fn send_msg(ws_tx: &mut WsSink, msg: &ServerMsg) -> Result<()> {699 unmodified linesExpand all "rev1", "TypeError: bad await", "/tmp/test", 1000, 0, 1, &policy, ) .await?; assert_eq!(res1, None, "first occurrence should not create work item");6 unmodified lines "rev1", "TypeError: bad await", "/tmp/test", 2000, 1, 2, &policy, ) .await?; assert!(res2.is_some(), "second occurrence should create work item");80 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344//! Explicit maintenance outcomes, objective dispositions, and attempt lifecycle.//!//! Separates attempt execution outcome from durable objective disposition.//! Objectives are only retired when supported by verifiable receipts;//! missing evidence or unrelated activations leave work unresolved and resumable.
use anyhow::{Context, Result};use klbr_runtime::{ protocol::OperationId, store::Store, work::{RetireResult, WorkItem},};use serde::{Deserialize, Serialize};
/// Explicit domain assessment outcome for a maintenance attempt.#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]#[serde(tag = "outcome", rename_all = "kebab-case")]pub enum MaintenanceDisposition { /// Work cannot proceed without missing evidence; stays unresolved and resumable. NeedsEvidence { missing_refs: Vec<String>, revisit_condition: String, }, /// Explicit supported assessment that no change is warranted. NoChange { reason: String, evidence_refs: Vec<String>, }, /// Objective retained but deferred until a checkpoint or next trigger. Deferred { reason: String, trigger_ms: Option<i64>, }, /// Candidate failed validation or checks; failure recorded and follow-up enabled. CandidateFailed { error: String, candidate_revision: Option<String>, }, /// Candidate was activated for this specific work item; remains under observation. Activated { work_item_id: String, candidate_revision: String, epoch: i64, check_receipts: Vec<String>, },}
impl MaintenanceDisposition { pub fn is_activated(&self) -> bool { matches!(self, Self::Activated { .. }) }
pub fn is_needs_evidence(&self) -> bool { matches!(self, Self::NeedsEvidence { .. }) }
pub fn is_no_change(&self) -> bool { matches!(self, Self::NoChange { .. }) }
pub fn is_candidate_failed(&self) -> bool { matches!(self, Self::CandidateFailed { .. }) }
pub fn is_deferred(&self) -> bool { matches!(self, Self::Deferred { .. }) }
/// Canonical serialized string representation for storage in work_items.disposition. pub fn canonical_disposition_string(&self) -> String { match self { Self::Activated { candidate_revision, epoch, check_receipts, .. } => { if check_receipts.is_empty() { format!("activated:revision={candidate_revision}:epoch={epoch}") } else { format!( "activated:revision={candidate_revision}:epoch={epoch}:checks={}", check_receipts.join(",") ) } } Self::NoChange { reason, .. } => format!("no-change: {reason}"), Self::NeedsEvidence { missing_refs, revisit_condition, } => { format!( "needs-evidence: missing={}; condition={}", missing_refs.join(","), revisit_condition ) } Self::CandidateFailed { error, .. } => format!("candidate-failed: {error}"), Self::Deferred { reason, .. } => format!("deferred: {reason}"), } }}
/// Evaluates the appropriate maintenance disposition from attempt facts.////// An epoch increase alone is NOT proof of benefit; missing evidence or/// unrelated activations must not mark the objective repaired.pub fn evaluate_attempt_disposition( work_item: &WorkItem, initial_epoch: i64, final_epoch: i64, _initial_revision: &str, final_revision: &str, activated_candidate_for_work: bool, check_receipts: Vec<String>, missing_evidence: Vec<String>, explicit_assessment: Option<MaintenanceDisposition>, last_error: Option<String>,) -> MaintenanceDisposition { // 1. Explicit supported assessment from the actor outranks passive heuristics if let Some(explicit) = explicit_assessment { return explicit; }
// 2. Missing evidence blocks resolution; objective stays unresolved & resumable if !missing_evidence.is_empty() { return MaintenanceDisposition::NeedsEvidence { missing_refs: missing_evidence, revisit_condition: "awaiting unresolvable source evidence".to_string(), }; }
// 3. Genuine candidate activation linked specifically to this work item if activated_candidate_for_work && final_epoch > initial_epoch { return MaintenanceDisposition::Activated { work_item_id: work_item.id.clone(), candidate_revision: final_revision.to_string(), epoch: final_epoch, check_receipts, }; }
// 4. Execution failure during candidate evaluation if let Some(err) = last_error { return MaintenanceDisposition::CandidateFailed { error: err, candidate_revision: None, }; }
// 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" .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(), }}
/// Applies the evaluated disposition to the persistent Store.////// NeedsEvidence leaves the work item unresolved and resumable./// 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, work_item: &WorkItem, attempt_id: &OperationId, disposition: &MaintenanceDisposition, now_ms: i64,) -> Result<RetireResult> { match disposition { MaintenanceDisposition::Activated { .. } | MaintenanceDisposition::NoChange { .. } => { let disp_str = disposition.canonical_disposition_string(); store .complete_work_item(&work_item.id, attempt_id, &disp_str, now_ms) .await .context("failed to complete work item with disposition") } MaintenanceDisposition::NeedsEvidence { missing_refs, revisit_condition, } => { // Objective stays unresolved and resumable: // 1. Defer the claim so the current attempt is retired let reason = format!( "needs-evidence: missing={}; condition={}", missing_refs.join(","), revisit_condition ); let res = store .defer_work_item(&work_item.id, attempt_id, &reason) .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) } 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, .. } => { let def_str = format!("deferred: {reason}"); store .defer_work_item(&work_item.id, attempt_id, &def_str) .await .context("failed to defer work item") } }}
#[cfg(test)]mod tests { use super::*; use klbr_runtime::work::{ConsequenceMetadata, Impact, Provenance, RunCause};
fn sample_work_item(id: &str) -> WorkItem { WorkItem::new( id, "test repair", RunCause::MaintenanceReady { item_id: id.into(), purpose: "repair".into(), }, ConsequenceMetadata::new(Impact::Normal), Provenance::agent("test", 1000), 1000, ) }
#[test] fn test_unrelated_activation_does_not_mark_objective_repaired() { let work = sample_work_item("work-unrelated"); // epoch increased from 1 to 2, but activated_candidate_for_work is FALSE let disp = evaluate_attempt_disposition( &work, 1, 2, "rev-1", "rev-unrelated", false, // NOT activated for this work! vec![], vec![], None, None, );
assert!( disp.is_no_change(), "Unrelated activation must result in NoChange, not Activated" ); assert!(!disp.is_activated()); }
#[test] fn test_linked_activation_marks_activated() { let work = sample_work_item("work-linked"); let disp = evaluate_attempt_disposition( &work, 1, 2, "rev-1", "rev-candidate", true, // genuine activation for this work vec!["receipt-chk1".into()], vec![], None, None, );
assert!(disp.is_activated()); assert_eq!( disp.canonical_disposition_string(), "activated:revision=rev-candidate:epoch=2:checks=receipt-chk1" ); }
#[test] fn test_missing_evidence_triggers_needs_evidence() { let work = sample_work_item("work-missing"); let disp = evaluate_attempt_disposition( &work, 1, 1, "rev-1", "rev-1", false, vec![], vec!["event:999".into()], // missing ref None, None, );
assert!(disp.is_needs_evidence()); }
#[test] fn test_explicit_assessment_outranks_heuristics() { let work = sample_work_item("work-explicit"); let explicit = MaintenanceDisposition::NoChange { reason: "diagnostic audit contradicted recorded telemetry".into(), evidence_refs: vec!["event:1".into()], };
let disp = evaluate_attempt_disposition( &work, 1, 2, // epoch even increased "rev-1", "rev-2", true, vec![], vec![], Some(explicit.clone()), None, );
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) })}123456pub mod daemon;pub mod disposition;pub mod ingress;pub mod maintenance;pub mod observation;pub mod session_registry;208 unmodified linesExpand all209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245213246247248249250357 unmodified linesExpand all208 unmodified linesExpand all let final_epoch = session.hook_epoch().await; let policy = session.policy().await; let final_revision = policy.release.id.clone();
let mut missing_refs = Vec::new(); for r in &ctx.evidence_refs { if let Some(rest) = r.strip_prefix("event:") { if let Ok(seq) = rest.parse::<i64>() { if session .store .get_event(&session.id, seq) .await .ok() .flatten() .is_none() { missing_refs.push(r.clone()); } } } else if let Some(case_id) = r.strip_prefix("case:") { if session .store .get_observer_case(case_id) .await .ok() .flatten() .is_none() { missing_refs.push(r.clone()); } } }
let verdict = if !missing_refs.is_empty() { MaintenanceVerdict::NeedsEvidence { missing: missing_refs, } let verdict = if final_epoch > initial_epoch { } else if final_epoch > initial_epoch { MaintenanceVerdict::Activated { revision: final_revision, epoch: final_epoch, }357 unmodified linesExpand all12345678910111220 unmodified lines333435363738394041424344454647484950515253545556575859606162636432656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118225 unmodified lines208209210211737475767721221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128221 unmodified lines304305306307107308309310110311312313112113114314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354//! Lightweight observation for execution errors, occurrence identity, and ordered replay.//!//! 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 anyhow::Result;use klbr_runtime::{protocol::SessionId, store::Store};use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]pub enum RecoveryAction {20 unmodified lines}
impl RecoveryObservationPolicy { pub fn evaluate(&self, current_count: i64) -> RecoveryAction { self.evaluate_with_timing(current_count, None, 0) }
pub fn evaluate_with_timing( &self, current_count: i64, last_seen_ms: Option<i64>, now_ms: i64, ) -> RecoveryAction { if self.min_recurrence == 0 { return RecoveryAction::Reject; }
// Consult grouping window: if the previous occurrence is older than // grouping_window_ms, treat this as a fresh start (count = 0). let effective_count = if let Some(last) = last_seen_ms { if self.grouping_window_ms > 0 && now_ms > last && (now_ms - last) > self.grouping_window_ms { 0 } else { current_count } } else { current_count };
if (current_count + 1) >= self.min_recurrence as i64 { if (effective_count + 1) >= self.min_recurrence as i64 { RecoveryAction::CreateWork } else { RecoveryAction::GroupOnly } }}
/// Categorizes raw diagnostic output so policy distinguishes execution/// failures from warnings and skipped peer calls.#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]pub enum DiagnosticClassification { ExecutionFailure, RecoveredError, StderrWarning, SkippedPeerCall, NormalOutput,}
/// Classifies a committed event to distinguish real failures, warnings, and skipped peer calls.pub fn classify_diagnostic( kind: &str, payload: &serde_json::Value,) -> (DiagnosticClassification, Option<String>, Option<String>) { match kind { "execution.completed" => { let state = payload.get("state").and_then(|v| v.as_str()); let outcome = payload.get("outcome");
if state == Some("error") { let err = outcome .and_then(|o| o.get("error")) .and_then(|e| e.as_str()) .unwrap_or_default() .to_string(); let sig = extract_error_signature(&err); let effective_sig = if sig.is_empty() { "ExecutionError".to_string() } else { sig }; return ( DiagnosticClassification::ExecutionFailure, Some(effective_sig), Some(err), ); }
// Check if cell output had stderr 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(); 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() { return ( DiagnosticClassification::StderrWarning, None, Some(text.to_string()), ); } } } }
(DiagnosticClassification::NormalOutput, None, None) } "scratchpad" => { let ptype = payload.get("type").and_then(|v| v.as_str()); if ptype == Some("tool_result") { let status = payload.get("status").and_then(|v| v.as_str()); if status == Some("skipped") { let reason = payload .get("reason") .and_then(|v| v.as_str()) .map(ToString::to_string); return (DiagnosticClassification::SkippedPeerCall, None, reason); } if status == Some("error") { let err = payload .get("error") .and_then(|v| v.as_str()) .or_else(|| payload.get("content").and_then(|v| v.as_str())) .unwrap_or_default() .to_string(); let sig = extract_error_signature(&err); let effective_sig = if sig.is_empty() { "ToolError".to_string() } else { sig }; return ( DiagnosticClassification::ExecutionFailure, Some(effective_sig), Some(err), ); } } (DiagnosticClassification::NormalOutput, None, None) } _ => (DiagnosticClassification::NormalOutput, None, None), }}
pub fn extract_error_signature(text: &str) -> String { let text = text.trim(); if text.is_empty() { return String::new();25 unmodified lines let collapsed: String = s.split_whitespace().collect::<Vec<_>>().join(" "); collapsed.chars().take(max_len).collect()}
#[derive(Debug, Clone)]pub struct RecoveredError { pub signature: String, pub text: String, pub operation: String,/// Replays committed events for a session from the ordered observer checkpoint.////// Guaranteed to discover unobserved occurrences after a crash or restart/// between event commit and observation callback.pub async fn replay_session_observations( store: &Store, session_id: &SessionId, workspace: &str, component_revision: &str, policy: &RecoveryObservationPolicy,) -> Result<Vec<String>> { let partition = format!("observer:recovered:{session_id}"); let cursor = store.get_observer_cursor(&partition).await?; let events = store.history(session_id, cursor, 256).await?;
let mut created_work_ids = Vec::new(); let mut highest_seq = cursor;
for event in events { if event.id <= cursor { continue; } if event.id > highest_seq { highest_seq = event.id; }
let (classification, sig_opt, _text_opt) = classify_diagnostic(&event.kind, &event.payload);
match classification { DiagnosticClassification::ExecutionFailure | DiagnosticClassification::RecoveredError => { if let Some(sig) = sig_opt { let op = if event.kind == "execution.completed" { "cell.execution" } else { "tool.execution" }; let wid = crate::daemon::observe_recovered_error( store, session_id, op, component_revision, &sig, workspace, event.created_ms, event.id, policy, ) .await?; if let Some(id) = wid { created_work_ids.push(id); } } else { // Failure event without signature: advance cursor so it is not rescanned every turn 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 let _ = store.advance_observer_cursor(&partition, event.id).await; } } }
Ok(created_work_ids)}
#[cfg(test)]mod tests {21 unmodified lines ); }
#[test] fn test_policy_evaluation() { fn test_policy_evaluation_with_grouping_window() { let policy = RecoveryObservationPolicy { min_recurrence: 2, ..Default::default() grouping_window_ms: 10_000, max_per_turn: 5, }; assert_eq!(policy.evaluate(0), RecoveryAction::GroupOnly); assert_eq!(policy.evaluate(1), RecoveryAction::CreateWork); assert_eq!(policy.evaluate(5), RecoveryAction::CreateWork);
// Within window: increments count and triggers CreateWork on threshold assert_eq!( policy.evaluate_with_timing(0, Some(1000), 2000), RecoveryAction::GroupOnly ); assert_eq!( policy.evaluate_with_timing(1, Some(1000), 2000), RecoveryAction::CreateWork );
// Outside window: window expired, resets to fresh count (GroupOnly) assert_eq!( policy.evaluate_with_timing(1, Some(1000), 20_000), RecoveryAction::GroupOnly ); }
#[test] fn test_diagnostic_classification_distinguishes_skipped_and_failure() { // 1. Skipped peer call let skipped_payload = serde_json::json!({ "type": "tool_result", "status": "skipped", "reason": "peer call cancelled", }); let (class, _, reason) = classify_diagnostic("scratchpad", &skipped_payload); assert_eq!(class, DiagnosticClassification::SkippedPeerCall); assert_eq!(reason.as_deref(), Some("peer call cancelled"));
// 2. Real tool error let err_payload = serde_json::json!({ "type": "tool_result", "status": "error", "error": "ValueError: bad argument", }); let (class, sig, _) = classify_diagnostic("scratchpad", &err_payload); assert_eq!(class, DiagnosticClassification::ExecutionFailure); assert_eq!(sig.as_deref(), Some("ValueError: bad argument")); }}