scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-07 08:27:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

checkpoint

Browse source 
⚒️ Generated with [Fabro](https://fabro.sh)
This commit is contained in:
Fabro 2026-07-30 00:25:33 +00:00
parent edcf7467e9
commit 7425261f11
6 changed files with 1224 additions and 343 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1012
run.json
View file

File diff suppressed because one or more lines are too long

534
stages/007-simplify_sol@1/diff.patch Normal file
View file

@ -0,0 +1,534 @@
diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs
index 7735c5e99..063b48f36 100644
--- a/lib/apps/fabro-server/src/server/tests.rs
+++ b/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
diff --git a/lib/components/fabro-store/src/error.rs b/lib/components/fabro-store/src/error.rs
index ff78213d0..df12e5994 100644
--- a/lib/components/fabro-store/src/error.rs
+++ b/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}")]
diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs
index 0eb246d04..e99eadd0f 100644
--- a/lib/components/fabro-store/src/lib.rs
+++ b/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::{
diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs
index c1a339b99..5db1a49b1 100644
--- a/lib/components/fabro-store/src/run_summary_store.rs
+++ b/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 {
diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs
index d9f542498..893b0d8f0 100644
--- a/lib/components/fabro-store/src/slate/mod.rs
+++ b/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!({
diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs
index 57f54a311..be9ca78bf 100644
--- a/lib/components/fabro-store/src/slate/run_store.rs
+++ b/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;
diff --git a/lib/components/fabro-store/src/test_util.rs b/lib/components/fabro-store/src/test_support/mod.rs
similarity index 55%
rename from lib/components/fabro-store/src/test_util.rs
rename to lib/components/fabro-store/src/test_support/mod.rs
index a5b7e17f9..7ae613652 100644
--- a/lib/components/fabro-store/src/test_util.rs
+++ b/lib/components/fabro-store/src/test_support/mod.rs
@@ -1,13 +1,33 @@
+#[cfg(test)]
use std::path::Path;
+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
diff --git a/lib/foundation/fabro-types/src/run_event/mod.rs b/lib/foundation/fabro-types/src/run_event/mod.rs
index 34cc88ed8..6680fff2e 100644
--- a/lib/foundation/fabro-types/src/run_event/mod.rs
+++ b/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,

6
stages/007-simplify_sol@1/status.json Normal file
View file

@ -0,0 +1,6 @@
{
"outcome": "succeeded",
"notes": "Stage completed: simplify_sol",
"failure_reason": null,
"timestamp": "2026-07-30T00:22:04.466010587Z"
}

1
stages/008-verify@1/output.log Normal file
View file

@ -0,0 +1 @@
blob://sha256/178cea91c4ccc809d878db2db8201bbd86ee05611d923ea234d185accd364e0f

6
stages/008-verify@1/script_invocation.json Normal file
View file

@ -0,0 +1,6 @@
{
"script": "git fetch origin main 2>&1 && git merge --no-edit --no-stat origin/main 2>&1 && cargo +nightly-2026-04-14 fmt --all 2>&1 && cargo dev docs refresh 2>&1 && cargo +nightly-2026-04-14 fmt --check --all 2>&1 && { command -v rg >/dev/null 2>&1 || { echo 'rg is required for verify'; exit 127; }; } && ! rg -n 'AuthMode::Disabled|RunAuthMethod|RunSubjectProvenance|\\bActorRef\\b|\\bActorKind\\b|AuthenticatedSubject|AuthenticatedService|AuthorizeRunScoped|AuthorizeRunBlob|AuthorizeStageArtifact|AuthorizeCommandLog|auth_method\\s*==\\s*\"disabled\"' lib/crates apps lib/packages docs/public/api-reference/fabro-api.yaml 2>&1 && cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --workspace --status-level slow --profile ci 2>&1 && cargo dev docs check 2>&1 && bun install --frozen-lockfile 2>&1 && (cd apps/fabro-web && bun run typecheck) 2>&1 && (cd apps/fabro-web && bun run test) 2>&1 && (cd lib/packages/fabro-api-client && bun run typecheck) 2>&1 && cargo dev build -- -p fabro-cli --release 2>&1",
"command": "exec 2>&1\ngit fetch origin main 2>&1 && git merge --no-edit --no-stat origin/main 2>&1 && cargo +nightly-2026-04-14 fmt --all 2>&1 && cargo dev docs refresh 2>&1 && cargo +nightly-2026-04-14 fmt --check --all 2>&1 && { command -v rg >/dev/null 2>&1 || { echo 'rg is required for verify'; exit 127; }; } && ! rg -n 'AuthMode::Disabled|RunAuthMethod|RunSubjectProvenance|\\bActorRef\\b|\\bActorKind\\b|AuthenticatedSubject|AuthenticatedService|AuthorizeRunScoped|AuthorizeRunBlob|AuthorizeStageArtifact|AuthorizeCommandLog|auth_method\\s*==\\s*\"disabled\"' lib/crates apps lib/packages docs/public/api-reference/fabro-api.yaml 2>&1 && cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --workspace --status-level slow --profile ci 2>&1 && cargo dev docs check 2>&1 && bun install --frozen-lockfile 2>&1 && (cd apps/fabro-web && bun run typecheck) 2>&1 && (cd apps/fabro-web && bun run test) 2>&1 && (cd lib/packages/fabro-api-client && bun run typecheck) 2>&1 && cargo dev build -- -p fabro-cli --release 2>&1",
"language": "shell",
"timeout_ms": 1200000
}

8
stages/008-verify@1/script_timing.json Normal file
View file

@ -0,0 +1,8 @@
{
"output": "blob://sha256/178cea91c4ccc809d878db2db8201bbd86ee05611d923ea234d185accd364e0f",
"exit_code": 1,
"duration_ms": 204855,
"termination": "exited",
"output_bytes": 8380,
"live_streaming": true
}