//! 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 std::collections::HashMap;
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 {
Reject,
GroupOnly,
CreateWork,
}
/// Explicit decision distinguishing burst coalescing from chronic recurrence.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum RecurrenceDecision {
/// Discarded or rejected by policy.
Reject,
/// Coalesced within burst window without reaching recurrence threshold.
BurstCoalesced,
/// Rapid recurrence reaching threshold within burst window.
BurstThresholdReached,
/// Slow / chronic recurrence under the same component revision beyond burst window.
ChronicRecurrence,
}
impl RecurrenceDecision {
pub fn is_chronic(&self) -> bool {
matches!(self, Self::ChronicRecurrence)
}
pub fn action(&self) -> RecoveryAction {
match self {
Self::Reject => RecoveryAction::Reject,
Self::BurstCoalesced => RecoveryAction::GroupOnly,
Self::BurstThresholdReached | Self::ChronicRecurrence => RecoveryAction::CreateWork,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RecoveryObservationPolicy {
pub min_recurrence: u64,
pub grouping_window_ms: i64,
pub max_per_turn: usize,
}
impl Default for RecoveryObservationPolicy {
fn default() -> Self {
Self {
min_recurrence: 2,
grouping_window_ms: 60_000,
max_per_turn: 5,
}
}
}
impl RecoveryObservationPolicy {
pub fn evaluate(&self, current_count: i64) -> RecoveryAction {
self.evaluate_with_timing(current_count, None, 0)
}
/// Explicitly decides whether an occurrence is burst-coalesced, a burst threshold hit,
/// or a chronic recurrence beyond the burst window.
pub fn decide_recurrence(
&self,
current_count: i64,
last_seen_ms: Option<i64>,
now_ms: i64,
) -> RecurrenceDecision {
if self.min_recurrence == 0 {
return RecurrenceDecision::Reject;
}
let total_count = current_count + 1;
if let Some(last) = last_seen_ms {
let is_burst = self.grouping_window_ms > 0
&& now_ms >= last
&& (now_ms - last) <= self.grouping_window_ms;
if is_burst {
if total_count >= self.min_recurrence as i64 {
RecurrenceDecision::BurstThresholdReached
} else {
RecurrenceDecision::BurstCoalesced
}
} else {
// Beyond the burst coalescing window (e.g. > 60s):
// Evaluated as chronic recurrence under the same component revision.
if total_count >= self.min_recurrence as i64 {
RecurrenceDecision::ChronicRecurrence
} else {
RecurrenceDecision::BurstCoalesced
}
}
} else {
// First occurrence
if total_count >= self.min_recurrence as i64 {
RecurrenceDecision::BurstThresholdReached
} else {
RecurrenceDecision::BurstCoalesced
}
}
}
pub fn evaluate_with_timing(
&self,
current_count: i64,
last_seen_ms: Option<i64>,
now_ms: i64,
) -> RecoveryAction {
self.decide_recurrence(current_count, last_seen_ms, now_ms)
.action()
}
}
/// Categorizes raw diagnostic output so policy distinguishes execution
/// 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,
) -> (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() {
let first_line = err
.lines()
.map(str::trim)
.find(|l| !l.is_empty())
.unwrap_or("ExecutionError");
truncate_signature(first_line, 96)
} else {
sig
};
return (
DiagnosticClassification::ExecutionFailure,
Some(effective_sig),
Some(err),
);
}
// 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 is_warning_diagnostic(text) {
return (
DiagnosticClassification::StderrWarning,
None,
Some(text.to_string()),
);
} else {
return (
DiagnosticClassification::Uncertain,
None,
Some(text.to_string()),
);
}
}
}
}
(DiagnosticClassification::NormalOutput, None, None)
}
"scratchpad" | "tool.result" => {
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() {
let first_line = err
.lines()
.map(str::trim)
.find(|l| !l.is_empty())
.unwrap_or("ToolError");
truncate_signature(first_line, 96)
} else {
sig
};
return (
DiagnosticClassification::ExecutionFailure,
Some(effective_sig),
Some(err),
);
}
}
(DiagnosticClassification::NormalOutput, None, None)
}
_ => (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();
}
if text.contains("Traceback (most recent call last):") {
if let Some(last_line) = text.lines().rev().map(str::trim).find(|l| !l.is_empty()) {
return truncate_signature(last_line, 96);
}
}
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")
|| 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);
}
}
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()
}
#[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.
///
/// 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,
fallback_revision: &str,
policy: &RecoveryObservationPolicy,
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, 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;
}
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"
};
// 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,
&effective_rev,
&sig,
&effective_ws,
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::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: advance cursor past clean events
let _ = store.advance_observer_cursor(&partition, event.id).await;
}
}
}
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)
}