Add blob upload: PUT /graphs/{id}/blob and omnigraph blob put

crates/omnigraph-api-types/Cargo.toml

@@ -13,5 +13,4 @@ omnigraph = { package = "omnigraph-engine", path = "../omnigraph", version = "0.
 omnigraph-compiler = { path = "../omnigraph-compiler", version = "0.8.1" }
 serde = { workspace = true }
 serde_json = { workspace = true }
-sha2 = { workspace = true }
 utoipa = { workspace = true }

crates/omnigraph-api-types/src/lib.rs

@@ -569,6 +569,23 @@ pub struct BlobQuery {
     pub snapshot: Option<String>,
 }
 
+/// Response body for `PUT /blob` (raw-bytes blob upload). Also printed
+/// verbatim by the CLI's `blob put` so both arms emit the identical object.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct BlobPutOutput {
+    #[serde(rename = "type")]
+    pub type_name: String,
+    pub id: String,
+    pub prop: String,
+    pub branch: String,
+    /// Stored payload size in bytes.
+    pub size: u64,
+    /// Strong quoted validator of the committed cell — identical to what a
+    /// follow-up `GET /blob` serves.
+    pub etag: String,
+    pub actor_id: Option<String>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct HealthOutput {
     pub status: String,
@@ -802,19 +819,30 @@ pub struct GraphListResponse {
 /// the server's `/blob` responses and the CLI's embedded `blob stat` so
 /// both arms derive the identical validator from one definition.
 pub fn blob_etag(tag: &omnigraph::db::BlobVersionTag, property: &str) -> String {
-    use sha2::{Digest, Sha256};
-    let mut hasher = Sha256::new();
-    hasher.update(tag.stable_table_id.to_le_bytes());
-    hasher.update(tag.table_incarnation_id.to_le_bytes());
-    hasher.update(tag.table_version.to_le_bytes());
-    hasher.update(tag.row_id.to_le_bytes());
-    hasher.update(property.as_bytes());
-    let digest = hasher.finalize();
-    let mut out = String::with_capacity(34);
-    out.push('"');
-    for byte in &digest[..16] {
-        out.push_str(&format!("{byte:02x}"));
+    omnigraph::db::blob_etag(tag, property)
+}
+
+/// RFC 9110 `If-Match` header -> the engine's transport-neutral
+/// [`omnigraph::db::BlobPrecondition`]. Shared by the server's `PUT /blob`
+/// handler and the CLI's embedded `blob put` arm so both boundaries parse
+/// the wire form identically (the engine itself compares opaque tokens
+/// only). `*` is `AnyExisting`; otherwise a comma-separated entity-tag
+/// list. If-Match requires STRONG comparison (RFC 9110 §13.1.1), so a weak
+/// `W/"..."` tag never matches and is dropped here — a header of only weak
+/// tags yields `Tags(vec![])`, which matches nothing (412). Splitting on
+/// commas is exact for the hex tags this server mints; an exotic quoted
+/// comma only costs a spurious 412, never a wrong success.
+pub fn parse_if_match(header: &str) -> omnigraph::db::BlobPrecondition {
+    let header = header.trim();
+    if header == "*" {
+        return omnigraph::db::BlobPrecondition::AnyExisting;
     }
-    out.push('"');
-    out
+    omnigraph::db::BlobPrecondition::Tags(
+        header
+            .split(',')
+            .map(str::trim)
+            .filter(|tag| !tag.is_empty() && !tag.starts_with("W/"))
+            .map(str::to_string)
+            .collect(),
+    )
 }

crates/omnigraph-cli/src/cli.rs

@@ -15,7 +15,7 @@ pub(crate) const DEFAULT_BEARER_TOKEN_ENV: &str = "OMNIGRAPH_BEARER_TOKEN";
 #[command(after_help = "\
 COMMANDS BY CAPABILITY:\n  \
 any — run against a graph, served (--server / --profile) or embedded (--store / a \
-URI): query, mutate, load, branch, blob get/stat, snapshot, export, commit, schema show/apply.\n  \
+URI): query, mutate, load, branch, blob get/put/stat, snapshot, export, commit, schema show/apply.\n  \
 served — require a server: graphs.\n  \
 direct — direct storage access; reject --server (init, optimize, repair, cleanup, \
 schema plan, lint).\n  \
@@ -514,6 +514,31 @@ pub(crate) enum BlobCommand {
         #[arg(long)]
         length: Option<u64>,
     },
+    /// Upload raw bytes into one existing row's Blob property
+    Put {
+        /// Node type name
+        #[arg(value_name = "TYPE")]
+        type_name: String,
+        /// Logical row id (the @key value)
+        id: String,
+        /// Blob property name
+        prop: String,
+        /// Graph URI
+        #[arg(long)]
+        uri: Option<String>,
+        /// Branch to write to (defaults to main)
+        #[arg(long)]
+        branch: Option<String>,
+        /// Read the payload from PATH instead of stdin
+        #[arg(long, value_name = "PATH")]
+        file: Option<PathBuf>,
+        /// Conditional write: entity-tag list or `*` (RFC 9110 If-Match,
+        /// strong comparison); a stale validator fails with the current one
+        #[arg(long, value_name = "ETAG")]
+        if_match: Option<String>,
+        #[arg(long)]
+        json: bool,
+    },
     /// Show one blob cell's metadata (kind, size, etag, external uri)
     Stat {
         /// Node type name

crates/omnigraph-cli/src/client.rs

@@ -27,7 +27,7 @@ use color_eyre::Result;
 use color_eyre::eyre::bail;
 use omnigraph::db::{BlobContent, Omnigraph, ReadTarget};
 use omnigraph_api_types::{
-    BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, BranchListOutput,
+    BlobPutOutput, BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, BranchListOutput,
     BranchMergeOutput, BranchMergeRequest, ChangeOutput, CommitListOutput, CommitOutput,
     ErrorOutput, ExportRequest, GraphListResponse, IngestOutput, IngestRequest,
     InvokeStoredQueryRequest, ReadOutput,
@@ -124,6 +124,15 @@ fn blob_target_param(target: &ReadTarget) -> (&'static str, String) {
     }
 }
 
+/// The one typed stale-precondition error BOTH blob-put arms return, so the
+/// remote 412 and the embedded typed outcome are indistinguishable at the
+/// CLI surface (pinned by the parity matrix).
+fn blob_precondition_error(current_etag: &str) -> color_eyre::Report {
+    color_eyre::eyre::eyre!(
+        "blob precondition failed: If-Match did not match the cell's current validator {current_etag}"
+    )
+}
+
 /// The `Location` of a `302` blob answer. A redirect without a location is
 /// a server contract violation, surfaced loudly rather than fabricated.
 fn blob_redirect_location(response: &reqwest::Response) -> Result<String> {
@@ -1049,6 +1058,93 @@ impl GraphClient {
         }
     }
 
+    /// `blob put` — upload raw payload bytes into one existing row's Blob
+    /// cell. Update-only; an optional `--if-match` validator (raw RFC 9110
+    /// header form, parsed HERE at the boundary) makes the write conditional.
+    /// Both arms surface a stale precondition through the SAME typed error
+    /// text carrying the current validator, so exit codes and messages hold
+    /// parity (remote derives it from the 412's ETag header, embedded from
+    /// the engine's typed outcome).
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) async fn blob_put(
+        &self,
+        branch: &str,
+        type_name: &str,
+        id: &str,
+        prop: &str,
+        bytes: Vec<u8>,
+        if_match: Option<&str>,
+    ) -> Result<BlobPutOutput> {
+        match self {
+            GraphClient::Remote {
+                http,
+                base_url,
+                token,
+            } => {
+                let url = remote_url(
+                    base_url,
+                    &["blob"],
+                    &[
+                        ("type", type_name),
+                        ("id", id),
+                        ("prop", prop),
+                        ("branch", branch),
+                    ],
+                )?;
+                let mut request = apply_bearer_token(http.request(Method::PUT, url), token.as_deref())
+                    .body(bytes);
+                if let Some(validator) = if_match {
+                    request = request.header(reqwest::header::IF_MATCH, validator);
+                }
+                let response = request.send().await?;
+                let status = response.status();
+                if status == reqwest::StatusCode::PRECONDITION_FAILED {
+                    let current = response
+                        .headers()
+                        .get(reqwest::header::ETAG)
+                        .and_then(|value| value.to_str().ok())
+                        .unwrap_or("<unknown>")
+                        .to_string();
+                    return Err(blob_precondition_error(&current));
+                }
+                if !status.is_success() {
+                    let text = response.text().await?;
+                    if let Ok(error) = serde_json::from_str::<ErrorOutput>(&text) {
+                        bail!(error.error);
+                    }
+                    bail!("server returned {}: {}", status, text);
+                }
+                Ok(response.json::<BlobPutOutput>().await?)
+            }
+            GraphClient::Embedded { uri, actor } => {
+                let db = Self::open_embedded(uri).await?;
+                let actor = actor.as_deref();
+                let precondition = if_match.map(omnigraph_api_types::parse_if_match);
+                let size_hint = bytes.len() as u64;
+                let outcome = db
+                    .write_blob_at_as(branch, type_name, id, prop, bytes, precondition, actor)
+                    .await?;
+                match outcome {
+                    omnigraph::db::BlobWriteOutcome::Written { etag, size } => {
+                        debug_assert_eq!(size, size_hint);
+                        Ok(BlobPutOutput {
+                            type_name: type_name.to_string(),
+                            id: id.to_string(),
+                            prop: prop.to_string(),
+                            branch: branch.to_string(),
+                            size,
+                            etag,
+                            actor_id: actor.map(String::from),
+                        })
+                    }
+                    omnigraph::db::BlobWriteOutcome::PreconditionFailed { current_etag } => {
+                        Err(blob_precondition_error(&current_etag))
+                    }
+                }
+            }
+        }
+    }
+
     /// `graphs list` — enumerate the graphs a remote multi-graph server
     /// serves (`GET /graphs`). Remote-only by design: there is no local
     /// enumeration endpoint, so the Embedded arm fails loudly. Routing it

crates/omnigraph-cli/src/main.rs

@@ -714,6 +714,47 @@ async fn main() -> Result<()> {
                     }
                 }
             }
+            BlobCommand::Put {
+                type_name,
+                id,
+                prop,
+                uri,
+                branch,
+                file,
+                if_match,
+                json,
+            } => {
+                let client = client::GraphClient::resolve_with_policy(
+                    cli.server.as_deref(),
+                    cli.graph.as_deref(),
+                    uri,
+                    cli.as_actor.as_deref(),
+                    cli.profile.as_deref(),
+                    cli.store.as_deref(),
+                )
+                .await?;
+                let branch = resolve_branch(branch, None, "main");
+                let bytes = match &file {
+                    Some(path) => std::fs::read(path)?,
+                    None => {
+                        let mut buffer = Vec::new();
+                        std::io::Read::read_to_end(&mut std::io::stdin().lock(), &mut buffer)?;
+                        buffer
+                    }
+                };
+                let output = client
+                    .blob_put(&branch, &type_name, &id, &prop, bytes, if_match.as_deref())
+                    .await?;
+                if json {
+                    print_json(&output)?;
+                } else {
+                    println!(
+                        "uploaded {} bytes to {} {} {} on {}",
+                        output.size, output.type_name, output.id, output.prop, output.branch
+                    );
+                    println!("etag: {}", output.etag);
+                }
+            }
             BlobCommand::Stat {
                 type_name,
                 id,

crates/omnigraph-cli/src/planes.rs

@@ -152,6 +152,7 @@ pub(crate) fn command_label(cmd: &Command) -> &'static str {
         Command::Branch { .. } => "branch",
         Command::Blob { command } => match command {
             BlobCommand::Get { .. } => "blob get",
+            BlobCommand::Put { .. } => "blob put",
             BlobCommand::Stat { .. } => "blob stat",
         },
         Command::Schema { command } => match command {

crates/omnigraph-cli/tests/cli_data.rs

@@ -2561,3 +2561,89 @@ fn blob_branch_and_snapshot_conflict() {
         "clap must reject the exclusive pair: {stderr}"
     );
 }
+
+#[test]
+fn blob_put_uploads_file_and_round_trips() {
+    let (temp, graph, _) = blob_graph();
+    let payload_path = temp.path().join("payload.bin");
+    fs::write(&payload_path, b"uploaded via file").unwrap();
+
+    let output = output_success(
+        blob_args(&mut cli(), "put", "readme", &graph)
+            .arg("--file")
+            .arg(&payload_path)
+            .arg("--json"),
+    );
+    let payload = parse_stdout_json(&output);
+    assert_eq!(payload["type"], "Document");
+    assert_eq!(payload["id"], "readme");
+    assert_eq!(payload["prop"], "content");
+    assert_eq!(payload["branch"], "main");
+    assert_eq!(payload["size"], b"uploaded via file".len() as u64);
+    let put_etag = payload["etag"].as_str().unwrap().to_string();
+
+    let get = output_success(blob_args(&mut cli(), "get", "readme", &graph));
+    assert_eq!(&get.stdout[..], b"uploaded via file");
+    // The stat validator equals the one the put reported.
+    let stat = output_success(blob_args(&mut cli(), "stat", "readme", &graph).arg("--json"));
+    assert_eq!(parse_stdout_json(&stat)["etag"].as_str().unwrap(), put_etag);
+}
+
+#[test]
+fn blob_put_uploads_from_stdin() {
+    let (_temp, graph, _) = blob_graph();
+    let output = output_success(
+        blob_args(&mut cli(), "put", "readme", &graph).write_stdin("streamed from stdin"),
+    );
+    assert!(String::from_utf8_lossy(&output.stdout).contains("uploaded 19 bytes"));
+    let get = output_success(blob_args(&mut cli(), "get", "readme", &graph));
+    assert_eq!(&get.stdout[..], b"streamed from stdin");
+}
+
+#[test]
+fn blob_put_is_update_only() {
+    let (_temp, graph, _) = blob_graph();
+    let output = output_failure(blob_args(&mut cli(), "put", "absent", &graph).write_stdin("x"));
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(
+        stderr.contains("no Document with id 'absent'"),
+        "update-only put must fail a missing row: {stderr}"
+    );
+}
+
+#[test]
+fn blob_put_if_match_mismatch_fails_with_current_validator() {
+    let (_temp, graph, _) = blob_graph();
+    let stat = output_success(blob_args(&mut cli(), "stat", "readme", &graph).arg("--json"));
+    let fresh = parse_stdout_json(&stat)["etag"].as_str().unwrap().to_string();
+
+    // Fresh validator lands.
+    output_success(
+        blob_args(&mut cli(), "put", "readme", &graph)
+            .arg("--if-match")
+            .arg(&fresh)
+            .write_stdin("v2"),
+    );
+
+    // The now-stale validator fails, naming the current one.
+    let current = parse_stdout_json(&output_success(
+        blob_args(&mut cli(), "stat", "readme", &graph).arg("--json"),
+    ))["etag"]
+        .as_str()
+        .unwrap()
+        .to_string();
+    let output = output_failure(
+        blob_args(&mut cli(), "put", "readme", &graph)
+            .arg("--if-match")
+            .arg(&fresh)
+            .write_stdin("must not land"),
+    );
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(stderr.contains("blob precondition failed"), "{stderr}");
+    assert!(
+        stderr.contains(&current),
+        "the error must carry the current validator: {stderr}"
+    );
+    let get = output_success(blob_args(&mut cli(), "get", "readme", &graph));
+    assert_eq!(&get.stdout[..], b"v2", "a failed precondition must write nothing");
+}