bobby_browser_client/
recovery.rs

1//! Workflow checkpoint and recovery decision types.
2
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5use sha2::{Digest, Sha256};
6
7use crate::{
8    AttemptId, CheckpointId, CommandClass, CommandId, CommandOutcome, Evidence, PageId, SessionId,
9    SkillDecision, SkillOutcome, WorkflowId,
10};
11
12pub const MAX_RECOVERY_RECEIPTS: usize = 64;
13pub const MAX_RECOVERY_RECEIPT_EVIDENCE: usize = 32;
14pub const MAX_RECOVERY_RECEIPT_BYTES: usize = 256 * 1024;
15
16/// Current recovery status for a workflow (`GET /v1/recovery/{id}`).
17#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
18#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
19#[serde(rename_all = "camelCase", deny_unknown_fields)]
20pub struct RecoveryStatus {
21    pub workflow_id: WorkflowId,
22    pub checkpoint: WorkflowCheckpoint,
23    pub receipts: Vec<RecoveryReceipt>,
24}
25
26#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
27#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
28#[serde(tag = "kind", rename_all = "camelCase")]
29pub enum CheckpointInvariant {
30    Url { value: String },
31    Title { value: String },
32    Text { selector: String, value: String },
33}
34
35#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
36#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
37#[serde(rename_all = "camelCase")]
38pub struct WorkflowCheckpoint {
39    pub schema_version: u16,
40    pub checkpoint_id: CheckpointId,
41    pub workflow_id: WorkflowId,
42    pub attempt_id: AttemptId,
43    pub session_id: SessionId,
44    pub page_id: PageId,
45    pub restart_url: String,
46    pub current_url: String,
47    pub cursor: Option<CommandId>,
48    #[serde(default)]
49    pub boundary_command_id: Option<CommandId>,
50    pub recovery_class: CommandClass,
51    pub invariants: Vec<CheckpointInvariant>,
52    pub replayable_inputs: Vec<String>,
53    pub evidence: Vec<Evidence>,
54    #[serde(default)]
55    pub recovery_history: Vec<RecoveryRecord>,
56    #[serde(default)]
57    pub recovery_receipts: Vec<RecoveryReceipt>,
58    pub created_at: DateTime<Utc>,
59}
60
61#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
62#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
63#[serde(rename_all = "camelCase")]
64pub struct RecoveryCommandIdentity {
65    pub command_id: CommandId,
66    pub workflow_id: WorkflowId,
67    pub attempt_id: AttemptId,
68    pub session_id: SessionId,
69    pub page_id: Option<PageId>,
70    pub command_class: CommandClass,
71    pub command_sha256: String,
72}
73
74impl RecoveryCommandIdentity {
75    #[allow(clippy::too_many_arguments)]
76    pub fn new(
77        command_id: CommandId,
78        workflow_id: WorkflowId,
79        attempt_id: AttemptId,
80        session_id: SessionId,
81        page_id: Option<PageId>,
82        command_class: CommandClass,
83        command_sha256: impl Into<String>,
84    ) -> Result<Self, String> {
85        let identity = Self {
86            command_id,
87            workflow_id,
88            attempt_id,
89            session_id,
90            page_id,
91            command_class,
92            command_sha256: command_sha256.into(),
93        };
94        identity.validate()?;
95        Ok(identity)
96    }
97
98    pub fn validate(&self) -> Result<(), String> {
99        validate_sha256(&self.command_sha256, "recovery command identity")
100    }
101}
102
103#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
104#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
105#[serde(rename_all = "camelCase")]
106pub enum RecoveryReceiptState {
107    PendingJournal,
108    Committed,
109    Unresolved,
110}
111
112#[derive(Debug, Clone, Serialize, PartialEq)]
113#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
114#[serde(rename_all = "camelCase")]
115pub struct RecoveryReceipt {
116    pub idempotency_key: CommandId,
117    pub identity: RecoveryCommandIdentity,
118    pub state: RecoveryReceiptState,
119    pub reservation_id: CommandId,
120    pub decision: SkillDecision,
121    pub command_outcome: CommandOutcome,
122    pub skill_outcome: SkillOutcome,
123    pub tactic_evidence: Vec<Evidence>,
124    pub outcome_sha256: String,
125    pub recorded_at: DateTime<Utc>,
126}
127
128#[derive(Deserialize)]
129#[serde(rename_all = "camelCase", deny_unknown_fields)]
130struct RecoveryReceiptWire {
131    idempotency_key: CommandId,
132    identity: RecoveryCommandIdentity,
133    state: RecoveryReceiptState,
134    reservation_id: CommandId,
135    decision: SkillDecision,
136    command_outcome: CommandOutcome,
137    skill_outcome: SkillOutcome,
138    tactic_evidence: Vec<Evidence>,
139    outcome_sha256: String,
140    recorded_at: DateTime<Utc>,
141}
142
143impl<'de> Deserialize<'de> for RecoveryReceipt {
144    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
145    where
146        D: serde::Deserializer<'de>,
147    {
148        let wire = RecoveryReceiptWire::deserialize(deserializer)?;
149        let receipt = Self {
150            idempotency_key: wire.idempotency_key,
151            identity: wire.identity,
152            state: wire.state,
153            reservation_id: wire.reservation_id,
154            decision: wire.decision,
155            command_outcome: wire.command_outcome,
156            skill_outcome: wire.skill_outcome,
157            tactic_evidence: wire.tactic_evidence,
158            outcome_sha256: wire.outcome_sha256,
159            recorded_at: wire.recorded_at,
160        };
161        receipt.validate().map_err(serde::de::Error::custom)?;
162        Ok(receipt)
163    }
164}
165
166impl RecoveryReceipt {
167    #[allow(clippy::too_many_arguments)]
168    pub fn new(
169        idempotency_key: CommandId,
170        identity: RecoveryCommandIdentity,
171        state: RecoveryReceiptState,
172        reservation_id: CommandId,
173        decision: SkillDecision,
174        command_outcome: CommandOutcome,
175        skill_outcome: SkillOutcome,
176        tactic_evidence: Vec<Evidence>,
177        recorded_at: DateTime<Utc>,
178    ) -> Result<Self, String> {
179        let outcome_sha256 =
180            Self::outcome_digest(&command_outcome, &skill_outcome, &tactic_evidence)?;
181        let receipt = Self {
182            idempotency_key,
183            identity,
184            state,
185            reservation_id,
186            decision,
187            command_outcome,
188            skill_outcome,
189            tactic_evidence,
190            outcome_sha256,
191            recorded_at,
192        };
193        receipt.validate()?;
194        Ok(receipt)
195    }
196
197    pub fn validate(&self) -> Result<(), String> {
198        self.identity.validate()?;
199        if self.tactic_evidence.len() > MAX_RECOVERY_RECEIPT_EVIDENCE {
200            return Err("recovery receipt tactic evidence exceeds its bound".into());
201        }
202        if self.idempotency_key != self.identity.command_id
203            || command_outcome_id(&self.command_outcome) != &self.identity.command_id
204        {
205            return Err("recovery receipt command identity does not match its outcome".into());
206        }
207        validate_sha256(&self.outcome_sha256, "recovery receipt outcome")?;
208        let actual = Self::outcome_digest(
209            &self.command_outcome,
210            &self.skill_outcome,
211            &self.tactic_evidence,
212        )?;
213        if actual != self.outcome_sha256 {
214            return Err("recovery receipt outcome digest changed".into());
215        }
216        serde_json::to_vec(&self.decision).map_err(|error| error.to_string())?;
217        let bytes = serde_json::to_vec(self).map_err(|error| error.to_string())?;
218        if bytes.len() > MAX_RECOVERY_RECEIPT_BYTES {
219            return Err("recovery receipt exceeds its byte bound".into());
220        }
221        Ok(())
222    }
223
224    pub fn outcome_digest(
225        command_outcome: &CommandOutcome,
226        skill_outcome: &SkillOutcome,
227        tactic_evidence: &[Evidence],
228    ) -> Result<String, String> {
229        let bytes = serde_json::to_vec(&(command_outcome, skill_outcome, tactic_evidence))
230            .map_err(|error| error.to_string())?;
231        Ok(hex::encode(Sha256::digest(bytes)))
232    }
233}
234
235fn command_outcome_id(outcome: &CommandOutcome) -> &CommandId {
236    match outcome {
237        CommandOutcome::Completed { command_id, .. }
238        | CommandOutcome::RetryableFailure { command_id, .. }
239        | CommandOutcome::NeedsReconciliation { command_id, .. }
240        | CommandOutcome::PolicyDenied { command_id, .. }
241        | CommandOutcome::ResourceExhausted { command_id, .. }
242        | CommandOutcome::Restarted { command_id, .. }
243        | CommandOutcome::Failed { command_id, .. } => command_id,
244    }
245}
246
247fn validate_sha256(value: &str, name: &str) -> Result<(), String> {
248    if value.len() != 64
249        || !value
250            .bytes()
251            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
252    {
253        return Err(format!("{name} requires a lowercase SHA-256 digest"));
254    }
255    Ok(())
256}
257
258#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
259#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
260#[serde(rename_all = "camelCase")]
261pub struct RecoveryRecord {
262    pub recorded_at: DateTime<Utc>,
263    pub decision: RecoveryDecision,
264}
265
266impl WorkflowCheckpoint {
267    pub const SCHEMA_VERSION: u16 = 1;
268}
269
270#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
271#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
272#[serde(rename_all = "camelCase")]
273pub struct RecoveryRequest {
274    pub workflow_id: WorkflowId,
275    pub checkpoint_id: CheckpointId,
276}
277
278#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
279#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
280#[serde(rename_all = "camelCase")]
281pub struct RestartLineage {
282    pub workflow_id: WorkflowId,
283    pub abandoned_attempt_id: AttemptId,
284    pub attempt_id: AttemptId,
285    pub reason: String,
286}
287
288#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
289#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
290#[serde(
291    tag = "status",
292    rename_all = "camelCase",
293    rename_all_fields = "camelCase"
294)]
295pub enum RecoveryDecision {
296    Resumed {
297        checkpoint_id: CheckpointId,
298        attempt_id: AttemptId,
299        evidence: Vec<Evidence>,
300    },
301    NeedsReconciliation {
302        checkpoint_id: CheckpointId,
303        attempt_id: AttemptId,
304        reason: String,
305        evidence: Vec<Evidence>,
306    },
307    Restarted {
308        checkpoint_id: CheckpointId,
309        lineage: RestartLineage,
310        #[serde(default)]
311        evidence: Vec<Evidence>,
312    },
313}
314
315/// `POST /v1/checkpoints` request body: the checkpoint plus the command ids
316/// whose evidence the runtime resolves from its own journal.
317#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
318#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
319#[serde(rename_all = "camelCase")]
320pub struct CheckpointRequest {
321    pub checkpoint: WorkflowCheckpoint,
322    #[serde(default, skip_serializing_if = "Vec::is_empty")]
323    pub evidence_refs: Vec<CommandId>,
324}
bobby_browser_client/recovery.rs

bobby_browser_client/
recovery.rs
