Add observation hierarchies with groups and multi-part payloads

Replace flat string labels with first-class Group types that form hierarchical
containers for observations. Groups are observations themselves (ObservationType::Group)
with parent pointers, enabling tree structures like request/response hierarchies.

Add multi-part payload support via ObservationPayloadHandle, allowing observations
to accumulate named payloads over time (e.g., request then response).

Key changes:
- GroupId type, GroupBuilder, GroupHandle, SendGroup in shared/client
- group! macro for creating groups
- ObservationBuilder: .label()/.labels() replaced with .group(&GroupHandle)
- Named payload methods: .named_payload(), .named_debug(), .named_raw_payload()
- AdditionalPayload uploader message for streaming named payloads
- Tracing layer uses GroupHandle::from_id with deterministic span ID mapping
- Axum request observer creates group hierarchies per-request
- All tests updated, server templates updated

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

Author: Dig-Doug

crates/observation-tools-client/openapi.json

@@ -105,6 +105,11 @@
         ],
         "type": "object"
       },
+      "GroupId": {
+        "description": "Unique identifier for a group (UUIDv7)",
+        "example": "018e9a3a2c1b7e3f8d2a4b5c6d7e8f9b",
+        "type": "string"
+      },
       "ListExecutionsResponse": {
         "description": "Response for listing executions",
         "properties": {
@@ -170,17 +175,17 @@
             "$ref": "#/components/schemas/ExecutionId",
             "description": "ID of the execution this observation belongs to"
           },
-          "id": {
-            "$ref": "#/components/schemas/ObservationId",
-            "description": "Unique identifier for this observation"
-          },
-          "labels": {
-            "description": "Hierarchical labels for grouping observations\nUses path convention (e.g., \"api/request/headers\")",
+          "group_ids": {
+            "description": "IDs of groups this observation belongs to",
             "items": {
-              "type": "string"
+              "$ref": "#/components/schemas/GroupId"
             },
             "type": "array"
           },
+          "id": {
+            "$ref": "#/components/schemas/ObservationId",
+            "description": "Unique identifier for this observation"
+          },
           "log_level": {
             "$ref": "#/components/schemas/LogLevel",
             "description": "Log level for this observation"
@@ -207,6 +212,11 @@
             "$ref": "#/components/schemas/ObservationType",
             "description": "Type of observation"
           },
+          "parent_group_id": {
+            "$ref": "#/components/schemas/GroupId",
+            "description": "Parent group ID (used when observation_type == Group)",
+            "nullable": true
+          },
           "parent_span_id": {
             "description": "Parent span ID (for tracing integration)",
             "nullable": true,
@@ -245,7 +255,8 @@
         "enum": [
           "LogEntry",
           "Payload",
-          "Span"
+          "Span",
+          "Group"
         ],
         "type": "string"
       },

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

@@ -1,5 +1,7 @@
 use crate::context;
 use crate::execution::ExecutionHandle;
+use crate::group::GroupBuilder;
+use crate::group::GroupHandle;
 use crate::observation::ObservationBuilder;
 use axum::body::Body;
 use axum::extract::Request;
@@ -37,6 +39,8 @@ struct StreamingObserverState {
   log_level: LogLevel,
   status_code: u16,
   execution: ExecutionHandle,
+  response_group: GroupHandle,
+  response_body_group: GroupHandle,
 }
 
 impl StreamingObserverState {
@@ -45,13 +49,17 @@ impl StreamingObserverState {
     log_level: LogLevel,
     status_code: u16,
     execution: ExecutionHandle,
+    response_group: GroupHandle,
+    response_body_group: GroupHandle,
   ) -> Self {
     Self {
       buffer: BytesMut::new(),
       content_type,
       log_level,
       status_code,
       execution,
+      response_group,
+      response_body_group,
     }
   }
 
@@ -74,8 +82,8 @@ impl Drop for StreamingObserverState {
     };
 
     ObservationBuilder::new("http/response/body")
-      .label("http/response")
-      .label("http/response/body")
+      .group(&self.response_group)
+      .group(&self.response_body_group)
       .metadata("status", &self.status_code.to_string())
       .log_level(self.log_level)
       .payload_with_execution(payload, &self.execution);
@@ -99,6 +107,8 @@ impl StreamingObserverBody {
     log_level: LogLevel,
     status_code: u16,
     execution: ExecutionHandle,
+    response_group: GroupHandle,
+    response_body_group: GroupHandle,
   ) -> Self {
     Self {
       inner,
@@ -107,6 +117,8 @@ impl StreamingObserverBody {
         log_level,
         status_code,
         execution,
+        response_group,
+        response_body_group,
       ))),
     }
   }
@@ -246,9 +258,25 @@ where
       };
 
       let (parts, body) = req.into_parts();
