finalize run

⚒️ Generated with [Fabro](https://fabro.sh)
This commit is contained in:
Fabro 2026-07-28 23:32:47 +00:00
parent 85b5e079f4
commit d05d863b18
5 changed files with 1551 additions and 59 deletions

475
run.json

File diff suppressed because one or more lines are too long

View file

@ -0,0 +1,758 @@
diff --git a/lib/components/fabro-store/src/error.rs b/lib/components/fabro-store/src/error.rs
index 43c26b44a..e5af1dbc9 100644
--- a/lib/components/fabro-store/src/error.rs
+++ b/lib/components/fabro-store/src/error.rs
@@ -14,6 +14,11 @@ pub enum Error {
Io(#[from] std::io::Error),
#[error("Invalid event payload: {0}")]
InvalidEvent(String),
+ #[error("Event rejected by run state: {source}")]
+ EventRejected {
+ #[source]
+ source: Box<Self>,
+ },
#[error("Run not found: {0}")]
RunNotFound(String),
#[error("Run already exists: {0}")]
diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs
index d80936dc7..57d1a126d 100644
--- a/lib/components/fabro-store/src/run_summary_store.rs
+++ b/lib/components/fabro-store/src/run_summary_store.rs
@@ -147,6 +147,11 @@ impl RunSummaryStore {
Self { pool }
}
+ #[cfg(test)]
+ pub(crate) fn test_pool(&self) -> &SqlitePool {
+ &self.pool
+ }
+
pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> {
let record = ProjectedRunSummary::from_entry(entry);
let mut connection = self.pool.acquire().await?;
diff --git a/lib/components/fabro-store/src/slate/projection_cache.rs b/lib/components/fabro-store/src/slate/projection_cache.rs
index 7cec9a204..79ff0c03b 100644
--- a/lib/components/fabro-store/src/slate/projection_cache.rs
+++ b/lib/components/fabro-store/src/slate/projection_cache.rs
@@ -5,8 +5,8 @@ use chrono::{DateTime, Utc};
use fabro_types::{Run, RunId, RunProjection};
use tokio::sync::Mutex;
-use crate::run_state::{RunProjectionReducer, build_summary};
-use crate::{Error, EventEnvelope, ListRunsQuery, Result};
+use crate::ListRunsQuery;
+use crate::run_state::build_summary;
#[derive(Debug, Clone)]
pub struct CachedRunProjection {
@@ -85,26 +85,6 @@ impl RunProjectionCacheState {
}
}
- fn update_parent_index(
- &mut self,
- run_id: RunId,
- previous_parent_id: Option<RunId>,
- parent_id: Option<RunId>,
- ) {
- if previous_parent_id == parent_id {
- return;
- }
- if let Some(previous_parent_id) = previous_parent_id {
- self.remove_parent_link(&previous_parent_id, &run_id);
- }
- if let Some(parent_id) = parent_id {
- self.children_by_parent
- .entry(parent_id)
- .or_default()
- .insert(run_id);
- }
- }
-
fn count_children(&self, run_id: &RunId) -> u64 {
self.children_by_parent
.get(run_id)
@@ -222,51 +202,6 @@ impl RunProjectionCache {
Some(entry.summary)
}
- pub(crate) async fn apply_event(
- &self,
- run_id: &RunId,
- event: &EventEnvelope,
- ) -> Result<CachedRunProjection> {
- let mut state = self.state.lock().await;
- let Some(entry) = state.entries.get(run_id) else {
- if event.seq == 1 {
- let projection = RunProjection::apply_events(std::slice::from_ref(event))?;
- let entry = CachedRunProjection::from_projection(*run_id, projection, event.seq);
- state.insert(entry.clone());
- return Ok(entry);
- }
- return Err(Error::InvalidEvent(format!(
- "projection cache cannot initialize run {run_id} from event seq {}",
- event.seq
- )));
- };
-
- let last_seq = entry.last_seq;
- if event.seq <= last_seq {
- return Ok(entry.clone());
- }
- if event.seq != last_seq.saturating_add(1) {
- return Err(Error::Other(format!(
- "projection cache sequence gap for run {run_id}: last_seq={}, event_seq={}",
- last_seq, event.seq
- )));
- }
-
- let (previous_parent_id, parent_id, entry) = {
- let entry = state
- .entries
- .get_mut(run_id)
- .expect("entry was read from the same locked map");
- let previous_parent_id = entry.summary.parent_id;
- Arc::make_mut(&mut entry.projection).apply_event(event)?;
- entry.summary = build_summary(&entry.projection, run_id);
- entry.last_seq = event.seq;
- (previous_parent_id, entry.summary.parent_id, entry.clone())
- };
- state.update_parent_index(*run_id, previous_parent_id, parent_id);
- Ok(entry)
- }
-
pub(crate) async fn remove(&self, run_id: &RunId) {
self.state.lock().await.remove(run_id);
}
diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs
index 9058627a6..c71fe7078 100644
--- a/lib/components/fabro-store/src/slate/run_store.rs
+++ b/lib/components/fabro-store/src/slate/run_store.rs
@@ -9,7 +9,7 @@ use futures::Stream;
use slatedb::{Db, DbIterator, DbRead};
use tokio::sync::{Mutex, broadcast, mpsc};
use tokio_stream::wrappers::UnboundedReceiverStream;
-use tracing::{error, warn};
+use tracing::warn;
use super::blob_store::BlobStore;
use super::projection_cache::{CachedRunProjection, RunProjectionCache};
@@ -192,6 +192,15 @@ impl RunDatabase {
}
async fn projected_state_locked(&self) -> Result<Arc<RunProjection>> {
+ self.projected_state_option_locked().await?.ok_or_else(|| {
+ Error::InvalidEvent(format!(
+ "run {} has no run.created event",
+ self.inner.run_id
+ ))
+ })
+ }
+
+ async fn projected_state_option_locked(&self) -> Result<Option<Arc<RunProjection>>> {
let next_seq = {
let cache = self.inner.projection_cache.lock().await;
cache.last_seq.saturating_add(1)
@@ -202,25 +211,14 @@ impl RunDatabase {
apply_cached_projection_event(&mut cache.state, event)?;
cache.last_seq = event.seq;
}
- cache.state.clone().ok_or_else(|| {
- Error::InvalidEvent(format!(
- "run {} has no run.created event",
- self.inner.run_id
- ))
- })
+ Ok(cache.state.clone())
}
- async fn cache_event(&self, event: &EventEnvelope) -> Result<()> {
+ async fn cache_event(&self, event: &EventEnvelope, projection: Arc<RunProjection>) {
{
let mut projection_cache = self.inner.projection_cache.lock().await;
- if projection_cache.state.is_none() && event.seq > 1 {
- drop(projection_cache);
- self.rebuild_local_projection_cache_through(event.seq)
- .await?;
- } else {
- apply_cached_projection_event(&mut projection_cache.state, event)?;
- projection_cache.last_seq = event.seq;
- }
+ projection_cache.state = Some(projection);
+ projection_cache.last_seq = event.seq;
}
let mut recent_events = self.inner.recent_events.lock().await;
recent_events.push_back(event.clone());
@@ -228,29 +226,6 @@ impl RunDatabase {
recent_events.pop_front();
}
let _ = self.inner.event_tx.send(event.clone());
- Ok(())
- }
-
- async fn rebuild_local_projection_cache_through(&self, seq: u32) -> Result<()> {
- let events = list_events_from(&self.inner.db, &self.inner.run_id, 1).await?;
- let Some(last_seq) = events.last().map(|event| event.seq) else {
- return Err(Error::InvalidEvent(format!(
- "run {} has no events while rebuilding projection cache",
- self.inner.run_id
- )));
- };
- if last_seq < seq {
- return Err(Error::InvalidEvent(format!(
- "run {} projection cache rebuild stopped at seq {last_seq}, before appended seq {seq}",
- self.inner.run_id
- )));
- }
-
- let state = RunProjection::apply_events(&events)?;
- let mut projection_cache = self.inner.projection_cache.lock().await;
- projection_cache.state = Some(Arc::new(state));
- projection_cache.last_seq = last_seq;
- Ok(())
}
async fn cached_events_from(&self, start_seq: u32, limit: usize) -> Option<Vec<EventEnvelope>> {
@@ -270,12 +245,40 @@ impl RunDatabase {
}
impl RunDatabase {
+ /// Appends `payload` to the run's authoritative event log.
+ ///
+ /// A payload rejected by the current run state is not written. Any other
+ /// error also means the event was not committed and is safe to retry. Once
+ /// the authoritative write succeeds, derived cache and SQLite updates are
+ /// best-effort and this method returns success even when one of them fails.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`Error::EventRejected`] when the run reducer rejects the
+ /// candidate event. Other errors report validation, sequence allocation,
+ /// projection loading, serialization, read-only access, or an
+ /// authoritative storage failure. No returned error represents a committed
+ /// append.
pub async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
Ok(self.append_event_envelope(payload).await?.seq)
}
/// Atomically appends `payload` when `predicate` matches the latest run
/// projection.
+ ///
+ /// `Ok(None)` means the predicate rejected the append and nothing was
+ /// written. A state-machine rejection or any other error also leaves the
+ /// event uncommitted and safe to retry. Once the authoritative write
+ /// succeeds, derived cache and SQLite updates are best-effort and this
+ /// method returns the committed sequence.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`Error::EventRejected`] when the predicate accepts but the run
+ /// reducer rejects the candidate event. Other errors report validation,
+ /// sequence allocation, projection loading, serialization, read-only
+ /// access, or an authoritative storage failure. No returned error
+ /// represents a committed append.
pub async fn append_event_if(
&self,
payload: &EventPayload,
@@ -290,90 +293,91 @@ impl RunDatabase {
if !predicate(&projection) {
return Ok(None);
}
- Ok(Some(self.append_event_envelope_locked(payload).await?.seq))
+ Ok(Some(
+ self.append_event_envelope_locked(payload, Some(projection))
+ .await?
+ .seq,
+ ))
}
+ /// Appends `payload` and returns its committed envelope.
+ ///
+ /// A payload rejected by the current run state is not written. Any other
+ /// error also means the event was not committed and is safe to retry. Once
+ /// the authoritative write succeeds, derived cache and SQLite updates are
+ /// best-effort and this method returns success even when one of them fails.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`Error::EventRejected`] when the run reducer rejects the
+ /// candidate event. Other errors report validation, sequence allocation,
+ /// projection loading, serialization, read-only access, or an
+ /// authoritative storage failure. No returned error represents a committed
+ /// append.
pub async fn append_event_envelope(&self, payload: &EventPayload) -> Result<EventEnvelope> {
if self.read_only {
return Err(Error::ReadOnly);
}
payload.validate(&self.inner.run_id)?;
let _state_guard = self.inner.state_lock.lock().await;
- self.append_event_envelope_locked(payload).await
+ self.append_event_envelope_locked(payload, None).await
}
- async fn append_event_envelope_locked(&self, payload: &EventPayload) -> Result<EventEnvelope> {
+ async fn append_event_envelope_locked(
+ &self,
+ payload: &EventPayload,
+ current_projection: Option<Arc<RunProjection>>,
+ ) -> Result<EventEnvelope> {
let event_seq = self.inner.event_seq.as_ref().ok_or(Error::ReadOnly)?;
- let seq = allocate_event_seq(event_seq)?;
+ let seq = event_seq.load(Ordering::SeqCst);
+ if seq > keys::MAX_EVENT_SEQ {
+ return Err(Error::EventSequenceExhausted {
+ max_seq: keys::MAX_EVENT_SEQ,
+ });
+ }
let event = EventEnvelope {
seq,
event: RunEvent::try_from(payload)?,
};
+ let projection = match current_projection {
+ Some(projection) => validate_projected_event(&projection, &event)?,
+ None => match self.projected_state_option_locked().await? {
+ Some(projection) => validate_projected_event(&projection, &event)?,
+ None => validate_first_event(&event)?,
+ },
+ };
+ let encoded = serde_json::to_vec(payload)?;
+ allocate_event_seq(event_seq)?;
self.inner
.db
.put(
keys::run_event_key(&self.inner.run_id, seq, Utc::now().timestamp_millis()),
- serde_json::to_vec(payload)?,
+ encoded,
)
.await?;
- self.cache_event(&event).await?;
- // Box::pin keeps append_event_envelope's future small enough for the
- // clippy::large_futures budget of its many callers.
- Box::pin(self.update_summary_projection_after_append(&event)).await?;
- Ok(event)
- }
- async fn update_summary_projection_after_append(&self, event: &EventEnvelope) -> Result<()> {
- let cached = match self
- .inner
+ let cached = CachedRunProjection::from_projection(
+ self.inner.run_id,
+ Arc::unwrap_or_clone(projection),
+ event.seq,
+ );
+ self.inner
.shared_projection_cache
- .apply_event(&self.inner.run_id, event)
- .await
- {
- Ok(entry) => entry,
- Err(err) => {
- match Self::build_cached_projection(&self.inner.db, &self.inner.run_id).await {
- Ok(Some(entry)) => {
- self.inner
- .shared_projection_cache
- .replace(entry.clone())
- .await;
- entry
- }
- rebuild => {
- self.inner
- .shared_projection_cache
- .remove(&self.inner.run_id)
- .await;
- if let Err(rebuild_err) = rebuild {
- warn!(
- run_id = %self.inner.run_id,
- error = %rebuild_err,
- "Failed to rebuild run projection cache after append"
- );
- }
- warn!(
- run_id = %self.inner.run_id,
- error = %err,
- "Failed to update run projection cache after append"
- );
- return Err(err);
- }
- }
- }
- };
+ .replace(cached.clone())
+ .await;
+ self.cache_event(&event, Arc::clone(&cached.projection))
+ .await;
if let Some(store) = self.inner.run_summary_store.get() {
if let Err(err) = store.upsert_projection(&cached).await {
- error!(
+ warn!(
run_id = %self.inner.run_id,
source_last_seq = cached.last_seq,
error = %err,
- "Failed to update SQLite run summary after append"
+ "Failed to update SQLite run summary after committed append"
);
- return Err(err);
}
}
- Ok(())
+ Ok(event)
}
pub async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
@@ -599,6 +603,27 @@ fn allocate_event_seq(event_seq: &AtomicU32) -> Result<u32> {
})
}
+fn validate_first_event(event: &EventEnvelope) -> Result<Arc<RunProjection>> {
+ RunProjection::apply_events(std::slice::from_ref(event))
+ .map(Arc::new)
+ .map_err(event_rejected)
+}
+
+fn validate_projected_event(
+ current: &RunProjection,
+ event: &EventEnvelope,
+) -> Result<Arc<RunProjection>> {
+ let mut projection = current.clone();
+ projection.apply_event(event).map_err(event_rejected)?;
+ Ok(Arc::new(projection))
+}
+
+fn event_rejected(source: Error) -> Error {
+ Error::EventRejected {
+ source: Box::new(source),
+ }
+}
+
fn apply_cached_projection_event(
state: &mut Option<Arc<RunProjection>>,
event: &EventEnvelope,
@@ -909,11 +934,15 @@ mod tests {
use std::sync::atomic::Ordering;
use std::time::Duration;
- use fabro_types::{Graph, RunId, SessionId, StageId, WorkflowSettings, test_support};
+ use chrono::Utc;
+ use fabro_types::{
+ Graph, RunId, RunStatus, SessionId, StageId, WorkflowSettings, test_support,
+ };
+ use fabro_util::error;
use object_store::memory::InMemory;
use serde_json::json;
- use crate::{Database, Error, EventPayload, keys};
+ use crate::{Database, Error, EventPayload, keys, test_util};
#[tokio::test]
async fn list_blobs_reads_global_cas_namespace() {
@@ -973,6 +1002,79 @@ mod tests {
.unwrap()
}
+ fn lifecycle_payload(
+ run_id: &RunId,
+ id: &str,
+ ts: &str,
+ event: &str,
+ properties: &serde_json::Value,
+ ) -> EventPayload {
+ EventPayload::new(
+ json!({
+ "id": id,
+ "ts": ts,
+ "run_id": run_id.to_string(),
+ "event": event,
+ "properties": properties,
+ }),
+ run_id,
+ )
+ .unwrap()
+ }
+
+ fn run_failed_payload(run_id: &RunId) -> EventPayload {
+ lifecycle_payload(
+ run_id,
+ "evt-failed",
+ "2026-04-09T12:00:04Z",
+ "run.failed",
+ &json!({
+ "failure": {
+ "reason": "workflow_error",
+ "detail": {
+ "message": "workflow failed",
+ "category": "deterministic",
+ },
+ },
+ "timing": {
+ "wall_time_ms": 1,
+ "inference_time_ms": 0,
+ "tool_time_ms": 0,
+ "active_time_ms": 0,
+ },
+ }),
+ )
+ }
+
+ async fn drive_to_runnable(run: &super::RunDatabase) {
+ let run_id = run.run_id();
+ for payload in [
+ lifecycle_payload(
+ &run_id,
+ "evt-submitted",
+ "2026-04-09T12:00:00Z",
+ "run.submitted",
+ &json!({ "definition_blob": null }),
+ ),
+ lifecycle_payload(
+ &run_id,
+ "evt-start-requested",
+ "2026-04-09T12:00:01Z",
+ "run.start_requested",
+ &json!({ "resume": false }),
+ ),
+ lifecycle_payload(
+ &run_id,
+ "evt-runnable",
+ "2026-04-09T12:00:02Z",
+ "run.runnable",
+ &json!({ "source": "start_requested" }),
+ ),
+ ] {
+ run.append_event(&payload).await.unwrap();
+ }
+ }
+
fn stage_prompt_payload_for_stage(
run_id: &RunId,
idx: u32,
@@ -1015,6 +1117,214 @@ mod tests {
run
}
+ #[tokio::test]
+ async fn invalid_transition_is_rejected_without_changing_durable_or_cached_state() {
+ let run = fresh_run().await;
+ let run_id = run.run_id();
+ drive_to_runnable(&run).await;
+ let events_before = run.list_events().await.unwrap();
+
+ let err = run
+ .append_event(&run_failed_payload(&run_id))
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ &err,
+ Error::EventRejected { source }
+ if matches!(source.as_ref(), Error::InvalidTransition(_))
+ ));
+ let error_chain = error::collect_chain(&err);
+ assert!(error_chain.len() >= 2);
+ assert!(
+ error_chain
+ .iter()
+ .skip(1)
+ .any(|cause| cause.contains("invalid status transition"))
+ );
+ assert_eq!(run.list_events().await.unwrap(), events_before);
+ assert!(
+ run.get_event(events_before.last().unwrap().seq + 1)
+ .await
+ .unwrap()
+ .is_none()
+ );
+ assert_eq!(run.state().await.unwrap().status, RunStatus::Runnable);
+ let cached = run
+ .inner
+ .shared_projection_cache
+ .get(&run_id)
+ .await
+ .unwrap();
+ assert_eq!(cached.last_seq, events_before.last().unwrap().seq);
+ assert_eq!(cached.projection.status, RunStatus::Runnable);
+
+ let next_seq = run
+ .append_event(&stage_prompt_payload(&run_id, 1, Some("alpha")))
+ .await
+ .unwrap();
+ assert_eq!(next_seq, events_before.last().unwrap().seq + 1);
+ }
+
+ #[tokio::test]
+ async fn rejected_transition_leaves_summary_available_after_reconcile() {
+ let object_store = Arc::new(InMemory::new());
+ let database = Database::new(object_store, "", Duration::from_millis(1), None);
+ let (_directory, summary_store) = test_util::sqlite_summary_store().await;
+ let summary_store = Arc::new(summary_store);
+ database.attach_run_summary_store(Arc::clone(&summary_store));
+ let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
+ let run = database.create_run(&run_id).await.unwrap();
+ run.append_event(&run_created_payload(&run_id))
+ .await
+ .unwrap();
+ drive_to_runnable(&run).await;
+
+ let err = run
+ .append_event(&run_failed_payload(&run_id))
+ .await
+ .unwrap_err();
+ assert!(matches!(err, Error::EventRejected { .. }));
+ let cached = run
+ .inner
+ .shared_projection_cache
+ .get(&run_id)
+ .await
+ .unwrap();
+ assert_eq!(cached.last_seq, 4);
+ summary_store.reconcile(&[cached]).await.unwrap();
+
+ assert!(
+ summary_store
+ .get(&run_id, Utc::now())
+ .await
+ .unwrap()
+ .is_some()
+ );
+ }
+
+ #[tokio::test]
+ async fn committed_append_succeeds_when_summary_update_fails_and_is_repairable() {
+ let object_store = Arc::new(InMemory::new());
+ let database = Database::new(object_store, "", Duration::from_millis(1), None);
+ let (_directory, summary_store) = test_util::sqlite_summary_store().await;
+ let summary_store = Arc::new(summary_store);
+ database.attach_run_summary_store(Arc::clone(&summary_store));
+ let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
+ let run = database.create_run(&run_id).await.unwrap();
+ run.append_event(&run_created_payload(&run_id))
+ .await
+ .unwrap();
+ sqlx::query(
+ "CREATE TRIGGER reject_run_summary_update BEFORE UPDATE ON runs BEGIN SELECT \
+ RAISE(ABORT, 'forced run summary update failure'); END",
+ )
+ .execute(summary_store.test_pool())
+ .await
+ .unwrap();
+ let first_update = lifecycle_payload(
+ &run_id,
+ "evt-title-1",
+ "2026-04-09T12:00:01Z",
+ "run.title.updated",
+ &json!({ "title": "first update" }),
+ );
+
+ let seq = run.append_event(&first_update).await.unwrap();
+
+ assert_eq!(run.get_event(seq).await.unwrap().unwrap().seq, seq);
+ assert_ne!(
+ summary_store
+ .get(&run_id, Utc::now())
+ .await
+ .unwrap()
+ .unwrap()
+ .title,
+ "first update"
+ );
+ sqlx::query("DROP TRIGGER reject_run_summary_update")
+ .execute(summary_store.test_pool())
+ .await
+ .unwrap();
+ let repaired_seq = run
+ .append_event(&lifecycle_payload(
+ &run_id,
+ "evt-title-2",
+ "2026-04-09T12:00:02Z",
+ "run.title.updated",
+ &json!({ "title": "repaired" }),
+ ))
+ .await
+ .unwrap();
+ let repaired = summary_store
+ .get(&run_id, Utc::now())
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(repaired.title, "repaired");
+ assert_eq!(repaired_seq, seq + 1);
+ }
+
+ #[tokio::test]
+ async fn first_event_must_initialize_a_projection_before_any_write() {
+ let object_store = Arc::new(InMemory::new());
+ let database = Database::new(object_store, "", Duration::from_millis(1), None);
+ let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
+ let run = database.create_run(&run_id).await.unwrap();
+
+ let err = run
+ .append_event(&stage_prompt_payload(&run_id, 1, Some("alpha")))
+ .await
+ .unwrap_err();
+
+ assert!(matches!(
+ &err,
+ Error::EventRejected { source }
+ if matches!(source.as_ref(), Error::InvalidEvent(_))
+ ));
+ assert!(run.list_events().await.unwrap().is_empty());
+ assert!(
+ run.inner
+ .shared_projection_cache
+ .projection_snapshot(&run_id)
+ .await
+ .is_none()
+ );
+ assert_eq!(
+ run.inner.event_seq.as_ref().unwrap().load(Ordering::SeqCst),
+ 1
+ );
+
+ assert_eq!(
+ run.append_event(&run_created_payload(&run_id))
+ .await
+ .unwrap(),
+ 1
+ );
+ assert_eq!(run.list_events().await.unwrap().len(), 1);
+ }
+
+ #[tokio::test]
+ async fn append_event_if_false_writes_nothing() {
+ let run = fresh_run().await;
+ let run_id = run.run_id();
+ let events_before = run.list_events().await.unwrap();
+
+ let result = run
+ .append_event_if(&stage_prompt_payload(&run_id, 1, Some("alpha")), |_| false)
+ .await
+ .unwrap();
+
+ assert_eq!(result, None);
+ assert_eq!(run.list_events().await.unwrap(), events_before);
+ assert!(
+ run.get_event(events_before.last().unwrap().seq + 1)
+ .await
+ .unwrap()
+ .is_none()
+ );
+ }
+
#[tokio::test]
async fn list_events_from_with_limit_does_not_read_past_limit_plus_one() {
let run = fresh_run().await;
@@ -1304,6 +1614,7 @@ mod tests {
.unwrap();
assert_eq!(seq, keys::MAX_EVENT_SEQ);
+ let events_before_error = run.list_events().await.unwrap();
let err = run
.append_event(&stage_prompt_payload(&run_id, 2, Some("beta")))
.await
@@ -1313,6 +1624,7 @@ mod tests {
Error::EventSequenceExhausted { max_seq }
if max_seq == keys::MAX_EVENT_SEQ
));
+ assert_eq!(run.list_events().await.unwrap(), events_before_error);
assert!(
run.get_event(keys::MAX_EVENT_SEQ + 1)
.await

View file

@ -0,0 +1,6 @@
{
"outcome": "succeeded",
"notes": "Stage completed: implement",
"failure_reason": null,
"timestamp": "2026-07-28T23:27:54.841342282Z"
}

View file

@ -0,0 +1,365 @@
Goal: # PR 1 — Make run-event appends validate before write and report commit status unambiguously
**Self-contained implementation plan.** Everything needed to implement this
is in this file plus the repository.
**Precondition:** none — this is foundational work with no dependency on
other in-flight changes. Re-verify the "Verified current state" section
against HEAD before starting; if the append path in
`lib/components/fabro-store/src/slate/run_store.rs` has been materially
restructured since the pinned commit, stop and state that in the PR
description instead of adapting blindly.
> **Token notation.** Interpolation tokens are written in this file without
> their enclosing double curly braces, so the file is safe to pass directly
> as a workflow goal (the goal templater would otherwise try to expand them).
> Read `secrets.NAME`, `env.NAME`, `vars.NAME` as the double-curly-brace
> token form used in the codebase, and write the real double-brace syntax in
> the code, tests, and docs you produce.
## Context and goal
Fabro's run state is event-sourced: each run has an append-only event log in
a shared SlateDB store (`fabro-store`), a reduced in-memory projection
(`RunProjection`), and a derived SQLite summary row used by all listing
endpoints. Run status transitions are enforced by a state machine
(`RunStatus::can_transition_to` / `transition_to` in
`lib/foundation/fabro-types/src/status.rs`) — for example, a run whose
durable status is `Runnable` may legally move to `Failed` only with reason
`Cancelled`; a `Failed { WorkflowError }` from `Runnable` is an invalid
transition and the reducer hard-errors on it.
The append path has two defects, and this PR fixes both at the store layer:
**Defect 1 — poison events.** `append_event_envelope_locked` writes the
event bytes to SlateDB *before* any reduction happens. If the event turns
out to be transition-invalid, the caller gets an error — but the invalid
event is already durably in the log. From then on the run's projection can
never be rebuilt: replay hits the same invalid transition every time. The
user-visible consequence is severe: at startup, projection warmup skips the
unreadable run, and the SQLite reconciler then *deletes its summary row*
because it is absent from the authoritative entries — the run disappears
from every listing, and get/cancel return 404. This is a real shipped bug:
several server failure helpers attempt exactly such illegal appends today
(e.g. a worker-launch failure helper appends `Failed { LaunchFailed }`
while the durable status is still `Runnable`). Those call sites are being
fixed in separate planned work — this PR's job is to make the store refuse
to write the poison event in the first place.
**Defect 2 — ambiguous append errors.** After the SlateDB put succeeds, the
append still does derived work: applying the event to the shared projection
cache and upserting the SQLite summary row. Failures in either currently
propagate as `Err` from the append — so callers cannot distinguish "the
event was not committed, safe to retry" from "the event IS committed but a
derived update failed." Worse, when the projection-cache update fails, the
current code removes the cache entry entirely. Upcoming scheduler work will
retry appends that report failure, so this ambiguity must be resolved
before it exists: retrying a committed append would attempt a duplicate
event.
**Goal:** after this PR, the append contract is unambiguous:
1. An event that the current projection cannot legally reduce is **rejected
before anything is written** — the log, the projection cache, and the
summary row are all untouched, and the caller gets a typed rejection
error.
2. A failure of the authoritative SlateDB put (or of event-sequence
allocation) returns a typed **not-committed** error — safe to retry.
3. Once the authoritative put succeeds, the append **is committed** and
reports success. Derived-state updates (projection cache install, event
cache, SQLite summary upsert) are best-effort: failures are logged
loudly with the run id but never surface as an append error. Derived
state is repairable (startup reconciliation rebuilds it; the summary
upsert is already guarded to be monotonic by event seq, so a later
successful append also repairs it).
Design rules (fixed — do not re-litigate):
- **Validation must reuse the same reduction code that replay uses.** The
invariant is "an event is written iff replay can reduce it." Any
divergence between the pre-write check and replay reintroduces poison
events. Apply the candidate event to a clone of the current projection
using the existing reducer entry points; do not write a parallel
validity checker.
- **No event schema changes and no public API changes.** This is a store
contract fix, not a wire change.
- **Do not rework the failing call sites.** Server helpers that attempt
illegal appends will now receive a clean rejection with nothing written —
that is the intended intermediate state. Fixing their logic is separate
planned work.
- **The rejection error must be a distinct variant** from the existing
`Error::InvalidEvent` (which means "malformed payload") so callers can
tell "rejected by the run's state machine" apart from "bad input" and
from "not committed, retry."
- **Do not attempt to repair logs that already contain poison events.**
Pre-existing corrupted logs remain unreadable and continue to be surfaced
by the existing unreadable-runs listing; repair tooling is out of scope.
## Verified current state (as of origin/main `1aa7a153b`, 2026-07-28 — re-verify before starting)
- `lib/components/fabro-store/src/slate/run_store.rs`:
- `append_event(&EventPayload)``append_event_envelope` → validates the
payload shape (`payload.validate(&run_id)`), takes the per-run
`state_lock`, then calls `append_event_envelope_locked` (≈ lines
273-305).
- `append_event_if(payload, predicate)` — same, but loads the current
projection under the lock and returns `Ok(None)` when the predicate
rejects (≈ 279-294). This method's contract must be preserved.
- `append_event_envelope_locked` (≈ 305-324): allocates the event seq
(can fail with `Error::EventSequenceExhausted`), builds the
`EventEnvelope` (`RunEvent::try_from(payload)?`), then **puts the event
bytes into SlateDB first**, then `cache_event`, then
`update_summary_projection_after_append`.
- `update_summary_projection_after_append` (≈ 325-377): applies the event
to the shared projection cache; on failure it attempts a full rebuild
from the db (which, for a just-written invalid event, fails again
because the poison event is in the log), **removes the cache entry**,
warns, and returns `Err`. If the SQLite summary store is attached
(`run_summary_store` is an `OnceLock` — absent in some deployments),
an upsert failure also returns `Err`. Both paths make a committed
append look failed.
- `lib/components/fabro-store/src/error.rs`: `Error` enum with
`InvalidEvent(String)`, `EventSequenceExhausted { max_seq }`,
`Slate(..)`, `Sqlite(..)`, etc. No variant distinguishes
state-machine rejection or commit status.
- `lib/foundation/fabro-types/src/status.rs` (:132-202): the transition
table; `transition_to` returns `Err(InvalidTransition)`. From `Runnable`,
`Failed` is legal only with reason `Cancelled`.
- `lib/foundation/fabro-types/src/run_projection.rs`: `try_apply_status`
(≈ :1025) is where reduction enforces transitions; the reducer dispatch
lives in `lib/components/fabro-store/src/run_state.rs`
(`apply_event` / `apply_events`, plus `projection_from_created` for the
first event). Both files were recently extended for new event kinds —
re-derive exact line numbers rather than trusting the ones here.
- Startup behavior that makes poison events user-visible:
`warm_projection_cache` in `lib/components/fabro-store/src/slate/mod.rs`
skips runs whose replay fails (per-run `warn!`), and
`RunSummaryStore::reconcile` deletes summary rows absent from the
authoritative entries (pinned by the existing test
`reconcile_removes_rows_absent_from_authoritative_entries` in
`run_summary_store.rs`). `list_unreadable_runs` (slate/mod.rs) surfaces
skipped runs.
- The summary upsert is monotonic by event seq (`WHERE excluded.source_last_seq > runs.source_last_seq`
in `run_summary_store.rs`), which is what makes "later append repairs the
row" true.
- Existing test pinning seq exhaustion:
`append_event_rejects_sequences_beyond_key_order_limit`
(run_store.rs ≈ :1292).
## Implementation
1. **Add the typed errors** in `lib/components/fabro-store/src/error.rs`.
Read `docs/internal/error-handling-strategy.md` first (required by
project convention when touching error types). Two additions, named to
read well at call sites — suggested shapes:
- `EventRejected { reason: String }` (or carrying the
`InvalidTransition` detail) — the event cannot be legally reduced by
the run's current projection; nothing was written.
- A way for callers to know an `Err` means not-committed. Simplest
honest contract: after this PR, **every** `Err` from append means
not-committed (rejection included), because post-put failures no
longer return `Err`. Prefer that global simplification over a wrapper
enum; document it on the append methods' doc comments explicitly.
2. **Validate before the put** in `append_event_envelope_locked` (all under
the already-held `state_lock`):
- Obtain the current projection: the cheapest correct source is the
same one `append_event_if` uses (`projected_state_locked`); for a run
with no events yet, the candidate must be validated through the
first-event path (`projection_from_created` route in
`run_state.rs`) — mirror however `apply_events` treats the initial
event so validation ≡ replay exactly.
- Apply the candidate envelope to a **clone** of that projection via the
existing reducer entry point. On reduction failure → return
`EventRejected`, having written nothing.
- Keep the pre-existing `payload.validate(...)` shape check where it is.
3. **Reorder the post-put work to be best-effort.** After a successful
SlateDB put:
- Install the already-validated clone into the shared projection cache
(replacing the apply-then-rebuild-then-remove dance — the clone IS the
correct post-append projection, computed before the write). Keep the
cache's seq bookkeeping consistent with the existing
`apply_event`/`replace` semantics.
- `cache_event` and the SQLite upsert stay in place but become
log-only on failure (`warn!`/`error!` with run id and seq, matching
the logging style already present in this file). The append returns
`Ok(envelope)` regardless of derived-state failures.
- Do NOT remove the projection-cache entry on derived failure paths
anymore; a stale entry that a later append or startup reconciliation
repairs is strictly better than an absent one.
4. **Seq allocation and put failures** already return `Err` before any
derived work — with step 3 in place these are now unambiguously
not-committed. Verify `EventSequenceExhausted` still propagates (the
existing test pins it).
5. **Audit append callers for compile-only impact.** Call sites that
currently treat any `Err` as "append failed" remain correct under the
new contract (their errors now genuinely mean not-committed). No caller
behavior changes in this PR. `append_event_if`'s `Ok(None)` predicate
contract is unchanged.
6. **Doc comments.** State the three-outcome contract (rejected-nothing-
written / not-committed / committed-with-best-effort-derived) on
`append_event`, `append_event_if`, and `append_event_envelope`.
## Scope boundaries — deliberately NOT in this PR
- **The server failure helpers that attempt illegal appends** (e.g. the
worker-launch failure path appending `Failed { LaunchFailed }` from
durable `Runnable`, and similar pre-worker failure sites in
`fabro-server`) — leave their logic as-is. They will now receive a clean
`EventRejected` and write nothing, which is the intended intermediate
state; reworking when/what they append is separate planned work. Do not
"fix" them to append legal events.
- **Admission/scheduler changes** (durable claims, retry/backoff, startup
re-admission of queued runs) — known follow-up work, deliberately
excluded here.
- **Repairing already-poisoned logs** or adding repair/diagnostic tooling —
known gap, addressed separately if needed. Pre-existing unreadable runs
keep their current behavior (skipped at warmup, surfaced by the
unreadable-runs listing).
- **Event schema, OpenAPI, or public API changes** — none. This PR is
entirely inside `fabro-store` (plus its error type).
- **SQLite schema changes** — none; the monotonic upsert and startup
reconcile already provide the repair path.
If work outside these boundaries seems genuinely required for this PR to
compile or pass its tests, stop and state that in the PR description rather
than expanding scope.
## Tests (write failing-first; hermetic — temp-dir fixtures, no ambient provider keys)
Existing store tests in `run_store.rs` / `run_summary_store.rs` show the
fixture style (temp-dir object store, in-memory SQLite). Add:
1. **Rejected transition writes nothing** — create a run, drive it to
durable `Runnable` (append the events the lifecycle uses today:
created/submitted/start-requested/runnable), then append a
`run.failed { WorkflowError }`-shaped event. Assert: the append returns
the rejection variant; `list_events` shows no new event; `state()` still
reduces successfully; the projection cache still holds an entry for the
run (not removed). *Property pinned: an event is written iff replay can
reduce it.*
2. **Rejected transition leaves listings consistent** — after the rejected
append, run the summary reconcile path and assert the run's summary row
still exists. *Property: no more vanishing runs from rejected appends.*
3. **Committed append survives derived-state failure** — attach a SQLite
summary store, then make its pool unusable (e.g. close the pool or drop
the underlying file) before appending a legal event. Assert: append
returns `Ok`; the event is in `list_events`; a warning/error was the
only symptom. Then restore/reopen the summary store and assert the row
is repairable (via reconcile or a subsequent append). If pool-closing
proves impractical through public seams, an injected failing summary
store behind the existing test-support feature is acceptable — but do
not weaken the assertion that append reports success. *Property:
committed is committed.*
4. **Not-committed errors are retryable** — the existing
seq-exhaustion test keeps passing; extend it (or add a sibling) to
assert the log is unchanged after the error, pinning "Err ⇒ nothing
written."
5. **First-event validation** — a malformed first event (one the reducer
cannot initialize a projection from) is rejected with nothing written;
a valid `run.created` still works. *Property: the empty-log path
validates like replay too.*
6. **append_event_if contract unchanged** — predicate-false still returns
`Ok(None)` with nothing written.
Run the full workspace suite; the reducer and lifecycle tests in
`fabro-store`, `fabro-workflow`, and `fabro-server` are the regression net
for "legal appends behave exactly as before."
## Acceptance / verification
- `cargo +nightly-2026-04-14 fmt --check --all`
- `cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings`
- `cargo nextest run --workspace`
- No OpenAPI/wire change (do not touch `docs/public/api-reference/`).
- `cargo build --workspace` without the `test-support` feature still
succeeds if any test helper was added behind it.
## Conventions
- Read `docs/internal/error-handling-strategy.md` before changing the error
enum, and `docs/internal/events-strategy.md` before touching anything
that emits or documents events.
- Never print or log a resolved secret value, including from tests.
- Plain-English commit messages, PR text, and comments — describe what the
change does; no internal planning identifiers or plan-file names in
anything that ships.
- PR description must state plainly: (1) the vanishing-runs failure mode
this fixes (invalid append → unreadable projection → summary row deleted
→ run 404s) and that call sites attempting such appends now get a clean
error with nothing written; (2) the new append contract, including that
a failed SQLite summary update after a committed append now logs loudly
and reports success instead of returning an error — operators see a
warning where they previously saw a failed operation; (3) that
pre-existing corrupted run logs are not repaired by this change.
- If implementation uncovers a caller that genuinely depends on the old
"Err after committed write" behavior, stop and surface it in the PR
description rather than working around it.
## Completed stages
- **toolchain**: succeeded
- Script: `command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1`
- Output:
```
cargo 1.96.0 (30a34c682 2026-05-25)
```
- **preflight_compile**: succeeded
- Script: `cargo check -q --workspace 2>&1`
- Output: (empty)
- **preflight_lint**: succeeded
- Script: `cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1`
- Output: (empty)
- **implement**: succeeded
- Model: gpt-5.6-sol
- Files: /home/daytona/workspace/fabro/lib/components/fabro-store/src/error.rs, /home/daytona/workspace/fabro/lib/components/fabro-store/src/run_summary_store.rs, /home/daytona/workspace/fabro/lib/components/fabro-store/src/slate/projection_cache.rs, /home/daytona/workspace/fabro/lib/components/fabro-store/src/slate/run_store.rs, /home/daytona/workspace/fabro/lib/components/fabro-store/src/test_util.rs
# Simplify: Code Review and Cleanup
Review all changed files for reuse, quality, and efficiency. Fix any issues found.
## Phase 1: Identify Changes
Run \`git diff\` (or \`git diff HEAD\` if there are staged changes) to see what changed. If there are no git changes, review the most recently modified files that the user mentioned or that you edited earlier in this conversation.
## Phase 2: Launch Three Review Agents in Parallel
Use the ${AGENT_TOOL_NAME} tool to launch all three agents concurrently in a single message. Pass each agent the full diff so it has the complete context.
### Agent 1: Code Reuse Review
For each change:
1. **Search for existing utilities and helpers** that could replace newly written code. Look for similar patterns elsewhere in the codebase — common locations are utility directories, shared modules, and files adjacent to the changed ones.
2. **Flag any new function that duplicates existing functionality.** Suggest the existing function to use instead.
3. **Flag any inline logic that could use an existing utility** — hand-rolled string manipulation, manual path handling, custom environment checks, ad-hoc type guards, and similar patterns are common candidates.
### Agent 2: Code Quality Review
Review the same changes for hacky patterns:
1. **Redundant state**: state that duplicates existing state, cached values that could be derived, observers/effects that could be direct calls
2. **Parameter sprawl**: adding new parameters to a function instead of generalizing or restructuring existing ones
3. **Copy-paste with slight variation**: near-duplicate code blocks that should be unified with a shared abstraction
4. **Leaky abstractions**: exposing internal details that should be encapsulated, or breaking existing abstraction boundaries
5. **Stringly-typed code**: using raw strings where constants, enums (string unions), or branded types already exist in the codebase
6. **Unnecessary JSX nesting**: wrapper Boxes/elements that add no layout value — check if inner component props (flexShrink, alignItems, etc.) already provide the needed behavior
7. **Unnecessary comments**: comments explaining WHAT the code does (well-named identifiers already do that), narrating the change, or referencing the task/caller — delete; keep only non-obvious WHY (hidden constraints, subtle invariants, workarounds)
### Agent 3: Efficiency Review
Review the same changes for efficiency:
1. **Unnecessary work**: redundant computations, repeated file reads, duplicate network/API calls, N+1 patterns
2. **Missed concurrency**: independent operations run sequentially when they could run in parallel
3. **Hot-path bloat**: new blocking work added to startup or per-request/per-render hot paths
4. **Recurring no-op updates**: state/store updates inside polling loops, intervals, or event handlers that fire unconditionally — add a change-detection guard so downstream consumers aren't notified when nothing changed. Also: if a wrapper function takes an updater/reducer callback, verify it honors same-reference returns (or whatever the "no change" signal is) — otherwise callers' early-return no-ops are silently defeated
5. **Unnecessary existence checks**: pre-checking file/resource existence before operating (TOCTOU anti-pattern) — operate directly and handle the error
6. **Memory**: unbounded data structures, missing cleanup, event listener leaks
7. **Overly broad operations**: reading entire files when only a portion is needed, loading all items when filtering for one
## Phase 3: Fix Issues
Wait for all three agents to complete. Aggregate their findings and fix each issue directly. If a finding is a false positive or not worth addressing, note it and move on — do not argue with the finding, just skip it.
When done, briefly summarize what was fixed (or confirm the code was already clean).

View file

@ -0,0 +1,6 @@
{
"mode": "agent",
"provider": "openrouter",
"model": "claude-fable-5",
"reasoning_effort": "xhigh"
}