1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
//! Deterministic session replay dal WAL — dry-run senza side-effect.
//!
//! [`SessionReplay::replay_session`] rilegge le azioni ordinate per `seq`,
//! ricostruisce la timeline e riverifica formalmente la sequenza FSM:
//! una storia registrata che viola la tabella di transizione è corrotta
//! e il replay fallisce invece di certificarla.
use serde::{Deserialize, Serialize};
use crate::error::{KernelError, KernelResult};
use crate::fsm::{AgentEvent, StateMachine};
use crate::types::ActionStatus;
use crate::wal::Wal;
/// Un passo della timeline ricostruita.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReplayStep {
/// Sequenza originale.
pub step_seq: u64,
/// Tool eseguito.
pub tool_id: String,
/// Stato finale persistito (`COMMITTED` / `COMPENSATED` / `FAILED`).
pub status: String,
/// Transizioni FSM simulate, in ordine (es. `StartExecution(fs.write)`).
pub transitions: Vec<String>,
/// `true` se lo step fu compensato dal rollback.
pub compensated: bool,
}
/// Timeline completa di una saga, ricostruita senza rieseguire nulla.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ReplayTimeline {
/// Sessione riprodotta.
pub session_id: String,
/// Durata parete in secondi (da `created_at`/`updated_at` del WAL).
pub duration_secs: f64,
/// Numero di azioni registrate.
pub total_steps: usize,
/// Passi in ordine di esecuzione.
pub actions_timeline: Vec<ReplayStep>,
/// Stato FSM derivato (`Failed`, `Verifying`, ...).
pub final_status: String,
/// Quante compensazioni furono eseguite.
pub compensations_executed: usize,
}
/// Replay dry-run di una sessione.
pub struct SessionReplay;
impl SessionReplay {
/// Ricostruisce e valida formalmente la storia di una sessione.
///
/// - Legge le azioni per `seq ASC`, senza chiamare alcun tool.
/// - Simula gli eventi storici sulla FSM: `COMMITTED`/`COMPENSATED`
/// implicano `StartExecution` + `ToolSucceeded`; un `FAILED` senza
/// output implica `StartExecution` + `ToolFailed`; un `FAILED` *con*
/// output è un execute riuscito la cui compensazione fallì.
/// - Un `PENDING` residuo rende il replay incoerente (lanciare prima
/// `recover_dangling_sessions`).
pub fn replay_session(wal: &Wal, session_id: &uuid::Uuid) -> KernelResult<ReplayTimeline> {
let sid = session_id.to_string();
let actions = wal.get_actions(&sid)?;
if actions.is_empty() {
return Err(KernelError::SessionNotFound(sid));
}
let mut fsm = StateMachine::new();
let mut timeline = Vec::with_capacity(actions.len());
let mut compensations = 0usize;
// Storia: Idle → Planning (ogni saga registrata è stata pianificata).
Self::apply(&mut fsm, &sid, AgentEvent::StartPlanning)?;
for action in &actions {
let mut transitions = Vec::new();
let compensated = action.status == ActionStatus::Compensated;
if compensated {
compensations += 1;
}
match action.status {
ActionStatus::Pending | ActionStatus::PendingApproval => {
return Err(KernelError::InvalidStatus(format!(
"replay: dangling {} at seq {} (run recovery first)",
action.status, action.step_seq
)));
}
ActionStatus::Committed | ActionStatus::Compensated => {
transitions.push(Self::apply(
&mut fsm,
&sid,
AgentEvent::StartExecution {
tool_name: action.tool_id.clone(),
},
)?);
transitions.push(Self::apply(&mut fsm, &sid, AgentEvent::ToolSucceeded)?);
}
ActionStatus::Failed if action.output.is_some() => {
// Execute OK, compensate KO: il forward fu un successo.
transitions.push(Self::apply(
&mut fsm,
&sid,
AgentEvent::StartExecution {
tool_name: action.tool_id.clone(),
},
)?);
transitions.push(Self::apply(&mut fsm, &sid, AgentEvent::ToolSucceeded)?);
}
ActionStatus::Failed => {
transitions.push(Self::apply(
&mut fsm,
&sid,
AgentEvent::StartExecution {
tool_name: action.tool_id.clone(),
},
)?);
transitions.push(Self::apply(&mut fsm, &sid, AgentEvent::ToolFailed)?);
}
}
timeline.push(ReplayStep {
step_seq: action.step_seq,
tool_id: action.tool_id.clone(),
status: action.status.as_str().to_owned(),
transitions,
compensated,
});
}
// Chiusura: se la simulazione è in Compensating, la storia include
// il completamento del rollback (→ Failed).
if *fsm.state() == crate::fsm::AgentState::Compensating {
Self::apply(&mut fsm, &sid, AgentEvent::CompensationCompleted)?;
}
Ok(ReplayTimeline {
session_id: sid.clone(),
duration_secs: wal.session_duration_secs(&sid)?.unwrap_or(0.0),
total_steps: actions.len(),
actions_timeline: timeline,
final_status: fsm.state().to_string(),
compensations_executed: compensations,
})
}
/// Applica un evento e ritorna la sua forma testuale per la timeline.
fn apply(fsm: &mut StateMachine, sid: &str, event: AgentEvent) -> KernelResult<String> {
let label = event.to_string();
fsm.transition(event).map_err(|e| {
KernelError::InvalidStatus(format!("replay incoherent for '{sid}': {e}"))
})?;
Ok(label)
}
}