scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Rename RunBlobId to BlobHash

Browse source
This commit is contained in:
Scott Werner 2026-08-11 13:37:03 -04:00
parent 0a40061783
commit 62ed7cb8a2
29 changed files with 152 additions and 156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
lib/apps/fabro-cli/src/commands/run/output.rs
View file

@ -5,7 +5,7 @@ use anyhow::{Context as _, Result};
use cli_table::format::{Border, Justify, Separator};
use cli_table::{Cell, CellStruct, Style, Table};
use fabro_api::types;
use fabro_types::{PullRequestLink, RunBlobId, RunId, StageId, parse_blob_ref};
use fabro_types::{BlobHash, PullRequestLink, RunId, StageId, parse_blob_ref};
use fabro_util::check_report::{CheckDetail, CheckReport, CheckResult, CheckSection, CheckStatus};
use fabro_util::error::render_with_causes;
use fabro_util::printer::Printer;
@ -341,7 +341,7 @@ async fn resolve_response_string(
}))
}
fn blob_id_from_response(response: &str) -> Option<RunBlobId> {
fn blob_id_from_response(response: &str) -> Option<BlobHash> {
parse_blob_ref(response)
}

6
lib/apps/fabro-cli/src/commands/run/runner.rs
View file

@ -21,7 +21,7 @@ use fabro_store::{EventEnvelope, RunProjection, RunProjectionReducer};
use fabro_tool::fabro_client::ClientBackend;
use fabro_types::settings::run::{RunMode, RunNamespace};
use fabro_types::{
ArtifactUpload, EventBody, FailureReason, Principal, RunBlobId, RunEvent, RunId,
ArtifactUpload, BlobHash, EventBody, FailureReason, Principal, RunEvent, RunId,
WorkflowSettings,
};
use fabro_vault::{SecretStore, Vault};
@ -1008,7 +1008,7 @@ impl RunStoreBackend for HttpRunStore {
self.apply_acknowledged_event(seq, event).await
}
async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
async fn write_blob(&self, data: &[u8]) -> Result<BlobHash> {
self.with_retries("write run blob", || {
let client = self.client.clone_for_reuse();
let run_id = self.run_id;
@ -1018,7 +1018,7 @@ impl RunStoreBackend for HttpRunStore {
.await
}
async fn read_blob(&self, id: &RunBlobId) -> Result<Option<bytes::Bytes>> {
async fn read_blob(&self, id: &BlobHash) -> Result<Option<bytes::Bytes>> {
self.with_retries("read run blob", || {
let client = self.client.clone_for_reuse();
let run_id = self.run_id;

4
lib/apps/fabro-server/src/principal_middleware.rs
View file

@ -7,7 +7,7 @@ use axum::http::StatusCode;
use axum::http::request::Parts;
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use fabro_types::{AuthMethod, IdpIdentity, Principal, RunBlobId, RunId, StageId, UserPrincipal};
use fabro_types::{AuthMethod, BlobHash, IdpIdentity, Principal, RunId, StageId, UserPrincipal};
use jsonwebtoken::decode_header;
use strum::IntoStaticStr;
@ -61,7 +61,7 @@ pub(crate) struct RequiredRunToolActor(pub(crate) Principal);
pub(crate) struct RequireRunScoped(pub(crate) RunId);
pub(crate) struct RequireWorkerRunScoped(pub(crate) RunId);
pub(crate) struct RequireRunManagementTarget(pub(crate) RunId, pub(crate) Principal);
pub(crate) struct RequireRunBlob(pub(crate) RunId, pub(crate) RunBlobId);
pub(crate) struct RequireRunBlob(pub(crate) RunId, pub(crate) BlobHash);
pub(crate) struct RequireRunStageScoped(pub(crate) RunId, pub(crate) String);
pub(crate) struct RequireStageArtifact(pub(crate) RunId, pub(crate) StageId);
pub(crate) struct RequireCommandLog(pub(crate) RunId, pub(crate) StageId);

12
lib/apps/fabro-server/src/server.rs
View file

@ -96,10 +96,10 @@ use fabro_types::settings::server::{
GithubIntegrationSettings, GithubIntegrationStrategy, LogDestination,
};
use fabro_types::{
AgentBackend, AskFabro, AskFabroUnavailableReason, EventBody, InterviewQuestionRecord, PairId,
PairMessageId, PairTarget, PendingReason, Principal, PullRequestLink, QuestionType, RunBlobId,
RunControlAction, RunEvent, RunId, RunRunnableSource, SandboxProviderKind, ServerSettings,
SessionCapability,
AgentBackend, AskFabro, AskFabroUnavailableReason, BlobHash, EventBody,
InterviewQuestionRecord, PairId, PairMessageId, PairTarget, PendingReason, Principal,
PullRequestLink, QuestionType, RunControlAction, RunEvent, RunId, RunRunnableSource,
SandboxProviderKind, ServerSettings, SessionCapability,
};
use fabro_util::error::{
SharedError, collect_causes, render_compact_with_causes, render_with_causes,
@ -2891,8 +2891,8 @@ pub(crate) fn parse_stage_id_path(stage_id: &str) -> Result<StageId, Response> {
clippy::result_large_err,
reason = "Blob ID parsing returns HTTP 400 responses directly."
)]
pub(crate) fn parse_blob_id_path(blob_id: &str) -> Result<RunBlobId, Response> {
RunBlobId::from_str(blob_id)
pub(crate) fn parse_blob_id_path(blob_id: &str) -> Result<BlobHash, Response> {
BlobHash::from_str(blob_id)
.map_err(|_| ApiError::bad_request("Invalid blob ID.").into_response())
}

20
lib/apps/fabro-server/src/server/tests.rs
View file

@ -26,12 +26,12 @@ use fabro_model::{Catalog, ModelRef, ProviderId, ReasoningEffort, Speed};
use fabro_types::settings::ServerAuthMethod;
use fabro_types::settings::run::EnvironmentProvider;
use fabro_types::{
AgentBackend, AttrValue, AuthMethod, CommandTermination, FailureCategory, FailureDetail, Graph,
InterviewQuestionRecord, Node, Outcome, ParallelBranchId, QuestionType, RunBlobId, RunId,
RunSpec, SandboxProviderKind, StageContextWindowBreakdownItem, StageContextWindowCategory,
StageContextWindowCountMethod, StageContextWindowProjection, StageContextWindowStaleness,
StageContextWindowWarning, StageModelUsage, StageTiming, SuccessReason, SystemActorKind,
WorkflowSettings, fixtures, test_support,
AgentBackend, AttrValue, AuthMethod, BlobHash, CommandTermination, FailureCategory,
FailureDetail, Graph, InterviewQuestionRecord, Node, Outcome, ParallelBranchId, QuestionType,
RunId, RunSpec, SandboxProviderKind, StageContextWindowBreakdownItem,
StageContextWindowCategory, StageContextWindowCountMethod, StageContextWindowProjection,
StageContextWindowStaleness, StageContextWindowWarning, StageModelUsage, StageTiming,
SuccessReason, SystemActorKind, WorkflowSettings, fixtures, test_support,
};
use fabro_util::check_report::CheckStatus;
use fabro_workflow::records::CheckpointExt;
@ -3890,7 +3890,7 @@ layer = "project"
let manifest_blob = created["properties"]["manifest_blob"]
.as_str()
.expect("run.created should carry the submitted source blob")
.parse::<RunBlobId>()
.parse::<BlobHash>()
.unwrap();
let persisted_manifest = run_store
.read_blob(&manifest_blob)
@ -10786,12 +10786,12 @@ async fn create_run_persists_manifest_and_definition_blobs_without_bundle_file()
let manifest_blob = created["properties"]["manifest_blob"]
.as_str()
.expect("run.created should carry manifest_blob")
.parse::<RunBlobId>()
.parse::<BlobHash>()
.unwrap();
let definition_blob = submitted["properties"]["definition_blob"]
.as_str()
.expect("run.submitted should carry definition_blob")
.parse::<RunBlobId>()
.parse::<BlobHash>()
.unwrap();
let submitted_manifest_bytes = run_store
@ -12058,7 +12058,7 @@ async fn worker_token_is_rejected_on_user_only_routes() {
let user_jwt = issue_test_user_jwt();
let run_id = create_run_with_bearer(&app, &user_jwt).await;
let worker_token = issue_test_worker_token(&run_id);
let blob_id = RunBlobId::new(b"blob");
let blob_id = BlobHash::new(b"blob");
let user_only_routes = vec![
(Method::GET, "/runs".to_string()),
(Method::POST, "/runs".to_string()),

12
lib/components/fabro-dump/src/lib.rs
View file

@ -16,10 +16,10 @@ use bytes::Bytes;
use fabro_store::{
EventEnvelope, RunProjection, SerializableProjection, StageId, retry_storage_segment,
};
use fabro_types::{RunBlobId, parse_blob_ref};
use fabro_types::{BlobHash, parse_blob_ref};
use futures::future::BoxFuture;
pub type BlobReader = Box<dyn FnMut(RunBlobId) -> BoxFuture<'static, Result<Option<Bytes>>> + Send>;
pub type BlobReader = Box<dyn FnMut(BlobHash) -> BoxFuture<'static, Result<Option<Bytes>>> + Send>;
const STAGE_RANK_WIDTH: usize = 3;
const MAX_STAGES_IN_DUMP: usize = {
@ -208,7 +208,7 @@ impl RunDump {
mut read_blob: F,
) -> Result<()>
where
F: FnMut(RunBlobId) -> BoxFuture<'a, Result<Option<Bytes>>>,
F: FnMut(BlobHash) -> BoxFuture<'a, Result<Option<Bytes>>>,
{
let mut cache = HashMap::new();
for entry in &mut self.entries {
@ -386,7 +386,7 @@ fn validate_relative_path(kind: &str, value: &str) -> Result<PathBuf> {
Ok(normalized)
}
fn collect_blob_refs_in_value(value: &serde_json::Value, blob_ids: &mut Vec<RunBlobId>) {
fn collect_blob_refs_in_value(value: &serde_json::Value, blob_ids: &mut Vec<BlobHash>) {
match value {
serde_json::Value::String(current) => {
if let Some(blob_id) = parse_blob_ref(current) {
@ -409,7 +409,7 @@ fn collect_blob_refs_in_value(value: &serde_json::Value, blob_ids: &mut Vec<RunB
fn replace_blob_refs_in_value(
value: &mut serde_json::Value,
cache: &HashMap<RunBlobId, serde_json::Value>,
cache: &HashMap<BlobHash, serde_json::Value>,
) -> Result<()> {
match value {
serde_json::Value::String(current) => {
@ -724,7 +724,7 @@ mod tests {
#[test]
fn hydrate_referenced_blobs_ignores_legacy_artifact_file_refs() {
let blob = serde_json::to_vec("hydrated legacy text").unwrap();
let blob_id = fabro_types::RunBlobId::new(&blob);
let blob_id = fabro_types::BlobHash::new(&blob);
let legacy_ref = format!("file:///sandbox/.fabro/artifacts/{blob_id}.json");
let mut dump = RunDump {
entries: vec![RunDumpEntry::json(

2
lib/components/fabro-store/src/lib.rs
View file

@ -20,7 +20,7 @@ pub use artifact_store::{
};
pub use error::{Error, Result};
pub use fabro_types::{
EventEnvelope, PendingInterviewRecord, Run, RunBlobId, RunProjection, StageId, StageProjection,
BlobHash, EventEnvelope, PendingInterviewRecord, Run, RunProjection, StageId, StageProjection,
};
pub use keyed_mutex::{KeyedMutex, KeyedMutexGuard};
pub use run_sessions::{

8
lib/components/fabro-store/src/record/record_id.rs
View file

@ -1,4 +1,4 @@
use fabro_types::{RunBlobId, RunId};
use fabro_types::{BlobHash, RunId};
use super::RecordId;
use crate::{Error, Result};
@ -38,7 +38,7 @@ impl RecordId for String {
}
}
impl RecordId for RunBlobId {
impl RecordId for BlobHash {
fn key_segments(&self) -> Vec<String> {
vec![self.to_string()]
}
@ -46,13 +46,13 @@ impl RecordId for RunBlobId {
fn from_key_segments(segs: &[&str]) -> Result<Self> {
let [segment] = segs else {
return Err(Error::KeyParse(format!(
"expected 1 segment for RunBlobId, got {}",
"expected 1 segment for BlobHash, got {}",
segs.len()
)));
};
segment
.parse()
.map_err(|err| Error::KeyParse(format!("invalid RunBlobId segment {segment:?}: {err}")))
.map_err(|err| Error::KeyParse(format!("invalid BlobHash segment {segment:?}: {err}")))
}
}

14
lib/components/fabro-store/src/run_state.rs
View file

@ -1690,11 +1690,11 @@ mod tests {
use fabro_types::settings::run::{DockerfileSource, EnvironmentProvider};
use fabro_types::{
AgentBackend, AgentControlState, AttrValue, AutomationRef, BilledModelUsage,
BilledTokenCounts, BlockedReason, Checkpoint, CheckpointRecord, CommandTermination,
EventBody, FailureCategory, FailureDetail, FailureReason, Graph, McpServerStatus, Node,
Outcome, ParallelBranchId, PendingReason, PermissionLevel, PullRequestCreationStatus,
PullRequestLink, QuestionType, ReasoningEffort, RunApprovalState, RunBlobId,
RunControlAction, RunDiff, RunEvent, RunSize, RunSpec, RunStatus, Speed,
BilledTokenCounts, BlobHash, BlockedReason, Checkpoint, CheckpointRecord,
CommandTermination, EventBody, FailureCategory, FailureDetail, FailureReason, Graph,
McpServerStatus, Node, Outcome, ParallelBranchId, PendingReason, PermissionLevel,
PullRequestCreationStatus, PullRequestLink, QuestionType, ReasoningEffort,
RunApprovalState, RunControlAction, RunDiff, RunEvent, RunSize, RunSpec, RunStatus, Speed,
StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod,
StageContextWindowProjection, StageContextWindowStaleness, StageContextWindowWarning,
StageHandler, StageModelUsage, StageOutcome, StageState, StageTiming, SubAgentStatus,
@ -4238,9 +4238,9 @@ mod tests {
#[test]
fn projection_serialization_includes_manifest_and_definition_blob_refs() {
let manifest_blob = RunBlobId::new(br#"{"version":1}"#).to_string();
let manifest_blob = BlobHash::new(br#"{"version":1}"#).to_string();
let definition_blob =
RunBlobId::new(br#"{"version":1,"workflow_path":"workflow.fabro"}"#).to_string();
BlobHash::new(br#"{"version":1,"workflow_path":"workflow.fabro"}"#).to_string();
let events = vec![
EventEnvelope {
seq: 1,

18
lib/components/fabro-store/src/slate/blob_store.rs
View file

@ -1,7 +1,7 @@
use std::sync::Arc;
use bytes::Bytes;
use fabro_types::RunBlobId;
use fabro_types::BlobHash;
use futures::StreamExt;
use tracing::warn;
@ -24,13 +24,13 @@ impl From<Bytes> for Blob {
}
impl Record for Blob {
type Id = RunBlobId;
type Id = BlobHash;
type Codec = RawBytesCodec;
const PREFIX: &'static str = "blobs/sha256";
fn id(&self) -> Self::Id {
RunBlobId::new(&self.0)
BlobHash::new(&self.0)
}
}
@ -51,22 +51,22 @@ impl BlobStore {
}
}
pub async fn write(&self, bytes: &[u8]) -> Result<RunBlobId> {
pub async fn write(&self, bytes: &[u8]) -> Result<BlobHash> {
let blob = Blob(Bytes::copy_from_slice(bytes));
let id = blob.id();
self.repo.put(&blob).await?;
Ok(id)
}
pub async fn read(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
pub async fn read(&self, id: &BlobHash) -> Result<Option<Bytes>> {
Ok(self.repo.get(id).await?.map(|blob| blob.0))
}
pub async fn exists(&self, id: &RunBlobId) -> Result<bool> {
pub async fn exists(&self, id: &BlobHash) -> Result<bool> {
self.repo.exists(id).await
}
pub(crate) async fn list(&self) -> Result<Vec<RunBlobId>> {
pub(crate) async fn list(&self) -> Result<Vec<BlobHash>> {
let mut stream = self.repo.scan_ids_stream();
let mut ids = Vec::new();
while let Some(result) = stream.next().await {
@ -89,7 +89,7 @@ mod tests {
use std::time::Duration;
use bytes::Bytes;
use fabro_types::RunBlobId;
use fabro_types::BlobHash;
use object_store::memory::InMemory;
use super::BlobStore;
@ -128,7 +128,7 @@ mod tests {
);
assert_eq!(store.write(bytes).await.unwrap(), id);
assert!(store.exists(&id).await.unwrap());
assert!(!store.exists(&RunBlobId::new(b"missing")).await.unwrap());
assert!(!store.exists(&BlobHash::new(b"missing")).await.unwrap());
}
#[tokio::test]

8
lib/components/fabro-store/src/slate/run_store.rs
View file

@ -4,7 +4,7 @@ use std::sync::{Arc, OnceLock};
use bytes::Bytes;
use chrono::Utc;
use fabro_types::{RunBlobId, RunEvent, RunId, SessionId};
use fabro_types::{BlobHash, RunEvent, RunId, SessionId};
use futures::Stream;
use slatedb::{Db, DbIterator, DbRead};
use tokio::sync::{Mutex, broadcast, mpsc};
@ -554,18 +554,18 @@ impl RunDatabase {
Ok(Box::pin(UnboundedReceiverStream::new(receiver)))
}
pub async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
pub async fn write_blob(&self, data: &[u8]) -> Result<BlobHash> {
if self.read_only {
return Err(Error::ReadOnly);
}
self.inner.blob_store.write(data).await
}
pub async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
pub async fn read_blob(&self, id: &BlobHash) -> Result<Option<Bytes>> {
self.inner.blob_store.read(id).await
}
pub async fn list_blobs(&self) -> Result<Vec<RunBlobId>> {
pub async fn list_blobs(&self) -> Result<Vec<BlobHash>> {
self.inner.blob_store.list().await
}

16
lib/components/fabro-workflow/src/artifact.rs
View file

@ -4,7 +4,7 @@ use std::path::{Path, PathBuf};
use fabro_agent::Sandbox;
use fabro_config::RunScratch;
use fabro_types::{
ParallelBranchResult, RunBlobId, format_blob_ref, parse_blob_ref, parse_managed_blob_file_ref,
BlobHash, ParallelBranchResult, format_blob_ref, parse_blob_ref, parse_managed_blob_file_ref,
};
use futures::future::BoxFuture;
use serde_json::Value;
@ -413,7 +413,7 @@ fn resolve_execution_value<'a>(
}
async fn materialize_blob_ref(
blob_id: &RunBlobId,
blob_id: &BlobHash,
run_store: &RunStoreHandle,
env: &dyn Sandbox,
run_dir: &Path,
@ -457,7 +457,7 @@ async fn materialize_blob_ref(
}
async fn read_required_blob(
blob_id: &RunBlobId,
blob_id: &BlobHash,
run_store: &RunStoreHandle,
) -> Result<bytes::Bytes> {
run_store
@ -508,7 +508,7 @@ async fn is_local_execution(env: &dyn Sandbox, run_dir: &Path) -> Result<bool> {
.map_err(|e| Error::engine_with_source("failed to inspect sandbox locality", e))
}
fn local_materialized_blob_path(run_dir: &Path, blob_id: &RunBlobId) -> PathBuf {
fn local_materialized_blob_path(run_dir: &Path, blob_id: &BlobHash) -> PathBuf {
RunScratch::new(run_dir)
.runtime_dir()
.join("blobs")
@ -549,7 +549,7 @@ mod tests {
let large_string = "x".repeat(BLOB_OFFLOAD_THRESHOLD + 1);
let serialized = serde_json::to_vec(&serde_json::json!(large_string.clone())).unwrap();
let expected_blob_id = fabro_types::RunBlobId::new(&serialized);
let expected_blob_id = fabro_types::BlobHash::new(&serialized);
let mut updates = HashMap::new();
updates.insert("response.plan".to_string(), serde_json::json!(large_string));
@ -636,7 +636,7 @@ mod tests {
Value::String("small".to_string());
BLOB_OFFLOAD_THRESHOLD / 4
]);
let expected_report_blob = RunBlobId::new(&serde_json::to_vec(&large_report).unwrap());
let expected_report_blob = BlobHash::new(&serde_json::to_vec(&large_report).unwrap());
let mut typed_results = vec![ParallelBranchResult {
id: "branch_a".to_string(),
index: Some(0),
@ -789,7 +789,7 @@ mod tests {
#[test]
fn normalize_durable_updates_rewrites_managed_blob_file_refs_recursively() {
let blob_id = fabro_types::RunBlobId::new(b"hello");
let blob_id = fabro_types::BlobHash::new(b"hello");
let mut updates = HashMap::from([(
"nested".to_string(),
serde_json::json!({
@ -870,7 +870,7 @@ mod tests {
#[test]
fn normalize_checkpoint_for_resume_converts_managed_blob_file_refs_and_drops_preamble() {
let blob_id = fabro_types::RunBlobId::new(b"managed");
let blob_id = fabro_types::BlobHash::new(b"managed");
let mut checkpoint = crate::records::Checkpoint {
timestamp: chrono::Utc::now(),
current_node: "work".to_string(),

12
lib/components/fabro-workflow/src/event/events.rs
View file

@ -1,12 +1,12 @@
use std::collections::BTreeMap;
use ::fabro_types::{
AutomationRef, BilledTokenCounts, BlockedReason, CommandTermination, DiffSummary,
AutomationRef, BilledTokenCounts, BlobHash, BlockedReason, CommandTermination, DiffSummary,
FailureReason, ForkSourceRef, GitContext, PairId, PairMessageId, PairSystemMessageKind,
PairTarget, ParallelBranchId, ParallelBranchResult, PendingReason, PermissionLevel, Principal,
PullRequestCreationId, PullRequestLink, ReviewTarget, RunBlobId, RunFailure, RunId,
RunNoticeLevel, RunPairEndedReason, RunPairFailedReason, RunProvenance, RunRunnableSource,
RunTiming, SandboxProviderKind, StageId, StageOutcome, StageTiming, SuccessReason,
PullRequestCreationId, PullRequestLink, ReviewTarget, RunFailure, RunId, RunNoticeLevel,
RunPairEndedReason, RunPairFailedReason, RunProvenance, RunRunnableSource, RunTiming,
SandboxProviderKind, StageId, StageOutcome, StageTiming, SuccessReason,
run_event as fabro_types,
};
use fabro_agent::{AgentEvent, SandboxEvent};
@ -39,7 +39,7 @@ pub enum Event {
automation: Option<AutomationRef>,
provenance: RunProvenance,
#[serde(default, skip_serializing_if = "Option::is_none")]
manifest_blob: Option<RunBlobId>,
manifest_blob: Option<BlobHash>,
#[serde(default, skip_serializing_if = "Option::is_none")]
git: Option<GitContext>,
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -67,7 +67,7 @@ pub enum Event {
},
RunSubmitted {
#[serde(default, skip_serializing_if = "Option::is_none")]
definition_blob: Option<RunBlobId>,
definition_blob: Option<BlobHash>,
},
RunStartRequested {
resume: bool,

8
lib/components/fabro-workflow/src/handler/command.rs
View file

@ -354,7 +354,7 @@ mod tests {
#[derive(Default)]
struct MemoryRunStoreBackend {
blobs: Mutex<std::collections::HashMap<fabro_types::RunBlobId, Bytes>>,
blobs: Mutex<std::collections::HashMap<fabro_types::BlobHash, Bytes>>,
}
#[async_trait::async_trait]
@ -389,8 +389,8 @@ mod tests {
Ok(())
}
async fn write_blob(&self, data: &[u8]) -> anyhow::Result<fabro_types::RunBlobId> {
let blob_id = fabro_types::RunBlobId::new(data);
async fn write_blob(&self, data: &[u8]) -> anyhow::Result<fabro_types::BlobHash> {
let blob_id = fabro_types::BlobHash::new(data);
self.blobs
.lock()
.await
@ -398,7 +398,7 @@ mod tests {
Ok(blob_id)
}
async fn read_blob(&self, id: &fabro_types::RunBlobId) -> anyhow::Result<Option<Bytes>> {
async fn read_blob(&self, id: &fabro_types::BlobHash) -> anyhow::Result<Option<Bytes>> {
Ok(self.blobs.lock().await.get(id).cloned())
}

2
lib/components/fabro-workflow/src/handler/parallel.rs
View file

@ -1847,7 +1847,7 @@ mod tests {
Some(serde_json::json!({"not": "an array"})),
Some(serde_json::json!("ordinary string")),
Some(serde_json::json!(format_blob_ref(
&fabro_types::RunBlobId::new(b"missing")
&fabro_types::BlobHash::new(b"missing")
))),
] {
let (handler, calls) = ScriptedHandler::new(Scripted::Succeed);

8
lib/components/fabro-workflow/src/lifecycle/git.rs
View file

@ -613,7 +613,7 @@ mod tests {
use fabro_model::Catalog;
use fabro_store::{Database, EventEnvelope, RunDatabase, RunProjection};
use fabro_types::run_event::{MetadataSnapshotFailureKind, MetadataSnapshotPhase};
use fabro_types::{EventBody, RunBlobId, RunEvent, WorkflowSettings, fixtures, test_support};
use fabro_types::{BlobHash, EventBody, RunEvent, WorkflowSettings, fixtures, test_support};
use object_store::memory::InMemory;
use super::*;
@ -1324,11 +1324,11 @@ mod tests {
Ok(())
}
async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
Ok(RunBlobId::new(data))
async fn write_blob(&self, data: &[u8]) -> Result<BlobHash> {
Ok(BlobHash::new(data))
}
async fn read_blob(&self, _id: &RunBlobId) -> Result<Option<Bytes>> {
async fn read_blob(&self, _id: &BlobHash) -> Result<Option<Bytes>> {
Ok(None)
}

8
lib/components/fabro-workflow/src/operations/retry.rs
View file

@ -117,8 +117,8 @@ mod tests {
use fabro_store::{Database, RunProjectionReducer};
use fabro_types::{
AuthMethod, DirtyStatus, FailureReason, ForkSourceRef, GitContext, Graph, IdpIdentity,
Principal, PullRequestLink, RunBlobId, RunRunnableSource, RunServerProvenance, RunTiming,
AuthMethod, BlobHash, DirtyStatus, FailureReason, ForkSourceRef, GitContext, Graph,
IdpIdentity, Principal, PullRequestLink, RunRunnableSource, RunServerProvenance, RunTiming,
WorkflowSettings, fixtures,
};
use object_store::memory::InMemory;
@ -164,7 +164,7 @@ mod tests {
async fn append_created(
store: &fabro_store::RunDatabase,
run_id: RunId,
manifest_blob: Option<RunBlobId>,
manifest_blob: Option<BlobHash>,
fork_source_ref: Option<ForkSourceRef>,
) {
let mut settings = WorkflowSettings::default();
@ -248,7 +248,7 @@ mod tests {
async fn seed_retryable_failed_source(
store: &Database,
source_run_id: RunId,
) -> (Option<RunBlobId>, Option<RunBlobId>, ForkSourceRef) {
) -> (Option<BlobHash>, Option<BlobHash>, ForkSourceRef) {
let source_store = store.create_run(&source_run_id).await.unwrap();
let manifest_blob = Some(
source_store

2
lib/components/fabro-workflow/src/operations/start.rs
View file

@ -570,7 +570,7 @@ fn vault_token_lookup(vault: &Vault, name: &str) -> Option<String> {
async fn load_accepted_run_definition(
run_store: &RunStoreHandle,
blob_id: fabro_types::RunBlobId,
blob_id: fabro_types::BlobHash,
) -> Result<RunDefinition, Error> {
let bytes = run_store
.read_blob(&blob_id)

8
lib/components/fabro-workflow/src/pipeline/finalize.rs
View file

@ -684,7 +684,7 @@ mod tests {
use fabro_store::{Database, EventEnvelope, RunDatabase, RunProjection};
use fabro_types::run_event::{MetadataSnapshotFailureKind, MetadataSnapshotPhase};
use fabro_types::{
BilledTokenCounts, EventBody, RunBlobId, RunEvent, RunId, RunSpec, StageCompletion,
BilledTokenCounts, BlobHash, EventBody, RunEvent, RunId, RunSpec, StageCompletion,
WorkflowSettings, first_event_seq, fixtures, test_support,
};
use object_store::memory::InMemory;
@ -1819,11 +1819,11 @@ mod tests {
Ok(())
}
async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
Ok(RunBlobId::new(data))
async fn write_blob(&self, data: &[u8]) -> Result<BlobHash> {
Ok(BlobHash::new(data))
}
async fn read_blob(&self, _id: &RunBlobId) -> Result<Option<Bytes>> {
async fn read_blob(&self, _id: &BlobHash) -> Result<Option<Bytes>> {
Ok(None)
}

14
lib/components/fabro-workflow/src/runtime_store.rs
View file

@ -4,7 +4,7 @@ use anyhow::Result;
use async_trait::async_trait;
use bytes::Bytes;
use fabro_store::{EventEnvelope, RunDatabase, RunProjection};
use fabro_types::{RunBlobId, RunEvent};
use fabro_types::{BlobHash, RunEvent};
use crate::event::build_redacted_event_payload;
@ -13,8 +13,8 @@ pub trait RunStoreBackend: Send + Sync {
async fn load_state(&self) -> Result<RunProjection>;
async fn list_events(&self) -> Result<Vec<EventEnvelope>>;
async fn append_run_event(&self, event: &RunEvent) -> Result<()>;
async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId>;
async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>>;
async fn write_blob(&self, data: &[u8]) -> Result<BlobHash>;
async fn read_blob(&self, id: &BlobHash) -> Result<Option<Bytes>>;
async fn read_run_log(&self) -> Result<Option<Vec<u8>>>;
}
@ -46,11 +46,11 @@ impl RunStoreHandle {
self.backend.append_run_event(event).await
}
pub async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
pub async fn write_blob(&self, data: &[u8]) -> Result<BlobHash> {
self.backend.write_blob(data).await
}
pub async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
pub async fn read_blob(&self, id: &BlobHash) -> Result<Option<Bytes>> {
self.backend.read_blob(id).await
}
@ -91,14 +91,14 @@ impl RunStoreBackend for LocalRunStoreBackend {
.map_err(anyhow::Error::from)
}
async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId> {
async fn write_blob(&self, data: &[u8]) -> Result<BlobHash> {
self.run_store
.write_blob(data)
.await
.map_err(anyhow::Error::from)
}
async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
async fn read_blob(&self, id: &BlobHash) -> Result<Option<Bytes>> {
self.run_store
.read_blob(id)
.await

2
lib/components/fabro-workflow/tests/it/daytona_integration.rs
View file

@ -544,7 +544,7 @@ async fn daytona_pipeline_artifact_offload_and_sync() {
.get("response.big_output")
.expect("context should have response.big_output");
let pointer_str = pointer_value.as_str().expect("pointer should be a string");
let expected_blob_id = fabro_types::RunBlobId::new(
let expected_blob_id = fabro_types::BlobHash::new(
&serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024)))
.expect("large value should serialize"),
);

4
lib/components/fabro-workflow/tests/it/integration.rs
View file

@ -10059,7 +10059,7 @@ async fn large_context_values_are_offloaded_to_artifact_store() {
.expect("context should have response.big_output");
let pointer_str = pointer_value.as_str().expect("pointer should be a string");
let expected_blob_id = fabro_types::RunBlobId::new(
let expected_blob_id = fabro_types::BlobHash::new(
&serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024)))
.expect("large value should serialize"),
);
@ -10258,7 +10258,7 @@ async fn artifact_pointers_rewritten_for_remote_sandbox() {
.get("response.big_output")
.expect("context should have response.big_output");
let pointer_str = pointer_value.as_str().expect("pointer should be a string");
let expected_blob_id = fabro_types::RunBlobId::new(
let expected_blob_id = fabro_types::BlobHash::new(
&serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024)))
.expect("large value should serialize"),
);

12
lib/foundation/fabro-client/src/client.rs
View file

@ -13,8 +13,8 @@ use fabro_http::multipart::{Form, Part};
use fabro_model::{Model, ModelTestMode, ProviderId};
use fabro_types::settings::run::MergeStrategy;
use fabro_types::{
ArtifactUpload, EventEnvelope, PairId, PairMessageRecord, PairMessageRequest, PairRecord,
PairStartRequest, PairTranscriptResponse, Run, RunBlobId, RunEvent, RunEventDetailResponse,
ArtifactUpload, BlobHash, EventEnvelope, PairId, PairMessageRecord, PairMessageRequest,
PairRecord, PairStartRequest, PairTranscriptResponse, Run, RunEvent, RunEventDetailResponse,
RunId, RunPairStatusResponse, RunProjection, SessionId, SessionRecord, StageId,
};
use fabro_util::exit::{ErrorExt, ExitClass};
@ -1828,7 +1828,7 @@ impl Client {
u32::try_from(response.into_inner().seq).context("append_run_event returned invalid seq")
}
pub async fn write_run_blob(&self, run_id: &RunId, data: &[u8]) -> Result<RunBlobId> {
pub async fn write_run_blob(&self, run_id: &RunId, data: &[u8]) -> Result<BlobHash> {
let response = self
.send_api(|client| async move {
client
@ -1846,11 +1846,7 @@ impl Client {
.context("write_run_blob returned invalid blob id")
}
pub async fn read_run_blob(
&self,
run_id: &RunId,
blob_id: &RunBlobId,
) -> Result<Option<Bytes>> {
pub async fn read_run_blob(&self, run_id: &RunId, blob_id: &BlobHash) -> Result<Option<Bytes>> {
let response = self
.current_state()
.client

42
lib/foundation/fabro-types/src/run_blob_id.rs → lib/foundation/fabro-types/src/blob_hash.rs
View file

@ -7,9 +7,9 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer};
use sha2::{Digest, Sha256};
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct RunBlobId([u8; 32]);
pub struct BlobHash([u8; 32]);
impl RunBlobId {
impl BlobHash {
pub fn new(content: &[u8]) -> Self {
let hash = Sha256::digest(content);
let mut bytes = [0_u8; 32];
@ -18,13 +18,13 @@ impl RunBlobId {
}
}
impl fmt::Display for RunBlobId {
impl fmt::Display for BlobHash {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&hex::encode(self.0))
}
}
impl FromStr for RunBlobId {
impl FromStr for BlobHash {
type Err = FromHexError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
@ -34,7 +34,7 @@ impl FromStr for RunBlobId {
}
}
impl Serialize for RunBlobId {
impl Serialize for BlobHash {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
@ -43,7 +43,7 @@ impl Serialize for RunBlobId {
}
}
impl<'de> Deserialize<'de> for RunBlobId {
impl<'de> Deserialize<'de> for BlobHash {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
@ -55,44 +55,44 @@ impl<'de> Deserialize<'de> for RunBlobId {
#[cfg(test)]
mod tests {
use crate::RunBlobId;
use crate::BlobHash;
#[test]
fn same_content_produces_same_blob_id() {
assert_eq!(RunBlobId::new(b"hello"), RunBlobId::new(b"hello"));
fn same_content_produces_same_blob_hash() {
assert_eq!(BlobHash::new(b"hello"), BlobHash::new(b"hello"));
}
#[test]
fn display_is_lowercase_sha256_hex() {
assert_eq!(
RunBlobId::new(b"hello").to_string(),
BlobHash::new(b"hello").to_string(),
"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
);
}
#[test]
fn different_content_produces_different_blob_ids() {
assert_ne!(RunBlobId::new(b"hello"), RunBlobId::new(b"world"));
fn different_content_produces_different_blob_hashes() {
assert_ne!(BlobHash::new(b"hello"), BlobHash::new(b"world"));
}
#[test]
fn display_and_parse_round_trip() {
let blob_id = RunBlobId::new(b"hello");
let parsed: RunBlobId = blob_id.to_string().parse().unwrap();
assert_eq!(parsed, blob_id);
let blob_hash = BlobHash::new(b"hello");
let parsed: BlobHash = blob_hash.to_string().parse().unwrap();
assert_eq!(parsed, blob_hash);
}
#[test]
fn serde_round_trip() {
let blob_id = RunBlobId::new(b"hello");
let value = serde_json::to_value(blob_id).unwrap();
let parsed: RunBlobId = serde_json::from_value(value).unwrap();
assert_eq!(parsed, blob_id);
let blob_hash = BlobHash::new(b"hello");
let value = serde_json::to_value(blob_hash).unwrap();
let parsed: BlobHash = serde_json::from_value(value).unwrap();
assert_eq!(parsed, blob_hash);
}
#[test]
fn parse_rejects_non_hex_blob_ids() {
let parsed = "not-a-blob-id".parse::<RunBlobId>();
fn parse_rejects_non_hex_blob_hashes() {
let parsed = "not-a-blob-hash".parse::<BlobHash>();
assert!(parsed.is_err());
}
}

40
lib/foundation/fabro-types/src/blob_ref.rs
View file

@ -1,35 +1,35 @@
use std::path::Path;
use crate::RunBlobId;
use crate::BlobHash;
const BLOB_REF_PREFIX: &str = "blob://sha256/";
#[must_use]
pub fn format_blob_ref(blob_id: &RunBlobId) -> String {
format!("{BLOB_REF_PREFIX}{blob_id}")
pub fn format_blob_ref(blob_hash: &BlobHash) -> String {
format!("{BLOB_REF_PREFIX}{blob_hash}")
}
#[must_use]
pub fn parse_blob_ref(value: &str) -> Option<RunBlobId> {
pub fn parse_blob_ref(value: &str) -> Option<BlobHash> {
value.strip_prefix(BLOB_REF_PREFIX)?.parse().ok()
}
#[must_use]
pub fn parse_managed_blob_file_ref(value: &str) -> Option<RunBlobId> {
pub fn parse_managed_blob_file_ref(value: &str) -> Option<BlobHash> {
let path = value.strip_prefix("file://")?;
let blob_id = parse_blob_file_name(path)?;
let blob_hash = parse_blob_file_name(path)?;
if has_path_suffix(path, &["runtime", "blobs"]) || has_path_suffix(path, &[".fabro", "blobs"]) {
Some(blob_id)
Some(blob_hash)
} else {
None
}
}
fn parse_blob_file_name(path: &str) -> Option<RunBlobId> {
fn parse_blob_file_name(path: &str) -> Option<BlobHash> {
let file_name = Path::new(path).file_name()?.to_str()?;
let blob_id = file_name.strip_suffix(".json")?;
blob_id.parse().ok()
let blob_hash = file_name.strip_suffix(".json")?;
blob_hash.parse().ok()
}
fn has_path_suffix(path: &str, suffix: &[&str]) -> bool {
@ -46,30 +46,30 @@ fn has_path_suffix(path: &str, suffix: &[&str]) -> bool {
#[cfg(test)]
mod tests {
use super::{format_blob_ref, parse_blob_ref, parse_managed_blob_file_ref};
use crate::RunBlobId;
use crate::BlobHash;
#[test]
fn blob_ref_round_trips() {
let blob_id = RunBlobId::new(br#"{"kind":"summary"}"#);
let formatted = format_blob_ref(&blob_id);
let blob_hash = BlobHash::new(br#"{"kind":"summary"}"#);
let formatted = format_blob_ref(&blob_hash);
assert_eq!(parse_blob_ref(&formatted), Some(blob_id));
assert_eq!(parse_blob_ref(&formatted), Some(blob_hash));
}
#[test]
fn managed_local_blob_file_ref_is_recognized() {
let blob_id = RunBlobId::new(b"hello");
let value = format!("file:///tmp/run/runtime/blobs/{blob_id}.json");
let blob_hash = BlobHash::new(b"hello");
let value = format!("file:///tmp/run/runtime/blobs/{blob_hash}.json");
assert_eq!(parse_managed_blob_file_ref(&value), Some(blob_id));
assert_eq!(parse_managed_blob_file_ref(&value), Some(blob_hash));
}
#[test]
fn managed_remote_blob_file_ref_is_recognized() {
let blob_id = RunBlobId::new(b"hello");
let value = format!("file:///sandbox/.fabro/blobs/{blob_id}.json");
let blob_hash = BlobHash::new(b"hello");
let value = format!("file:///sandbox/.fabro/blobs/{blob_hash}.json");
assert_eq!(parse_managed_blob_file_ref(&value), Some(blob_id));
assert_eq!(parse_managed_blob_file_ref(&value), Some(blob_hash));
}
#[test]

4
lib/foundation/fabro-types/src/lib.rs
View file

@ -3,6 +3,7 @@ extern crate self as fabro_types;
pub mod artifact;
pub mod auth;
pub mod billing;
pub mod blob_hash;
pub mod blob_ref;
pub mod checkpoint;
pub mod command_output;
@ -26,7 +27,6 @@ pub mod pull_request;
pub mod reasoning;
pub mod repository;
pub mod run;
pub mod run_blob_id;
pub mod run_event;
pub mod run_failure;
pub mod run_id;
@ -63,6 +63,7 @@ pub use billing::{
ModelBillingFacts, ModelBillingInput, ModelPricing, ModelPricingPolicy, ModelRef, ModelUsage,
OpenAiBillingFacts, OpenAiModelPricing, PricePerMTok, Speed, TokenCounts, UsdMicros,
};
pub use blob_hash::BlobHash;
pub use blob_ref::{format_blob_ref, parse_blob_ref, parse_managed_blob_file_ref};
pub use checkpoint::Checkpoint;
pub use command_output::{CommandOutputStream, CommandTermination};
@ -113,7 +114,6 @@ pub use run::{
DirtyStatus, ForkSourceRef, GitContext, RunClientProvenance, RunProvenance,
RunServerProvenance, RunSpec,
};
pub use run_blob_id::RunBlobId;
pub use run_event::{
AgentMcpToolSummary, AgentMemoryFileProps, AgentSkillActivationSource, AgentSkillSummary,
AgentToolCategory, AgentToolSource, AgentToolSummary, AgentToolsAvailableProps, EventBody,

6
lib/foundation/fabro-types/src/run.rs
View file

@ -3,9 +3,9 @@ use std::collections::HashMap;
use serde::{Deserialize, Serialize};
use crate::WorkflowSettings;
use crate::blob_hash::BlobHash;
use crate::graph::Graph;
use crate::principal::Principal;
use crate::run_blob_id::RunBlobId;
use crate::run_id::RunId;
use crate::run_summary::AutomationRef;
@ -73,9 +73,9 @@ pub struct RunSpec {
pub labels: HashMap<String, String>,
pub provenance: RunProvenance,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub manifest_blob: Option<RunBlobId>,
pub manifest_blob: Option<BlobHash>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub definition_blob: Option<RunBlobId>,
pub definition_blob: Option<BlobHash>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub git: Option<GitContext>,
#[serde(default, skip_serializing_if = "Option::is_none")]

6
lib/foundation/fabro-types/src/run_event/mod.rs
View file

@ -937,7 +937,7 @@ mod tests {
use super::*;
use crate::{
AuthMethod, CommandTermination, Edge, Graph, IdpIdentity, Node, PendingReason, RunBlobId,
AuthMethod, BlobHash, CommandTermination, Edge, Graph, IdpIdentity, Node, PendingReason,
WorkflowSettings, fixtures, test_support,
};
@ -1059,7 +1059,7 @@ mod tests {
"labels": {},
"source_directory": "/tmp/run",
"provenance": test_support::test_run_provenance(),
"manifest_blob": RunBlobId::new(br#"{"version":1}"#).to_string()
"manifest_blob": BlobHash::new(br#"{"version":1}"#).to_string()
}
});
@ -1337,7 +1337,7 @@ mod tests {
"run_id": fixtures::RUN_1,
"event": "run.submitted",
"properties": {
"definition_blob": RunBlobId::new(br#"{"workflow_path":"workflow.fabro"}"#).to_string()
"definition_blob": BlobHash::new(br#"{"workflow_path":"workflow.fabro"}"#).to_string()
}
});

6
lib/foundation/fabro-types/src/run_event/run.rs
View file

@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
use super::{BilledTokenCounts, ExecOutputTail, RunNoticeLevel};
use crate::status::{BlockedReason, PendingReason, SuccessReason};
use crate::{
AutomationRef, DiffSummary, ForkSourceRef, GitContext, Graph, PairId, PairTarget, RunBlobId,
AutomationRef, BlobHash, DiffSummary, ForkSourceRef, GitContext, Graph, PairId, PairTarget,
RunControlAction, RunFailure, RunId, RunProvenance, RunTiming, WorkflowSettings,
};
@ -27,7 +27,7 @@ pub struct RunCreatedProps {
pub automation: Option<AutomationRef>,
pub provenance: RunProvenance,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub manifest_blob: Option<RunBlobId>,
pub manifest_blob: Option<BlobHash>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub git: Option<GitContext>,
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -132,7 +132,7 @@ pub struct RunPairFailedProps {
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunSubmittedProps {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub definition_blob: Option<RunBlobId>,
pub definition_blob: Option<BlobHash>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]