everruns_core/
durability.rs

1//! Neutral contracts for durable tool results and stream recovery.
2
3use crate::error::Result;
4use crate::typed_id::{MessageId, SessionId};
5use async_trait::async_trait;
6
7/// Result of a claim attempt on the per-tool-call idempotency store.
8#[derive(Debug)]
9pub enum ToolCallClaimResult {
10    /// First claim for this (turn_id, tool_call_id); caller should execute the tool.
11    /// `claim_token` must be passed to `settle_tool_call` to verify ownership.
12    Claimed { claim_token: uuid::Uuid },
13    /// A prior run already settled this call; replay the stored result.
14    AlreadySettled {
15        result_json: serde_json::Value,
16        args_fingerprint: String,
17    },
18    /// A prior run started but never settled. For `AtMostOnce` tools the
19    /// caller should NOT re-execute; for `Pure`/`Idempotent` tools the caller
20    /// may re-execute and then try to settle (the settle CAS will be a no-op if
21    /// a different claimer wins first).
22    AlreadyRunning { args_fingerprint: String },
23    /// A settled row exists but its `args_fingerprint` does not match the
24    /// current call — this is a determinism violation (workflow replay with
25    /// different inputs). The workflow should be failed loudly.
26    DeterminismViolation {
27        stored_fingerprint: String,
28        current_fingerprint: String,
29    },
30}
31
32/// Read-only status of a tool call in durable storage (EVE-533).
33#[derive(Debug, Clone)]
34pub enum DurableToolCallStatus {
35    /// Tool completed successfully or with an error; result is stored.
36    Settled { result_json: serde_json::Value },
37    /// Tool was settled with `interrupted` status; result may contain error details.
38    Interrupted {
39        result_json: Option<serde_json::Value>,
40    },
41    /// A claim exists but the tool never finished.
42    Running,
43}
44
45/// Durable per-tool-call idempotency store (EVE-530).
46///
47/// Implements the claim/settle CAS that prevents double-execution of
48/// `AtMostOnce` tools on worker reclaim/replay.
49#[async_trait]
50pub trait DurableToolResultStore: Send + Sync + 'static {
51    /// Atomically claim `(turn_id, tool_call_id)` before tool dispatch.
52    ///
53    /// - Inserts a `running` row if none exists → `Claimed`.
54    /// - Finds an existing `settled` row → `AlreadySettled`.
55    /// - Finds an existing `running` row → `AlreadyRunning`.
56    /// - Finds a `settled` row with a mismatched `args_fingerprint`
57    ///   (determinism violation) → `DeterminismViolation`.
58    async fn try_claim_tool_call(
59        &self,
60        turn_id: &str,
61        tool_call_id: &str,
62        tool_name: &str,
63        args_fingerprint: &str,
64    ) -> Result<ToolCallClaimResult>;
65
66    /// Settle a previously claimed tool call with its result.
67    ///
68    /// `claim_token` must match the token returned by `try_claim_tool_call`.
69    /// Returns `Ok(true)` if the row was updated, `Ok(false)` if the claim
70    /// token no longer matches (ownership lost — treat as a warning).
71    async fn settle_tool_call(
72        &self,
73        turn_id: &str,
74        tool_call_id: &str,
75        result_json: serde_json::Value,
76        status: &str,
77        claim_token: uuid::Uuid,
78    ) -> Result<bool>;
79
80    /// Read-only lookup of a tool call's current status in durable storage (EVE-533).
81    ///
82    /// Used by transcript repair to decide whether to replay a stored result or
83    /// synthesize an interrupted placeholder. Returns `None` if no row exists.
84    async fn get_tool_call_status(
85        &self,
86        turn_id: &str,
87        tool_call_id: &str,
88    ) -> Result<Option<DurableToolCallStatus>>;
89}
90
91// ============================================================================
92// StreamHeartbeater — per-stream liveness signal for Reason activity (EVE-531)
93// ============================================================================
94
95/// Progress snapshot carried in each stream heartbeat.
96#[derive(Debug, Clone)]
97pub struct StreamProgress {
98    /// Accumulated text + thinking length (characters) at the time of heartbeat.
99    pub accumulated_len: usize,
100    /// Wall-clock time of the most recent received token (Unix seconds).
101    pub last_delta_at: u64,
102}
103
104/// Heartbeater the Reason streaming loop calls on delta batches and a keepalive
105/// timer, signalling that the provider connection is alive.
106///
107/// Implementations bridge to the durable-execution layer (e.g. gRPC).
108#[async_trait]
109pub trait StreamHeartbeater: Send + Sync {
110    /// Signal stream liveness with current progress.
111    ///
112    /// Must be best-effort: errors must not propagate to the caller.
113    /// Cancel-safety is critical — if the worker dies the heartbeat stops
114    /// and the existing task-level reclaim takes over.
115    async fn heartbeat(&self, progress: StreamProgress);
116}
117
118// ============================================================================
119// PartialStreamStore — partial-stream recovery for Reason activity (EVE-532)
120// ============================================================================
121
122/// State of a partially-streamed assistant message detected in the event log.
123#[derive(Debug, Clone)]
124pub struct PartialStreamState {
125    /// Prepared Astra effort recovered from the matching stream-start event.
126    pub reasoning_state: Option<everruns_provider::reasoning_updates::ReasoningState>,
127    /// Stable public id from the latest `output.message.started` event.
128    pub message_id: MessageId,
129
130    /// Accumulated text from the last `output.message.delta` for the turn.
131    /// Empty when `output.message.started` was emitted but no delta arrived.
132    pub accumulated: String,
133}
134
135/// Consults the persisted event log to detect whether a `reason` activity
136/// was interrupted after `output.message.started` but before
137/// `output.message.completed` or `output.message.replaced`.
138///
139/// Used by `ReasonAtom` on re-entry to apply the ContinuePartial recovery
140/// policy (EVE-532): finalize the partial text without a second provider call,
141/// or restart clean if the partial is unusable.
142#[async_trait]
143pub trait PartialStreamStore: Send + Sync {
144    /// Return the partial-stream state for `(session_id, turn_id)` if an
145    /// in-flight assistant message exists (started but not completed).
146    async fn get_partial_stream(
147        &self,
148        session_id: SessionId,
149        turn_id: &str,
150    ) -> Result<Option<PartialStreamState>>;
151}
everruns_core/durability.rs

everruns_core/
durability.rs
