Replace buffered dataset spans with typed resource writes

src/api/client.rs

@@ -62,19 +62,21 @@ impl Client {
         })
     }
     /// Reuse the SDK's configured transport without copying session state into
-    /// the API layer. This scope is immutable and used only for ingestion.
-    pub(crate) fn for_ingestion(
+    /// the API layer. Credentials and endpoint routes stay immutable.
+    pub(crate) fn for_scope(
         http: reqwest::Client,
         api_url: &Url,
+        app_url: &Url,
         token: &str,
         org_name: Option<&str>,
     ) -> Result<Self, Error> {
         let api_url = base_url(api_url.as_str())?;
+        let app_url = base_url(app_url.as_str())?;
         let headers = scoped_headers(token, org_name)?;
         Ok(Self {
             inner: Arc::new(Inner {
                 http,
-                app_url: api_url.clone(),
+                app_url,
                 api_url,
                 headers,
             }),

src/api/datasets.rs

@@ -1,4 +1,4 @@
-//! Dataset registration and summary endpoint contracts.
+//! Dataset registration, row mutation, and summary endpoint contracts.
 use super::{Client, Error, Plane};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
@@ -36,6 +36,76 @@ pub struct DatasetInfo {
     pub name: String,
 }
 
+/// Fields stored in a dataset record. Absent fields are omitted from the wire.
+#[derive(Debug, Clone, Default, Serialize)]
+pub struct DatasetFields {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub input: Option<Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub expected: Option<Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub metadata: Option<Map<String, Value>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub tags: Option<Vec<String>>,
+}
+
+/// A dataset row mutation, independent of tracing/span payloads.
+#[derive(Debug, Clone)]
+pub enum DatasetWrite {
+    Insert {
+        dataset_id: String,
+        id: String,
+        fields: DatasetFields,
+    },
+    /// Merge only the supplied fields into an existing record.
+    Update {
+        dataset_id: String,
+        id: String,
+        fields: DatasetFields,
+    },
+    Delete {
+        dataset_id: String,
+        id: String,
+    },
+}
+
+impl Serialize for DatasetWrite {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        #[derive(Serialize)]
+        struct Row<'a> {
+            dataset_id: &'a str,
+            id: &'a str,
+            #[serde(flatten)]
+            fields: Option<&'a DatasetFields>,
+            #[serde(rename = "_is_merge", skip_serializing_if = "Option::is_none")]
+            is_merge: Option<bool>,
+            #[serde(rename = "_object_delete", skip_serializing_if = "Option::is_none")]
+            is_delete: Option<bool>,
+        }
+        let (dataset_id, id, fields, is_merge, is_delete) = match self {
+            Self::Insert {
+                dataset_id,
+                id,
+                fields,
+            } => (dataset_id, id, Some(fields), None, None),
+            Self::Update {
+                dataset_id,
+                id,
+                fields,
+            } => (dataset_id, id, Some(fields), Some(true), None),
+            Self::Delete { dataset_id, id } => (dataset_id, id, None, None, Some(true)),
+        };
+        Row {
+            dataset_id,
+            id,
+            fields,
+            is_merge,
+            is_delete,
+        }
+        .serialize(serializer)
+    }
+}
+
 /// Response from dataset summary API.
 #[derive(Debug, Clone, Deserialize)]
 pub struct DatasetSummaryResponse {
@@ -68,6 +138,26 @@ impl Datasets {
             .post_json(Plane::Control, "api/dataset/register", &request)
             .await
     }
+    /// Send exactly one typed dataset mutation request on the data plane.
+    /// No retry, buffering, splitting, or overflow upload is performed.
+    pub async fn write(&self, rows: &[DatasetWrite]) -> Result<(), Error> {
+        #[derive(Serialize)]
+        struct Request<'a> {
+            rows: &'a [DatasetWrite],
+            api_version: u8,
+        }
+        self.client
+            .post_empty(
+                Plane::Data,
+                "logs3",
+                &Request {
+                    rows,
+                    api_version: crate::types::LOGS_API_VERSION,
+                },
+            )
+            .await
+    }
+
     pub async fn summarize(&self, dataset_id: &str) -> Result<DatasetSummaryResponse, Error> {
         self.client
             .get_json(