+
+      // Create groups for this request
+      let request_group = GroupBuilder::new("request")
+        .metadata("method", parts.method.to_string())
+        .metadata("uri", parts.uri.to_string())
+        .build_with_execution(&execution)
+        .into_handle();
+      let request_headers_group = request_group
+        .child("headers")
+        .build_with_execution(&execution)
+        .into_handle();
+      let request_body_group = request_group
+        .child("body")
+        .build_with_execution(&execution)
+        .into_handle();
+
       ObservationBuilder::new("http/request/headers")
-        .label("http/request")
-        .label("http/request/headers")
+        .group(&request_group)
+        .group(&request_headers_group)
         .metadata("method", parts.method.to_string())
         .metadata("uri", parts.uri.to_string())
         .serde(&json!(filter_headers(
@@ -262,8 +290,8 @@ where
         .map(|collected| collected.to_bytes())
         .unwrap_or_else(|_| Bytes::new());
       ObservationBuilder::new("http/request/body")
-        .label("http/request")
-        .label("http/request/body")
+        .group(&request_group)
+        .group(&request_body_group)
         .metadata("method", parts.method.to_string())
         .metadata("uri", parts.uri.to_string())
         .payload(bytes_to_payload(&request_body_bytes, &parts.headers));
@@ -279,9 +307,22 @@ where
         500..=599 => LogLevel::Error,
         _ => LogLevel::Info,
       };
+
+      let response_group = GroupBuilder::new("response")
+        .build_with_execution(&execution)
+        .into_handle();
+      let response_headers_group = response_group
+        .child("headers")
+        .build_with_execution(&execution)
+        .into_handle();
+      let response_body_group = response_group
+        .child("body")
+        .build_with_execution(&execution)
+        .into_handle();
+
       ObservationBuilder::new("http/response/headers")
-        .label("http/response")
-        .label("http/response/headers")
+        .group(&response_group)
+        .group(&response_headers_group)
         .metadata("status", &parts.status.as_u16().to_string())
         .log_level(log_level)
         .serde(&json!(filter_headers(
@@ -303,6 +344,8 @@ where
         log_level,
         parts.status.as_u16(),
         execution,
+        response_group,
+        response_body_group,
       );
 
       Ok(Response::from_parts(parts, Body::new(streaming_body)))

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

@@ -36,6 +36,12 @@ pub(crate) enum UploaderMessage {
     handle: ObservationHandle,
     uploaded_tx: tokio::sync::watch::Sender<ObservationUploadResult>,
   },
+  AdditionalPayload {
+    observation_id: observation_tools_shared::ObservationId,
+    execution_id: observation_tools_shared::models::ExecutionId,
+    name: String,
+    payload: observation_tools_shared::Payload,
+  },
   Flush,
   Shutdown,
 }
@@ -57,6 +63,15 @@ impl std::fmt::Debug for UploaderMessage {
         .field("observations", observations)
         .field("handle", handle)
         .finish(),
+      Self::AdditionalPayload {
+        observation_id,
+        name,
+        ..
+      } => f
+        .debug_struct("AdditionalPayload")
+        .field("observation_id", observation_id)
+        .field("name", name)
+        .finish(),
       Self::Flush => write!(f, "Flush"),
       Self::Shutdown => write!(f, "Shutdown"),
     }
@@ -279,6 +294,15 @@ impl ClientBuilder {
   }
 }
 
+/// Data for an additional named payload to be sent alongside observations
+#[derive(Debug)]
+pub(crate) struct AdditionalPayloadData {
+  pub(crate) observation_id: observation_tools_shared::ObservationId,
+  pub(crate) execution_id: observation_tools_shared::models::ExecutionId,
+  pub(crate) name: String,
+  pub(crate) payload: observation_tools_shared::Payload,
+}
+
 async fn uploader_task(
   api_client: crate::server_client::Client,
   rx: async_channel::Receiver<UploaderMessage>,
@@ -291,12 +315,14 @@ async fn uploader_task(
     tokio::sync::watch::Sender<ObservationUploadResult>,
   );
 
-  let flush_observations = async |buffer: &mut Vec<ObservationWithPayload>,
-                                  senders: &mut Vec<ObservationSender>| {
-    if buffer.is_empty() {
+  let flush = async |buffer: &mut Vec<ObservationWithPayload>,
+                     senders: &mut Vec<ObservationSender>,
+                     additional_payloads: &mut Vec<AdditionalPayloadData>| {
+    if buffer.is_empty() && additional_payloads.is_empty() {
       return;
     }
-    let result = upload_observations(&api_client, buffer.drain(..).collect()).await;
+    let result =
+      upload_observations(&api_client, buffer.drain(..).collect(), additional_payloads.drain(..).collect()).await;
     match result {
       Ok(()) => {
         // Signal all senders that observations were uploaded successfully
@@ -316,6 +342,7 @@ async fn uploader_task(
   };
   let mut observation_buffer: Vec<ObservationWithPayload> = Vec::new();
   let mut sender_buffer: Vec<ObservationSender> = Vec::new();
+  let mut additional_payload_buffer: Vec<AdditionalPayloadData> = Vec::new();
   loop {
     let msg = rx.recv().await.ok();
     match msg {
@@ -345,14 +372,27 @@ async fn uploader_task(
         observation_buffer.extend(observations);
         sender_buffer.push((handle, uploaded_tx));
         if observation_buffer.len() >= BATCH_SIZE {
-          flush_observations(&mut observation_buffer, &mut sender_buffer).await;
+          flush(&mut observation_buffer, &mut sender_buffer, &mut additional_payload_buffer).await;
         }
       }
+      Some(UploaderMessage::AdditionalPayload {
+        observation_id,
+        execution_id,
+        name,
+        payload,
+      }) => {
+        additional_payload_buffer.push(AdditionalPayloadData {
+          observation_id,
+          execution_id,
+          name,
+          payload,
+        });
+      }
       Some(UploaderMessage::Flush) => {
-        flush_observations(&mut observation_buffer, &mut sender_buffer).await;
+        flush(&mut observation_buffer, &mut sender_buffer, &mut additional_payload_buffer).await;
       }
       Some(UploaderMessage::Shutdown) | None => {
-        flush_observations(&mut observation_buffer, &mut sender_buffer).await;
+        flush(&mut observation_buffer, &mut sender_buffer, &mut additional_payload_buffer).await;
         break;
       }
     }
@@ -384,30 +424,41 @@ async fn upload_execution(
 async fn upload_observations(
   client: &crate::server_client::Client,
   observations: Vec<ObservationWithPayload>,
+  additional_payloads: Vec<AdditionalPayloadData>,
 ) -> Result<()> {
-  if observations.is_empty() {
+  if observations.is_empty() && additional_payloads.is_empty() {
     return Ok(());
   }
 
   // Group by execution_id
-  let mut by_execution: std::collections::HashMap<_, Vec<_>> = std::collections::HashMap::new();
+  let mut by_execution: std::collections::HashMap<_, (Vec<_>, Vec<_>)> =
+    std::collections::HashMap::new();
   for obs in observations {
     by_execution
       .entry(obs.observation.execution_id)
       .or_default()
+      .0
       .push(obs);
   }
+  for ap in additional_payloads {
+    by_execution
+      .entry(ap.execution_id)
+      .or_default()
+      .1
+      .push(ap);
+  }
 
   // Upload each batch via multipart form
-  for (execution_id, observations) in by_execution {
+  for (execution_id, (observations, additional_payloads)) in by_execution {
     trace!(
-      "Uploading {} observations for execution {}",
+      "Uploading {} observations + {} additional payloads for execution {}",
       observations.len(),
+      additional_payloads.len(),
       execution_id
     );
 
     client
-      .create_observations_multipart(&execution_id.to_string(), observations)
+      .create_observations_multipart(&execution_id.to_string(), observations, additional_payloads)
       .await
       .map_err(|e| crate::error::Error::Config(e.to_string()))?;
   }

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

@@ -84,6 +84,15 @@ impl ExecutionHandle {
   pub fn base_url(&self) -> &str {
     &self.base_url
   }
+
+  /// Create a placeholder handle (for stub observations when no execution context exists)
+  pub(crate) fn placeholder() -> Self {
+    Self {
+      execution_id: ExecutionId::nil(),
+      uploader_tx: async_channel::unbounded().0,
+      base_url: String::new(),
+    }
+  }
 }
 
 #[napi]