Unify payload handling with PayloadBuilder and payload manifest

Remove mime_type/payload_size from Observation struct since these are
per-payload concerns, not observation-level metadata. All payloads now
flow through a single unified path instead of separate primary and
additional payload code paths.

Key changes:
- Add PayloadBuilder for creating named payloads in shared crate
- Replace ObservationWithPayload with separate Observation + Payload
  messages in the uploader channel
- Send payload_manifest JSON in multipart uploads so the server gets
  authoritative MIME types (falls back to heuristic for old clients)
- Add mime_type/size to GetPayload API response for per-payload metadata
- Fix named_payload sending payload twice (as both primary and additional)
- Update templates to derive type/size from per-payload data

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: Dig-Doug

crates/observation-tools-client/openapi.json

@@ -116,13 +116,22 @@
           "id": {
             "$ref": "#/components/schemas/PayloadId"
           },
+          "mime_type": {
+            "type": "string"
+          },
           "name": {
             "type": "string"
+          },
+          "size": {
+            "minimum": 0,
+            "type": "integer"
           }
         },
         "required": [
           "id",
           "name",
+          "mime_type",
+          "size",
           "data"
         ],
         "type": "object"
@@ -222,10 +231,6 @@
             },
             "type": "object"
           },
-          "mime_type": {
-            "description": "MIME type of the payload (e.g., \"text/plain\", \"application/json\")",
-            "type": "string"
-          },
           "name": {
             "description": "User-defined name for this observation",
             "type": "string"
@@ -244,11 +249,6 @@
             "nullable": true,
             "type": "string"
           },
-          "payload_size": {
-            "description": "Size of the payload in bytes",
-            "minimum": 0,
-            "type": "integer"
-          },
           "source": {
             "$ref": "#/components/schemas/SourceInfo",
             "description": "Source location where this observation was created",
@@ -261,9 +261,7 @@
           "name",
           "observation_type",
           "log_level",
-          "created_at",
-          "mime_type",
-          "payload_size"
+          "created_at"
         ],
         "type": "object"
       },

crates/observation-tools-client/src/client.rs

@@ -4,13 +4,13 @@ use crate::error::Result;
 use crate::execution::BeginExecution;
 use crate::execution::ExecutionHandle;
 use crate::observation_handle::ObservationHandle;
-use crate::ObservationWithPayload;
 use async_channel;
 use log::error;
 use log::info;
 use log::trace;
 use napi_derive::napi;
 use observation_tools_shared::models::Execution;
+use observation_tools_shared::Observation;
 // Re-export constants from shared crate for convenience
 pub use observation_tools_shared::BATCH_SIZE;
 pub use observation_tools_shared::BLOB_THRESHOLD_BYTES;
@@ -31,12 +31,12 @@ pub(crate) enum UploaderMessage {
     handle: ExecutionHandle,
     uploaded_tx: tokio::sync::watch::Sender<ExecutionUploadResult>,
   },
