fabro(01KYQMV1VW6139EGNHEM1RGF2G): simplify_sol (succeeded)

Fabro-Run: 01KYQMV1VW6139EGNHEM1RGF2G
Fabro-Completed: 7
Fabro-Checkpoint: edcf7467e9

⚒️ Generated with [Fabro](https://fabro.sh)

lib/apps/fabro-server/src/server/tests.rs

@@ -6366,34 +6366,32 @@ async fn create_unreadable_durable_run(state: &Arc<AppState>, run_id: RunId) {
     workflow_event::append_event(&run_store, &run_id, &workflow_event::Event::RunRunning)
         .await
         .unwrap();
-    // Appends now validate before write, so a corrupted log can only exist if
-    // it predates that check. Write the poison event directly to simulate one.
-    state
-        .stores
-        .runs
-        .test_put_unvalidated_run_event(
-            &run_id,
-            5,
-            &json!({
-                "id": "evt-unreadable-run-completed",
-                "ts": "2026-05-05T20:46:33Z",
-                "run_id": run_id,
-                "event": "run.completed",
-                "properties": {
-                    "timing": {
-                        "wall_time_ms": 1,
-                        "inference_time_ms": 0,
-                        "tool_time_ms": 0,
-                        "active_time_ms": 0
-                    },
-                    "artifact_count": 0,
-                    "status": "legacy-status",
-                    "reason": "completed",
-                },
-            }),
-        )
-        .await
-        .unwrap();
+    let seq = run_store.last_event_seq().await.unwrap().unwrap() + 1;
+    let completed = workflow_event::to_run_event_at(
+        &run_id,
+        &workflow_event::Event::WorkflowRunCompleted {
+            timing:               fabro_types::RunTiming::wall_only(1),
+            artifact_count:       0,
+            status:               "legacy-status".to_string(),
+            reason:               SuccessReason::Completed,
+            total_usd_micros:     None,
+            final_git_commit_sha: None,
+            final_patch:          None,
+            diff_summary:         None,
+            billing:              None,
+        },
+        "2026-05-05T20:46:33Z".parse().unwrap(),
+        None,
+    );
+    let payload = workflow_event::build_redacted_event_payload(&completed, &run_id).unwrap();
+    fabro_store::test_support::put_unvalidated_run_event(
+        &state.stores.runs,
+        &run_id,
+        seq,
+        payload.as_value(),
+    )
+    .await
+    .unwrap();
     let err = run_store
         .state()
         .await

lib/components/fabro-store/src/error.rs

@@ -14,8 +14,11 @@ pub enum Error {
     Io(#[from] std::io::Error),
     #[error("Invalid event payload: {0}")]
     InvalidEvent(String),
-    #[error("event rejected by run projection: {reason}")]
-    EventRejected { reason: String },
+    #[error("event rejected by run projection: {source}")]
+    EventRejected {
+        #[source]
+        source: Box<Self>,
+    },
     #[error("Run not found: {0}")]
     RunNotFound(String),
     #[error("Run already exists: {0}")]

lib/components/fabro-store/src/lib.rs

@@ -10,8 +10,8 @@ mod run_state;
 mod run_summary_store;
 mod serializable_projection;
 mod slate;
-#[cfg(test)]
-mod test_util;
+#[cfg(any(test, feature = "test-support"))]
+pub mod test_support;
 mod types;
 
 pub use artifact_store::{

lib/components/fabro-store/src/run_summary_store.rs

@@ -576,7 +576,7 @@ mod tests {
         RunSummaryVisibility,
     };
     use crate::slate::CachedRunProjection;
-    use crate::test_util;
+    use crate::test_support as store_test_support;
 
     fn dt(value: &str) -> DateTime<Utc> {
         value.parse().unwrap()
@@ -613,7 +613,7 @@ mod tests {
     }
 
     async fn store() -> (tempfile::TempDir, RunSummaryStore) {
-        test_util::sqlite_summary_store().await
+        store_test_support::sqlite_summary_store().await
     }
 
     fn sample_status(kind: RunStatusKind) -> RunStatus {

lib/components/fabro-store/src/slate/mod.rs

@@ -322,10 +322,8 @@ impl Database {
         Ok(unreadable)
     }
 
-    /// Writes a raw event record without append validation, simulating a
-    /// pre-existing poison event in the log for unreadable-run tests.
     #[cfg(any(test, feature = "test-support"))]
-    pub async fn test_put_unvalidated_run_event(
+    pub(crate) async fn put_unvalidated_run_event(
         &self,
         run_id: &RunId,
         seq: u32,
@@ -541,7 +539,7 @@ mod tests {
     use object_store::path::Path;
 
     use super::*;
-    use crate::{EventPayload, keys, test_util};
+    use crate::{EventPayload, keys, test_support as store_test_support};
 
     fn dt(value: &str) -> DateTime<Utc> {
         value.parse().unwrap()
@@ -590,7 +588,7 @@ mod tests {
     }
 
     async fn make_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
-        let (directory, store) = test_util::sqlite_summary_store().await;
+        let (directory, store) = store_test_support::sqlite_summary_store().await;
         (directory, Arc::new(store))
     }
 
@@ -931,13 +929,18 @@ mod tests {
             .await
             .unwrap_err();
 
-        let Error::EventRejected { reason } = err else {
+        let Error::EventRejected { source } = err else {
             panic!("expected event rejection");
         };
-        assert_eq!(
-            reason,
-            "invalid status transition: runnable -> failed(workflow_error)"
-        );
+        assert!(matches!(
+            *source,
+            Error::InvalidTransition(fabro_types::InvalidTransition {
+                from: RunStatus::Runnable,
+                to:   RunStatus::Failed {
+                    reason: FailureReason::WorkflowError,
+                },
+            })
+        ));
         assert_eq!(run.list_events().await.unwrap(), events_before);
         assert_eq!(run.state().await.unwrap().status, RunStatus::Runnable);
         let cached = store.get_cached_run(&run_id).await.unwrap().unwrap();
@@ -971,7 +974,7 @@ mod tests {
 
     #[tokio::test]
     async fn committed_append_succeeds_when_summary_update_fails_and_is_repairable() {
-        let (_object_store, store) = make_store();
+        let (object_store, store) = make_store();
         let (directory, summaries) = make_summary_store().await;
         store.attach_run_summary_store(Arc::clone(&summaries));
         let run_id = test_run_id("run-1");
@@ -996,7 +999,8 @@ mod tests {
         let stored = run.get_event(2).await.unwrap().unwrap();
         assert_eq!(stored.event, result.unwrap().event);
 
-        let repaired_summaries = test_util::sqlite_summary_store_at(directory.path()).await;
+        let repaired_summaries =
+            Arc::new(store_test_support::sqlite_summary_store_at(directory.path()).await);
         let stale = repaired_summaries
             .get(&run_id, Utc::now())
             .await
@@ -1004,11 +1008,9 @@ mod tests {
             .unwrap();
         assert_ne!(stale.title, "Committed title");
 
-        let entries = store
-            .list_cached_runs(&ListRunsQuery::default(), Utc::now())
-            .await
-            .unwrap();
-        repaired_summaries.reconcile(&entries).await.unwrap();
+        let reopened = Database::new(object_store, "runs/", Duration::from_millis(1), None);
+        reopened.attach_run_summary_store(Arc::clone(&repaired_summaries));
+        reopened.warm_projection_cache().await.unwrap();
         let repaired = repaired_summaries
             .get(&run_id, Utc::now())
             .await
@@ -1541,7 +1543,7 @@ mod tests {
             .await
             .unwrap();
         store
-            .test_put_unvalidated_run_event(
+            .put_unvalidated_run_event(
                 &bad_run_id,
                 1,
                 &serde_json::json!({ "not": "a valid run event" }),
@@ -1590,7 +1592,7 @@ mod tests {
             .unwrap();
         run_settings.remove("integrations");
         store
-            .test_put_unvalidated_run_event(
+            .put_unvalidated_run_event(
                 &bad_run_id,
                 1,
                 &serde_json::json!({

lib/components/fabro-store/src/slate/run_store.rs

@@ -231,10 +231,10 @@ impl RunDatabase {
         self.projected_state_option_locked().await
     }
 
-    async fn install_derived_state_after_append(
+    async fn install_in_memory_state_after_append(
         &self,
         event: &EventEnvelope,
-        cached: CachedRunProjection,
+        cached: &CachedRunProjection,
     ) {
         {
             let mut projection_cache = self.inner.projection_cache.lock().await;
@@ -253,14 +253,16 @@ impl RunDatabase {
         }
         drop(recent_events);
         let _ = self.inner.event_tx.send(event.clone());
+    }
 
+    async fn update_summary_after_committed_append(&self, cached: &CachedRunProjection) {
         if let Some(store) = self.inner.run_summary_store.get() {
-            if let Err(err) = store.upsert_projection(&cached).await {
+            if let Err(err) = store.upsert_projection(cached).await {
                 warn!(
                     run_id = %self.inner.run_id,
-                    source_last_seq = event.seq,
-                    error = %err,
-                    "Failed to update SQLite run summary after committed append"
+                    source_last_seq = cached.last_seq,
+                    error = ?err,
+                    "failed to update SQLite run summary after committed append"
                 );
             }
         }
@@ -312,12 +314,19 @@ impl RunDatabase {
             return Err(Error::ReadOnly);
         }
         payload.validate(&self.inner.run_id)?;
-        let _state_guard = self.inner.state_lock.lock().await;
-        let projection = self.projected_state_locked().await?;
-        if !predicate(&projection) {
-            return Ok(None);
-        }
-        Ok(Some(self.append_event_envelope_locked(payload).await?.seq))
+        let (envelope, cached) = {
+            let _state_guard = self.inner.state_lock.lock().await;
+            let projection = self.projected_state_locked().await?;
+            if !predicate(&projection) {
+                return Ok(None);
+            }
+            let event = RunEvent::try_from(payload)?;
+            let event_bytes = serde_json::to_vec(payload)?;
+            self.append_event_envelope_locked(event, event_bytes)
+                .await?
+        };
+        self.update_summary_after_committed_append(&cached).await;
+        Ok(Some(envelope.seq))
     }
 
     /// Appends and returns the stored event envelope after pre-write reduction.
@@ -331,23 +340,30 @@ impl RunDatabase {
             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
+        let event = RunEvent::try_from(payload)?;
+        let event_bytes = serde_json::to_vec(payload)?;
+        let (envelope, cached) = {
+            let _state_guard = self.inner.state_lock.lock().await;
+            self.append_event_envelope_locked(event, event_bytes)
+                .await?
+        };
+        self.update_summary_after_committed_append(&cached).await;
+        Ok(envelope)
     }
 
-    async fn append_event_envelope_locked(&self, payload: &EventPayload) -> Result<EventEnvelope> {
+    async fn append_event_envelope_locked(
+        &self,
+        event: RunEvent,
+        event_bytes: Vec<u8>,
+    ) -> Result<(EventEnvelope, CachedRunProjection)> {
         let event_seq = self.inner.event_seq.as_ref().ok_or(Error::ReadOnly)?;
-        let seq = allocate_event_seq(event_seq)?;
-        let event = EventEnvelope {
-            seq,
-            event: RunEvent::try_from(payload)?,
-        };
+        let seq = next_event_seq(event_seq)?;
+        let envelope = EventEnvelope { seq, event };
         // Validation reduces through the exact code replay uses, so an event
         // is written iff replay can reduce it. `Arc::make_mut` copy-on-writes,
         // leaving the local projection cache untouched on rejection.
         let mut next_state = self.projected_state_for_append_locked(seq).await?;
-        apply_cached_projection_event(&mut next_state, &event)
-            .map_err(|err| event_rejected(&err))?;
+        apply_cached_projection_event(&mut next_state, &envelope).map_err(event_rejected)?;
         let next_projection =
             next_state.expect("apply_cached_projection_event sets the state on success");
         let cached = CachedRunProjection::from_projection(
@@ -355,7 +371,7 @@ impl RunDatabase {
             Arc::unwrap_or_clone(next_projection),
             seq,
         );
-        let event_bytes = serde_json::to_vec(payload)?;
+        reserve_event_seq(event_seq, seq)?;
         self.inner
             .db
             .put(
@@ -366,8 +382,8 @@ impl RunDatabase {
         // Box::pin keeps this future small enough for the
         // clippy::large_futures budget of append_event_envelope's many
         // callers.
-        Box::pin(self.install_derived_state_after_append(&event, cached)).await;
-        Ok(event)
+        Box::pin(self.install_in_memory_state_after_append(&envelope, &cached)).await;
+        Ok((envelope, cached))
     }
 
     pub async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
@@ -583,20 +599,27 @@ impl RunDatabase {
     }
 }
 
-fn event_rejected(error: &Error) -> Error {
+fn event_rejected(error: Error) -> Error {
     Error::EventRejected {
-        reason: error.to_string(),
+        source: Box::new(error),
     }
 }
 
-fn allocate_event_seq(event_seq: &AtomicU32) -> Result<u32> {
-    event_seq
-        .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |seq| {
-            (seq <= keys::MAX_EVENT_SEQ).then_some(seq + 1)
-        })
-        .map_err(|_| Error::EventSequenceExhausted {
+fn next_event_seq(event_seq: &AtomicU32) -> Result<u32> {
+    let seq = event_seq.load(Ordering::SeqCst);
+    if seq > keys::MAX_EVENT_SEQ {
+        return Err(Error::EventSequenceExhausted {
             max_seq: keys::MAX_EVENT_SEQ,
-        })
+        });
+    }
+    Ok(seq)
+}
+
+fn reserve_event_seq(event_seq: &AtomicU32, seq: u32) -> Result<()> {
+    event_seq
+        .compare_exchange(seq, seq + 1, Ordering::SeqCst, Ordering::SeqCst)
+        .map(|_| ())
+        .map_err(|_| Error::Other("event sequence changed while append lock was held".to_string()))
 }
 
 fn apply_cached_projection_event(
@@ -1288,6 +1311,29 @@ mod tests {
         assert_eq!(seqs, vec![5, 4, 3]);
     }
 
+    #[tokio::test]
+    async fn rejected_event_does_not_consume_last_available_sequence() {
+        let run = fresh_run().await;
+        let run_id = run.run_id();
+        run.inner
+            .event_seq
+            .as_ref()
+            .unwrap()
+            .store(keys::MAX_EVENT_SEQ, Ordering::SeqCst);
+
+        let err = run
+            .append_event(&run_created_payload(&run_id))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::EventRejected { .. }));
+
+        let seq = run
+            .append_event(&stage_prompt_payload(&run_id, 1, Some("alpha")))
+            .await
+            .unwrap();
+        assert_eq!(seq, keys::MAX_EVENT_SEQ);
+    }
+
     #[tokio::test]
     async fn append_event_rejects_sequences_beyond_key_order_limit() {
         let run = fresh_run().await;

lib/components/fabro-store/src/test_support/mod.rs

@@ -1,13 +1,33 @@
+#[cfg(test)]
 use std::path::Path;
 
-use crate::RunSummaryStore;
+use fabro_types::RunId;
 
+#[cfg(test)]
+use crate::RunSummaryStore;
+use crate::{Database, Result};
+
+/// Writes an event without append validation to model a log corrupted by an
+/// older Fabro version.
+pub async fn put_unvalidated_run_event(
+    database: &Database,
+    run_id: &RunId,
+    seq: u32,
+    payload: &serde_json::Value,
+) -> Result<()> {
+    database
+        .put_unvalidated_run_event(run_id, seq, payload)
+        .await
+}
+
+#[cfg(test)]
 pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStore) {
     let directory = tempfile::tempdir().unwrap();
     let store = sqlite_summary_store_at(directory.path()).await;
     (directory, store)
 }
 
+#[cfg(test)]
 pub(crate) async fn sqlite_summary_store_at(directory: &Path) -> RunSummaryStore {
     let database = fabro_db::Database::connect(directory.join("fabro.sqlite3"))
         .await

lib/foundation/fabro-types/src/run_event/mod.rs

@@ -12,7 +12,7 @@ pub use fabro_model::BilledTokenCounts;
 pub use infra::*;
 pub use misc::*;
 pub use run::*;
-use serde::de::Error as _;
+use serde::de::Error as DeError;
 use serde::ser::Error as SerError;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_json::{Map, Value, json};
@@ -764,6 +764,57 @@ impl RunEvent {
         })
     }
 
+    pub fn from_ref(value: &Value) -> serde_json::Result<Self> {
+        fn opt_field<T: for<'a> Deserialize<'a>>(
+            obj: &Map<String, Value>,
+            key: &str,
+        ) -> serde_json::Result<Option<T>> {
+            match obj.get(key) {
+                Some(value) if !value.is_null() => Ok(Some(T::deserialize(value)?)),
+                _ => Ok(None),
+            }
+        }
+
+        let obj = value.as_object().ok_or_else(|| {
+            <serde_json::Error as DeError>::custom("run event must be a JSON object")
+        })?;
+        let opt_str = |key: &str| obj.get(key).and_then(Value::as_str).map(str::to_string);
+        let id = obj.get("id").and_then(Value::as_str).ok_or_else(|| {
+            <serde_json::Error as DeError>::custom("missing or non-string field: id")
+        })?;
+        let ts = obj
+            .get("ts")
+            .ok_or_else(|| <serde_json::Error as DeError>::custom("missing field: ts"))
+            .and_then(DateTime::<Utc>::deserialize)?;
+        let run_id = obj
+            .get("run_id")
+            .ok_or_else(|| <serde_json::Error as DeError>::custom("missing field: run_id"))
+            .and_then(RunId::deserialize)?;
+        let event = obj.get("event").and_then(Value::as_str).ok_or_else(|| {
+            <serde_json::Error as DeError>::custom("missing or non-string field: event")
+        })?;
+        let properties = obj
+            .get("properties")
+            .cloned()
+            .unwrap_or_else(default_properties);
+        Self::from_parts(RunEventParts {
+            id: id.to_string(),
+            ts,
+            run_id,
+            node_id: opt_str("node_id"),
+            node_label: opt_str("node_label"),
+            stage_id: opt_field(obj, "stage_id")?,
+            parallel_group_id: opt_field(obj, "parallel_group_id")?,
+            parallel_branch_id: opt_field(obj, "parallel_branch_id")?,
+            session_id: opt_str("session_id"),
+            parent_session_id: opt_str("parent_session_id"),
+            tool_call_id: opt_str("tool_call_id"),
+            actor: opt_field(obj, "actor")?,
+            event,
+            properties: &properties,
+        })
+    }
+
     fn from_parts(parts: RunEventParts<'_>) -> serde_json::Result<Self> {
         let body_payload = json!({
             "event": parts.event,