refactor: route durable context through lifecycle processor instead of directly to logs agent

Removes the direct trace_agent → logs_agent dependency by having the
lifecycle processor (Processor) own durable_context_tx and forward
durable execution context to the logs pipeline from add_tracer_span().
The trace agent now sends aws.lambda spans to the lifecycle processor
(alongside INVOCATION_SPAN_RESOURCE spans) and the lifecycle processor
handles the durable context extraction and forwarding.

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

Author: lym953

bottlecap/src/bin/bottlecap/main.rs

@@ -330,6 +330,7 @@ async fn extension_loop_active(
             Arc::clone(&aws_config),
             metrics_aggregator_handle.clone(),
             Arc::clone(&propagator),
+            durable_context_tx,
         );
     tokio::spawn(async move {
         invocation_processor_service.run().await;
@@ -362,7 +363,6 @@ async fn extension_loop_active(
         &tags_provider,
         invocation_processor_handle.clone(),
         appsec_processor.clone(),
-        durable_context_tx,
         &shared_client,
     );
 
@@ -1091,7 +1091,6 @@ fn start_trace_agent(
     tags_provider: &Arc<TagProvider>,
     invocation_processor_handle: InvocationProcessorHandle,
     appsec_processor: Option<Arc<TokioMutex<AppSecProcessor>>>,
-    durable_context_tx: Sender<DurableContextUpdate>,
     client: &Client,
 ) -> (
     Sender<SendDataBuilderInfo>,
@@ -1177,7 +1176,6 @@ fn start_trace_agent(
         Arc::clone(tags_provider),
         stats_concentrator_handle.clone(),
         span_dedup_handle,
-        durable_context_tx,
     );
     let trace_agent_channel = trace_agent.get_sender_copy();
     let shutdown_token = trace_agent.shutdown_token();

bottlecap/src/lifecycle/invocation/processor.rs

@@ -11,6 +11,8 @@ use serde_json::Value;
 use tokio::time::Instant;
 use tracing::{debug, trace, warn};
 
+use tokio::sync::mpsc;
+
 use crate::{
     config::{self, aws::AwsConfig},
     extension::telemetry::events::{
@@ -24,6 +26,7 @@ use crate::{
         span_inferrer::{self, SpanInferrer},
         triggers::get_default_service_name,
     },
+    logs::agent::DurableContextUpdate,
     metrics::enhanced::lambda::{EnhancedMetricData, Lambda as EnhancedMetrics},
     proc::{
         self, CPUData, NetworkData,
@@ -88,6 +91,10 @@ pub struct Processor {
     /// Tracks whether if first invocation after init has been received in Managed Instance mode.
     /// Used to determine if we should search for the empty context on an invocation.
     awaiting_first_invocation: bool,
+    /// Sender used to forward durable execution context extracted from `aws.lambda` spans to the
+    /// logs pipeline. Decouples the trace agent from the logs agent: the trace agent sends spans
+    /// to the lifecycle processor, which extracts durable context and relays it here.
+    durable_context_tx: mpsc::Sender<DurableContextUpdate>,
 }
 
 impl Processor {
@@ -98,6 +105,7 @@ impl Processor {
         aws_config: Arc<AwsConfig>,
         metrics_aggregator: dogstatsd::aggregator::AggregatorHandle,
         propagator: Arc<DatadogCompositePropagator>,
+        durable_context_tx: mpsc::Sender<DurableContextUpdate>,
     ) -> Self {
         let resource = tags_provider
             .get_canonical_resource_name()
@@ -127,6 +135,7 @@ impl Processor {
             dynamic_tags: HashMap::new(),
             active_invocations: 0,
             awaiting_first_invocation: false,
+            durable_context_tx,
         }
     }
 
@@ -1337,10 +1346,27 @@ impl Processor {
     ///
     /// This is used to enrich the invocation span with additional metadata from the tracers
     /// top level span, since we discard the tracer span when we create the invocation span.
+    ///
+    /// Also forwards durable execution context to the logs pipeline when the span carries
+    /// `durable_function_execution_id` and `durable_function_execution_name` metadata.
     pub fn add_tracer_span(&mut self, span: &Span) {
         if let Some(request_id) = span.meta.get("request_id") {
             self.context_buffer.add_tracer_span(request_id, span);
         }
+
+        if let (Some(request_id), Some(exec_id), Some(exec_name)) = (
+            span.meta.get("request_id"),
+            span.meta.get("durable_function_execution_id"),
+            span.meta.get("durable_function_execution_name"),
+        ) {
+            if let Err(e) = self.durable_context_tx.try_send((
+                request_id.clone(),
+                exec_id.clone(),
+                exec_name.clone(),
+            )) {
+                warn!("LIFECYCLE | Failed to forward durable context to logs pipeline: {e}");
+            }
+        }
     }
 }
 
@@ -1387,7 +1413,8 @@ mod tests {
         tokio::spawn(service.run());
 
         let propagator = Arc::new(DatadogCompositePropagator::new(Arc::clone(&config)));
-        Processor::new(tags_provider, config, aws_config, handle, propagator)
+        let (durable_context_tx, _) = tokio::sync::mpsc::channel(1);
+        Processor::new(tags_provider, config, aws_config, handle, propagator, durable_context_tx)
     }
 
     #[test]
@@ -1924,7 +1951,8 @@ mod tests {
 
         let propagator = Arc::new(DatadogCompositePropagator::new(Arc::clone(&config)));
 
-        let processor = Processor::new(tags_provider, config, aws_config, handle, propagator);
+        let (durable_context_tx, _) = tokio::sync::mpsc::channel(1);
+        let processor = Processor::new(tags_provider, config, aws_config, handle, propagator, durable_context_tx);
 
         assert!(
             processor.is_managed_instance_mode(),

bottlecap/src/lifecycle/invocation/processor_service.rs

@@ -17,6 +17,7 @@ use crate::{
         context::{Context, ReparentingInfo},
         processor::Processor,
     },
+    logs::agent::DurableContextUpdate,
     tags::provider,
     traces::{
         context::SpanContext, propagation::DatadogCompositePropagator,
@@ -430,6 +431,7 @@ impl InvocationProcessorService {
         aws_config: Arc<AwsConfig>,
         metrics_aggregator_handle: AggregatorHandle,
         propagator: Arc<DatadogCompositePropagator>,
+        durable_context_tx: mpsc::Sender<DurableContextUpdate>,
     ) -> (InvocationProcessorHandle, Self) {
         let (sender, receiver) = mpsc::channel(1000);
 
@@ -439,6 +441,7 @@ impl InvocationProcessorService {
             aws_config,
             metrics_aggregator_handle,
             propagator,
+            durable_context_tx,
         );
 
         let handle = InvocationProcessorHandle { sender };

bottlecap/src/proxy/interceptor.rs

@@ -499,13 +499,16 @@ mod tests {
             initialization_type: "on-demand".into(),
         });
         let propagator = Arc::new(DatadogCompositePropagator::new(Arc::clone(&config)));
+        let (durable_context_tx, _durable_context_rx) =
+            tokio::sync::mpsc::channel(1);
         let (invocation_processor_handle, invocation_processor_service) =
             InvocationProcessorService::new(
                 Arc::clone(&tags_provider),
                 Arc::clone(&config),
                 Arc::clone(&aws_config),
                 metrics_aggregator,
                 Arc::clone(&propagator),
+                durable_context_tx,
             );
         tokio::spawn(async move {
             invocation_processor_service.run().await;

bottlecap/src/traces/trace_agent.rs

@@ -21,7 +21,6 @@ use tokio_util::sync::CancellationToken;
 use tower_http::limit::RequestBodyLimitLayer;
 use tracing::{debug, error, warn};
 
-use crate::logs::agent::DurableContextUpdate;
 use crate::traces::trace_processor::SendingTraceProcessor;
 use crate::{
     appsec::processor::Processor as AppSecProcessor,
@@ -92,7 +91,6 @@ pub struct TraceState {
     pub invocation_processor_handle: InvocationProcessorHandle,
     pub tags_provider: Arc<provider::Provider>,
     pub span_deduper: DedupHandle,
-    pub durable_context_tx: Sender<DurableContextUpdate>,
 }
 
 #[derive(Clone)]
@@ -120,7 +118,6 @@ pub struct TraceAgent {
     tx: Sender<SendDataBuilderInfo>,
     stats_concentrator: StatsConcentratorHandle,
     span_deduper: DedupHandle,
-    durable_context_tx: Sender<DurableContextUpdate>,
 }
 
 #[derive(Clone, Copy)]
@@ -144,7 +141,6 @@ impl TraceAgent {
         tags_provider: Arc<provider::Provider>,
         stats_concentrator: StatsConcentratorHandle,
         span_deduper: DedupHandle,
-        durable_context_tx: Sender<DurableContextUpdate>,
     ) -> TraceAgent {
         // Set up a channel to send processed traces to our trace aggregator. tx is passed through each
         // endpoint_handler to the trace processor, which uses it to send de-serialized
@@ -174,7 +170,6 @@ impl TraceAgent {
             shutdown_token: CancellationToken::new(),
             stats_concentrator,
             span_deduper,
-            durable_context_tx,
         }
     }
 
@@ -230,7 +225,6 @@ impl TraceAgent {
             invocation_processor_handle: self.invocation_processor_handle.clone(),
             tags_provider: Arc::clone(&self.tags_provider),
             span_deduper: self.span_deduper.clone(),
-            durable_context_tx: self.durable_context_tx.clone(),
         };
 
         let stats_state = StatsState {
@@ -310,7 +304,6 @@ impl TraceAgent {
             state.invocation_processor_handle,
             state.tags_provider,
             state.span_deduper,
-            state.durable_context_tx,
             ApiVersion::V04,
         )
         .await
@@ -324,7 +317,6 @@ impl TraceAgent {
             state.invocation_processor_handle,
             state.tags_provider,
             state.span_deduper,
-            state.durable_context_tx,
             ApiVersion::V05,
         )
         .await
@@ -481,7 +473,6 @@ impl TraceAgent {
         invocation_processor_handle: InvocationProcessorHandle,
         tags_provider: Arc<provider::Provider>,
         deduper: DedupHandle,
-        durable_context_tx: Sender<DurableContextUpdate>,
         version: ApiVersion,
     ) -> Response {
         let start = Instant::now();
@@ -593,7 +584,7 @@ impl TraceAgent {
                     }
                 }
 
-                if span.resource == INVOCATION_SPAN_RESOURCE
+                if (span.resource == INVOCATION_SPAN_RESOURCE || span.name == "aws.lambda")
                     && let Err(e) = invocation_processor_handle
                         .add_tracer_span(span.clone())
                         .await
@@ -602,24 +593,6 @@ impl TraceAgent {
                 }
                 handle_reparenting(&mut reparenting_info, &mut span);
 
-                if span.name == "aws.lambda" {
-                    // If this aws.lambda span carries durable function context, forward it to
-                    // the logs agent so it can release any held logs and set durable execution context on them.
-                    if let (Some(request_id), Some(execution_id), Some(execution_name)) = (
-                        span.meta.get("request_id"),
-                        span.meta.get("durable_function_execution_id"),
-                        span.meta.get("durable_function_execution_name"),
-                    ) {
-                        let _ = durable_context_tx
-                            .send((
-                                request_id.clone(),
-                                execution_id.clone(),
-                                execution_name.clone(),
-                            ))
-                            .await;
-                    }
-                }
-
                 // Keep the span
                 chunk.push(span);
             }