-  Observations {
-    observations: Vec<ObservationWithPayload>,
+  Observation {
+    observation: Observation,
     handle: ObservationHandle,
     uploaded_tx: tokio::sync::watch::Sender<ObservationUploadResult>,
   },
-  AdditionalPayload {
+  Payload {
     observation_id: observation_tools_shared::ObservationId,
     execution_id: observation_tools_shared::models::ExecutionId,
     payload_id: observation_tools_shared::PayloadId,
@@ -55,22 +55,22 @@ impl std::fmt::Debug for UploaderMessage {
         .debug_struct("Execution")
         .field("execution", execution)
         .finish(),
-      Self::Observations {
-        observations,
+      Self::Observation {
+        observation,
         handle,
         ..
       } => f
-        .debug_struct("Observations")
-        .field("observations", observations)
+        .debug_struct("Observation")
+        .field("observation", observation)
         .field("handle", handle)
         .finish(),
-      Self::AdditionalPayload {
+      Self::Payload {
         observation_id,
         payload_id,
         name,
         ..
       } => f
-        .debug_struct("AdditionalPayload")
+        .debug_struct("Payload")
         .field("observation_id", observation_id)
         .field("payload_id", payload_id)
         .field("name", name)
@@ -297,14 +297,16 @@ impl ClientBuilder {
   }
 }
 
-/// Data for an additional named payload to be sent alongside observations
+/// Data for a payload ready to be uploaded
 #[derive(Debug)]
-pub(crate) struct AdditionalPayloadData {
+pub(crate) struct PayloadUploadData {
   pub(crate) observation_id: observation_tools_shared::ObservationId,
   pub(crate) execution_id: observation_tools_shared::models::ExecutionId,
   pub(crate) payload_id: observation_tools_shared::PayloadId,
   pub(crate) name: String,
-  pub(crate) payload: observation_tools_shared::Payload,
+  pub(crate) mime_type: String,
+  pub(crate) size: usize,
+  pub(crate) data: Vec<u8>,
 }
 
 async fn uploader_task(
@@ -319,14 +321,14 @@ async fn uploader_task(
     tokio::sync::watch::Sender<ObservationUploadResult>,
   );
 
-  let flush = async |buffer: &mut Vec<ObservationWithPayload>,
+  let flush = async |observation_buffer: &mut Vec<Observation>,
                      senders: &mut Vec<ObservationSender>,
-                     additional_payloads: &mut Vec<AdditionalPayloadData>| {
-    if buffer.is_empty() && additional_payloads.is_empty() {
+                     payload_buffer: &mut Vec<PayloadUploadData>| {
+    if observation_buffer.is_empty() && payload_buffer.is_empty() {
       return;
     }
     let result =
-      upload_observations(&api_client, buffer.drain(..).collect(), additional_payloads.drain(..).collect()).await;
+      upload_observations(&api_client, observation_buffer.drain(..).collect(), payload_buffer.drain(..).collect()).await;
     match result {
       Ok(()) => {
         // Signal all senders that observations were uploaded successfully
@@ -344,9 +346,9 @@ async fn uploader_task(
       }
     }
   };
-  let mut observation_buffer: Vec<ObservationWithPayload> = Vec::new();
+  let mut observation_buffer: Vec<Observation> = Vec::new();
   let mut sender_buffer: Vec<ObservationSender> = Vec::new();
-  let mut additional_payload_buffer: Vec<AdditionalPayloadData> = Vec::new();
+  let mut payload_buffer: Vec<PayloadUploadData> = Vec::new();
   loop {
     let msg = rx.recv().await.ok();
     match msg {
@@ -358,7 +360,6 @@ async fn uploader_task(
         let result = upload_execution(&api_client, execution).await;
         match result {
           Ok(()) => {
-            // Signal successful upload with handle
             let _ = uploaded_tx.send(Some(Ok(handle)));
           }
           Err(e) => {
@@ -368,37 +369,36 @@ async fn uploader_task(
           }
         }
       }
-      Some(UploaderMessage::Observations {
-        observations,
+      Some(UploaderMessage::Observation {
+        observation,
         handle,
         uploaded_tx,
       }) => {
-        observation_buffer.extend(observations);
+        observation_buffer.push(observation);
         sender_buffer.push((handle, uploaded_tx));
-        if observation_buffer.len() >= BATCH_SIZE {
-          flush(&mut observation_buffer, &mut sender_buffer, &mut additional_payload_buffer).await;
-        }
       }
-      Some(UploaderMessage::AdditionalPayload {
+      Some(UploaderMessage::Payload {
         observation_id,
         execution_id,
         payload_id,
         name,
         payload,
       }) => {
-        additional_payload_buffer.push(AdditionalPayloadData {
+        payload_buffer.push(PayloadUploadData {
           observation_id,
           execution_id,
           payload_id,
           name,
-          payload,
+          mime_type: payload.mime_type,
+          size: payload.size,
+          data: payload.data,
         });
       }
       Some(UploaderMessage::Flush) => {
-        flush(&mut observation_buffer, &mut sender_buffer, &mut additional_payload_buffer).await;
+        flush(&mut observation_buffer, &mut sender_buffer, &mut payload_buffer).await;
       }
       Some(UploaderMessage::Shutdown) | None => {
-        flush(&mut observation_buffer, &mut sender_buffer, &mut additional_payload_buffer).await;
+        flush(&mut observation_buffer, &mut sender_buffer, &mut payload_buffer).await;
         break;
       }
     }
@@ -429,10 +429,10 @@ async fn upload_execution(
 
 async fn upload_observations(
   client: &crate::server_client::Client,
-  observations: Vec<ObservationWithPayload>,
-  additional_payloads: Vec<AdditionalPayloadData>,
+  observations: Vec<Observation>,
+  payloads: Vec<PayloadUploadData>,
 ) -> Result<()> {
-  if observations.is_empty() && additional_payloads.is_empty() {
+  if observations.is_empty() && payloads.is_empty() {
     return Ok(());
   }
 
@@ -441,30 +441,30 @@ async fn upload_observations(
     std::collections::HashMap::new();
   for obs in observations {
     by_execution
-      .entry(obs.observation.execution_id)
+      .entry(obs.execution_id)
       .or_default()
       .0
       .push(obs);
   }
-  for ap in additional_payloads {
+  for p in payloads {
     by_execution
-      .entry(ap.execution_id)
+      .entry(p.execution_id)
       .or_default()
       .1
-      .push(ap);
+      .push(p);
   }
 
   // Upload each batch via multipart form
-  for (execution_id, (observations, additional_payloads)) in by_execution {
+  for (execution_id, (observations, payloads)) in by_execution {
     trace!(
-      "Uploading {} observations + {} additional payloads for execution {}",
+      "Uploading {} observations + {} payloads for execution {}",
       observations.len(),
-      additional_payloads.len(),
+      payloads.len(),
       execution_id
     );
 
     client
-      .create_observations_multipart(&execution_id.to_string(), observations, additional_payloads)
+      .create_observations_multipart(&execution_id.to_string(), observations, payloads)
       .await
       .map_err(|e| crate::error::Error::Config(e.to_string()))?;
   }