diff --git a/lib/datadog/appsec/contrib/graphql/gateway/watcher.rb b/lib/datadog/appsec/contrib/graphql/gateway/watcher.rb
index 68a1c5acf11..4520e05b111 100644
--- a/lib/datadog/appsec/contrib/graphql/gateway/watcher.rb
+++ b/lib/datadog/appsec/contrib/graphql/gateway/watcher.rb
@@ -2,8 +2,8 @@
 
 require 'json'
 require_relative '../../../instrumentation/gateway'
+require_relative '../../../reactive/engine'
 require_relative '../reactive/multiplex'
-require_relative '../../../reactive/operation'
 
 module Datadog
   module AppSec
@@ -24,26 +24,24 @@ def watch_multiplex(gateway = Instrumentation.gateway)
                 gateway.watch('graphql.multiplex', :appsec) do |stack, gateway_multiplex|
                   block = false
                   event = nil
-
                   scope = AppSec::Scope.active_scope
+                  engine = AppSec::Reactive::Engine.new
 
                   if scope
-                    AppSec::Reactive::Operation.new('graphql.multiplex') do |op|
-                      GraphQL::Reactive::Multiplex.subscribe(op, scope.processor_context) do |result|
-                        event = {
-                          waf_result: result,
-                          trace: scope.trace,
-                          span: scope.service_entry_span,
-                          multiplex: gateway_multiplex,
-                          actions: result.actions
-                        }
-
-                        Datadog::AppSec::Event.tag_and_keep!(scope, result)
-                        scope.processor_context.events << event
-                      end
+                    GraphQL::Reactive::Multiplex.subscribe(engine, scope.processor_context) do |result|
+                      event = {
+                        waf_result: result,
+                        trace: scope.trace,
+                        span: scope.service_entry_span,
+                        multiplex: gateway_multiplex,
+                        actions: result.actions
+                      }
 
-                      block = GraphQL::Reactive::Multiplex.publish(op, gateway_multiplex)
+                      Datadog::AppSec::Event.tag_and_keep!(scope, result)
+                      scope.processor_context.events << event
                     end
+
+                    block = GraphQL::Reactive::Multiplex.publish(engine, gateway_multiplex)
                   end
 
                   next [nil, [[:block, event]]] if block
diff --git a/lib/datadog/appsec/contrib/graphql/reactive/multiplex.rb b/lib/datadog/appsec/contrib/graphql/reactive/multiplex.rb
index 4d08c60b861..cfbf31dc1b8 100644
--- a/lib/datadog/appsec/contrib/graphql/reactive/multiplex.rb
+++ b/lib/datadog/appsec/contrib/graphql/reactive/multiplex.rb
@@ -12,16 +12,16 @@ module Multiplex
             ].freeze
             private_constant :ADDRESSES
 
-            def self.publish(op, gateway_multiplex)
+            def self.publish(engine, gateway_multiplex)
               catch(:block) do
-                op.publish('graphql.server.all_resolvers', gateway_multiplex.arguments)
+                engine.publish('graphql.server.all_resolvers', gateway_multiplex.arguments)
 
                 nil
               end
             end
 
-            def self.subscribe(op, waf_context)
-              op.subscribe(*ADDRESSES) do |*values|
+            def self.subscribe(engine, waf_context)
+              engine.subscribe(*ADDRESSES) do |*values|
                 Datadog.logger.debug { "reacted to #{ADDRESSES.inspect}: #{values.inspect}" }
                 arguments = values[0]
 
diff --git a/lib/datadog/appsec/contrib/rack/gateway/watcher.rb b/lib/datadog/appsec/contrib/rack/gateway/watcher.rb
index a66387329c3..fb7e095f834 100644
--- a/lib/datadog/appsec/contrib/rack/gateway/watcher.rb
+++ b/lib/datadog/appsec/contrib/rack/gateway/watcher.rb
@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 require_relative '../../../instrumentation/gateway'
-require_relative '../../../reactive/operation'
+require_relative '../../../reactive/engine'
 require_relative '../reactive/request'
 require_relative '../reactive/request_body'
 require_relative '../reactive/response'
@@ -25,32 +25,29 @@ def watch
 
               def watch_request(gateway = Instrumentation.gateway)
                 gateway.watch('rack.request', :appsec) do |stack, gateway_request|
-                  block = false
                   event = nil
                   scope = gateway_request.env[Datadog::AppSec::Ext::SCOPE_KEY]
-
-                  AppSec::Reactive::Operation.new('rack.request') do |op|
-                    Rack::Reactive::Request.subscribe(op, scope.processor_context) do |result|
-                      if result.status == :match
-                        # TODO: should this hash be an Event instance instead?
-                        event = {
-                          waf_result: result,
-                          trace: scope.trace,
-                          span: scope.service_entry_span,
-                          request: gateway_request,
-                          actions: result.actions
-                        }
-
-                        # We want to keep the trace in case of security event
-                        scope.trace.keep! if scope.trace
-                        Datadog::AppSec::Event.tag_and_keep!(scope, result)
-                        scope.processor_context.events << event
-                      end
+                  engine = AppSec::Reactive::Engine.new
+
+                  Rack::Reactive::Request.subscribe(engine, scope.processor_context) do |result|
+                    if result.status == :match
+                      # TODO: should this hash be an Event instance instead?
+                      event = {
+                        waf_result: result,
+                        trace: scope.trace,
+                        span: scope.service_entry_span,
+                        request: gateway_request,
+                        actions: result.actions
+                      }
+
+                      # We want to keep the trace in case of security event
+                      scope.trace.keep! if scope.trace
+                      Datadog::AppSec::Event.tag_and_keep!(scope, result)
+                      scope.processor_context.events << event
                     end
-
-                    block = Rack::Reactive::Request.publish(op, gateway_request)
                   end
 
+                  block = Rack::Reactive::Request.publish(engine, gateway_request)
                   next [nil, [[:block, event]]] if block
 
                   ret, res = stack.call(gateway_request.request)
@@ -66,33 +63,29 @@ def watch_request(gateway = Instrumentation.gateway)
 
               def watch_response(gateway = Instrumentation.gateway)
                 gateway.watch('rack.response', :appsec) do |stack, gateway_response|
-                  block = false
-
                   event = nil
                   scope = gateway_response.scope
-
-                  AppSec::Reactive::Operation.new('rack.response') do |op|
-                    Rack::Reactive::Response.subscribe(op, scope.processor_context) do |result|
-                      if result.status == :match
-                        # TODO: should this hash be an Event instance instead?
-                        event = {
-                          waf_result: result,
-                          trace: scope.trace,
-                          span: scope.service_entry_span,
-                          response: gateway_response,
-                          actions: result.actions
-                        }
-
-                        # We want to keep the trace in case of security event
-                        scope.trace.keep! if scope.trace
-                        Datadog::AppSec::Event.tag_and_keep!(scope, result)
-                        scope.processor_context.events << event
-                      end
+                  engine = AppSec::Reactive::Engine.new
+
+                  Rack::Reactive::Response.subscribe(engine, scope.processor_context) do |result|
+                    if result.status == :match
+                      # TODO: should this hash be an Event instance instead?
+                      event = {
+                        waf_result: result,
+                        trace: scope.trace,
+                        span: scope.service_entry_span,
+                        response: gateway_response,
+                        actions: result.actions
+                      }
+
+                      # We want to keep the trace in case of security event
+                      scope.trace.keep! if scope.trace
+                      Datadog::AppSec::Event.tag_and_keep!(scope, result)
+                      scope.processor_context.events << event
                     end
-
-                    block = Rack::Reactive::Response.publish(op, gateway_response)
                   end
 
+                  block = Rack::Reactive::Response.publish(engine, gateway_response)
                   next [nil, [[:block, event]]] if block
 
                   ret, res = stack.call(gateway_response.response)
@@ -108,33 +101,29 @@ def watch_response(gateway = Instrumentation.gateway)
 
               def watch_request_body(gateway = Instrumentation.gateway)
                 gateway.watch('rack.request.body', :appsec) do |stack, gateway_request|
-                  block = false
-
                   event = nil
                   scope = gateway_request.env[Datadog::AppSec::Ext::SCOPE_KEY]
-
-                  AppSec::Reactive::Operation.new('rack.request.body') do |op|
-                    Rack::Reactive::RequestBody.subscribe(op, scope.processor_context) do |result|
-                      if result.status == :match
-                        # TODO: should this hash be an Event instance instead?
-                        event = {
-                          waf_result: result,
-                          trace: scope.trace,
-                          span: scope.service_entry_span,
-                          request: gateway_request,
-                          actions: result.actions
-                        }
-
-                        # We want to keep the trace in case of security event
-                        scope.trace.keep! if scope.trace
-                        Datadog::AppSec::Event.tag_and_keep!(scope, result)
-                        scope.processor_context.events << event
-                      end
+                  engine = AppSec::Reactive::Engine.new
+
+                  Rack::Reactive::RequestBody.subscribe(engine, scope.processor_context) do |result|
+                    if result.status == :match
+                      # TODO: should this hash be an Event instance instead?
+                      event = {
+                        waf_result: result,
+                        trace: scope.trace,
+                        span: scope.service_entry_span,
+                        request: gateway_request,
+                        actions: result.actions
+                      }
+
+                      # We want to keep the trace in case of security event
+                      scope.trace.keep! if scope.trace
+                      Datadog::AppSec::Event.tag_and_keep!(scope, result)
+                      scope.processor_context.events << event
                     end
-
-                    block = Rack::Reactive::RequestBody.publish(op, gateway_request)
                   end
 
+                  block = Rack::Reactive::RequestBody.publish(engine, gateway_request)
                   next [nil, [[:block, event]]] if block
 
                   ret, res = stack.call(gateway_request.request)
diff --git a/lib/datadog/appsec/contrib/rack/reactive/request.rb b/lib/datadog/appsec/contrib/rack/reactive/request.rb
index 4f88a8aa989..78074268af6 100644
--- a/lib/datadog/appsec/contrib/rack/reactive/request.rb
+++ b/lib/datadog/appsec/contrib/rack/reactive/request.rb
@@ -17,21 +17,21 @@ module Request
             ].freeze
             private_constant :ADDRESSES
 
-            def self.publish(op, gateway_request)
+            def self.publish(engine, gateway_request)
               catch(:block) do
-                op.publish('request.query', gateway_request.query)
-                op.publish('request.headers', gateway_request.headers)
-                op.publish('request.uri.raw', gateway_request.fullpath)
-                op.publish('request.cookies', gateway_request.cookies)
-                op.publish('request.client_ip', gateway_request.client_ip)
-                op.publish('server.request.method', gateway_request.method)
+                engine.publish('request.query', gateway_request.query)
+                engine.publish('request.headers', gateway_request.headers)
+                engine.publish('request.uri.raw', gateway_request.fullpath)
+                engine.publish('request.cookies', gateway_request.cookies)
+                engine.publish('request.client_ip', gateway_request.client_ip)
+                engine.publish('server.request.method', gateway_request.method)
 
                 nil
               end
             end
 
-            def self.subscribe(op, waf_context)
-              op.subscribe(*ADDRESSES) do |*values|
+            def self.subscribe(engine, waf_context)
+              engine.subscribe(*ADDRESSES) do |*values|
                 Datadog.logger.debug { "reacted to #{ADDRESSES.inspect}: #{values.inspect}" }
 
                 headers = values[0]
diff --git a/lib/datadog/appsec/contrib/rack/reactive/request_body.rb b/lib/datadog/appsec/contrib/rack/reactive/request_body.rb
index 1cf98518077..57395ac83b8 100644
--- a/lib/datadog/appsec/contrib/rack/reactive/request_body.rb
+++ b/lib/datadog/appsec/contrib/rack/reactive/request_body.rb
@@ -12,17 +12,17 @@ module RequestBody
             ].freeze
             private_constant :ADDRESSES
 
-            def self.publish(op, gateway_request)
+            def self.publish(engine, gateway_request)
               catch(:block) do
                 # params have been parsed from the request body
-                op.publish('request.body', gateway_request.form_hash)
+                engine.publish('request.body', gateway_request.form_hash)
 
                 nil
               end
             end
 
-            def self.subscribe(op, waf_context)
-              op.subscribe(*ADDRESSES) do |*values|
+            def self.subscribe(engine, waf_context)
+              engine.subscribe(*ADDRESSES) do |*values|
                 Datadog.logger.debug { "reacted to #{ADDRESSES.inspect}: #{values.inspect}" }
                 body = values[0]
 
diff --git a/lib/datadog/appsec/contrib/rack/reactive/response.rb b/lib/datadog/appsec/contrib/rack/reactive/response.rb
index 850e3c022f8..8c0cd706d26 100644
--- a/lib/datadog/appsec/contrib/rack/reactive/response.rb
+++ b/lib/datadog/appsec/contrib/rack/reactive/response.rb
@@ -13,17 +13,17 @@ module Response
             ].freeze
             private_constant :ADDRESSES
 
-            def self.publish(op, gateway_response)
+            def self.publish(engine, gateway_response)
               catch(:block) do
-                op.publish('response.status', gateway_response.status)
-                op.publish('response.headers', gateway_response.headers)
+                engine.publish('response.status', gateway_response.status)
+                engine.publish('response.headers', gateway_response.headers)
 
                 nil
               end
             end
 
-            def self.subscribe(op, waf_context)
-              op.subscribe(*ADDRESSES) do |*values|
+            def self.subscribe(engine, waf_context)
+              engine.subscribe(*ADDRESSES) do |*values|
                 Datadog.logger.debug { "reacted to #{ADDRESSES.inspect}: #{values.inspect}" }
 
                 response_status = values[0]
diff --git a/lib/datadog/appsec/contrib/rails/gateway/watcher.rb b/lib/datadog/appsec/contrib/rails/gateway/watcher.rb
index ab1bfe612ec..6fb24027f61 100644
--- a/lib/datadog/appsec/contrib/rails/gateway/watcher.rb
+++ b/lib/datadog/appsec/contrib/rails/gateway/watcher.rb
@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 require_relative '../../../instrumentation/gateway'
-require_relative '../../../reactive/operation'
+require_relative '../../../reactive/engine'
 require_relative '../reactive/action'
 require_relative '../../../event'
 
@@ -21,33 +21,29 @@ def watch
 
               def watch_request_action(gateway = Instrumentation.gateway)
                 gateway.watch('rails.request.action', :appsec) do |stack, gateway_request|
-                  block = false
-
                   event = nil
                   scope = gateway_request.env[Datadog::AppSec::Ext::SCOPE_KEY]
+                  engine = AppSec::Reactive::Engine.new
 
-                  AppSec::Reactive::Operation.new('rails.request.action') do |op|
-                    Rails::Reactive::Action.subscribe(op, scope.processor_context) do |result|
-                      if result.status == :match
-                        # TODO: should this hash be an Event instance instead?
-                        event = {
-                          waf_result: result,
-                          trace: scope.trace,
-                          span: scope.service_entry_span,
-                          request: gateway_request,
-                          actions: result.actions
-                        }
+                  Rails::Reactive::Action.subscribe(engine, scope.processor_context) do |result|
+                    if result.status == :match
+                      # TODO: should this hash be an Event instance instead?
+                      event = {
+                        waf_result: result,
+                        trace: scope.trace,
+                        span: scope.service_entry_span,
+                        request: gateway_request,
+                        actions: result.actions
+                      }
 
-                        # We want to keep the trace in case of security event
-                        scope.trace.keep! if scope.trace
-                        Datadog::AppSec::Event.tag_and_keep!(scope, result)
-                        scope.processor_context.events << event
-                      end
+                      # We want to keep the trace in case of security event
+                      scope.trace.keep! if scope.trace
+                      Datadog::AppSec::Event.tag_and_keep!(scope, result)
+                      scope.processor_context.events << event
                     end
-
-                    block = Rails::Reactive::Action.publish(op, gateway_request)
                   end
 
+                  block = Rails::Reactive::Action.publish(engine, gateway_request)
                   next [nil, [[:block, event]]] if block
 
                   ret, res = stack.call(gateway_request.request)
diff --git a/lib/datadog/appsec/contrib/rails/reactive/action.rb b/lib/datadog/appsec/contrib/rails/reactive/action.rb
index eb28c9983c0..9dbe8697e51 100644
--- a/lib/datadog/appsec/contrib/rails/reactive/action.rb
+++ b/lib/datadog/appsec/contrib/rails/reactive/action.rb
@@ -15,18 +15,18 @@ module Action
             ].freeze
             private_constant :ADDRESSES
 
-            def self.publish(op, gateway_request)
+            def self.publish(engine, gateway_request)
               catch(:block) do
                 # params have been parsed from the request body
-                op.publish('rails.request.body', gateway_request.parsed_body)
-                op.publish('rails.request.route_params', gateway_request.route_params)
+                engine.publish('rails.request.body', gateway_request.parsed_body)
+                engine.publish('rails.request.route_params', gateway_request.route_params)
 
                 nil
               end
             end
 
-            def self.subscribe(op, waf_context)
-              op.subscribe(*ADDRESSES) do |*values|
+            def self.subscribe(engine, waf_context)
+              engine.subscribe(*ADDRESSES) do |*values|
                 Datadog.logger.debug { "reacted to #{ADDRESSES.inspect}: #{values.inspect}" }
                 body = values[0]
                 path_params = values[1]
diff --git a/lib/datadog/appsec/contrib/sinatra/gateway/watcher.rb b/lib/datadog/appsec/contrib/sinatra/gateway/watcher.rb
index 91383478c29..442d829cd34 100644
--- a/lib/datadog/appsec/contrib/sinatra/gateway/watcher.rb
+++ b/lib/datadog/appsec/contrib/sinatra/gateway/watcher.rb
@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 require_relative '../../../instrumentation/gateway'
-require_relative '../../../reactive/operation'
+require_relative '../../../reactive/engine'
 require_relative '../../rack/reactive/request_body'
 require_relative '../reactive/routed'
 require_relative '../../../event'
@@ -23,33 +23,29 @@ def watch
 
               def watch_request_dispatch(gateway = Instrumentation.gateway)
                 gateway.watch('sinatra.request.dispatch', :appsec) do |stack, gateway_request|
-                  block = false
-
                   event = nil
                   scope = gateway_request.env[Datadog::AppSec::Ext::SCOPE_KEY]
-
-                  AppSec::Reactive::Operation.new('sinatra.request.dispatch') do |op|
-                    Rack::Reactive::RequestBody.subscribe(op, scope.processor_context) do |result|
-                      if result.status == :match
-                        # TODO: should this hash be an Event instance instead?
-                        event = {
-                          waf_result: result,
-                          trace: scope.trace,
-                          span: scope.service_entry_span,
-                          request: gateway_request,
-                          actions: result.actions
-                        }
-
-                        # We want to keep the trace in case of security event
-                        scope.trace.keep! if scope.trace
-                        Datadog::AppSec::Event.tag_and_keep!(scope, result)
-                        scope.processor_context.events << event
-                      end
+                  engine = AppSec::Reactive::Engine.new
+
+                  Rack::Reactive::RequestBody.subscribe(engine, scope.processor_context) do |result|
+                    if result.status == :match
+                      # TODO: should this hash be an Event instance instead?
+                      event = {
+                        waf_result: result,
+                        trace: scope.trace,
+                        span: scope.service_entry_span,
+                        request: gateway_request,
+                        actions: result.actions
+                      }
+
+                      # We want to keep the trace in case of security event
+                      scope.trace.keep! if scope.trace
+                      Datadog::AppSec::Event.tag_and_keep!(scope, result)
+                      scope.processor_context.events << event
                     end
-
-                    block = Rack::Reactive::RequestBody.publish(op, gateway_request)
                   end
 
+                  block = Rack::Reactive::RequestBody.publish(engine, gateway_request)
                   next [nil, [[:block, event]]] if block
 
                   ret, res = stack.call(gateway_request.request)
@@ -65,33 +61,29 @@ def watch_request_dispatch(gateway = Instrumentation.gateway)
 
               def watch_request_routed(gateway = Instrumentation.gateway)
                 gateway.watch('sinatra.request.routed', :appsec) do |stack, (gateway_request, gateway_route_params)|
-                  block = false
-
                   event = nil
                   scope = gateway_request.env[Datadog::AppSec::Ext::SCOPE_KEY]
-
-                  AppSec::Reactive::Operation.new('sinatra.request.routed') do |op|
-                    Sinatra::Reactive::Routed.subscribe(op, scope.processor_context) do |result|
-                      if result.status == :match
-                        # TODO: should this hash be an Event instance instead?
-                        event = {
-                          waf_result: result,
-                          trace: scope.trace,
-                          span: scope.service_entry_span,
-                          request: gateway_request,
-                          actions: result.actions
-                        }
-
-                        # We want to keep the trace in case of security event
-                        scope.trace.keep! if scope.trace
-                        Datadog::AppSec::Event.tag_and_keep!(scope, result)
-                        scope.processor_context.events << event
-                      end
+                  engine = AppSec::Reactive::Engine.new
+
+                  Sinatra::Reactive::Routed.subscribe(engine, scope.processor_context) do |result|
+                    if result.status == :match
+                      # TODO: should this hash be an Event instance instead?
+                      event = {
+                        waf_result: result,
+                        trace: scope.trace,
+                        span: scope.service_entry_span,
+                        request: gateway_request,
+                        actions: result.actions
+                      }
+
+                      # We want to keep the trace in case of security event
+                      scope.trace.keep! if scope.trace
+                      Datadog::AppSec::Event.tag_and_keep!(scope, result)
+                      scope.processor_context.events << event
                     end
-
-                    block = Sinatra::Reactive::Routed.publish(op, [gateway_request, gateway_route_params])
                   end
 
+                  block = Sinatra::Reactive::Routed.publish(engine, [gateway_request, gateway_route_params])
                   next [nil, [[:block, event]]] if block
 
                   ret, res = stack.call(gateway_request.request)
diff --git a/lib/datadog/appsec/contrib/sinatra/reactive/routed.rb b/lib/datadog/appsec/contrib/sinatra/reactive/routed.rb
index 60e0f7e6301..e34dfb4ccb3 100644
--- a/lib/datadog/appsec/contrib/sinatra/reactive/routed.rb
+++ b/lib/datadog/appsec/contrib/sinatra/reactive/routed.rb
@@ -12,18 +12,18 @@ module Routed
             ].freeze
             private_constant :ADDRESSES
 
-            def self.publish(op, data)
+            def self.publish(engine, data)
               _request, route_params = data
 
               catch(:block) do
-                op.publish('sinatra.request.route_params', route_params.params)
+                engine.publish('sinatra.request.route_params', route_params.params)
 
                 nil
               end
             end
 
-            def self.subscribe(op, waf_context)
-              op.subscribe(*ADDRESSES) do |*values|
+            def self.subscribe(engine, waf_context)
+              engine.subscribe(*ADDRESSES) do |*values|
                 Datadog.logger.debug { "reacted to #{ADDRESSES.inspect}: #{values.inspect}" }
                 path_params = values[0]
 
diff --git a/lib/datadog/appsec/monitor/gateway/watcher.rb b/lib/datadog/appsec/monitor/gateway/watcher.rb
index 74fc4d3fd60..e8beb102e52 100644
--- a/lib/datadog/appsec/monitor/gateway/watcher.rb
+++ b/lib/datadog/appsec/monitor/gateway/watcher.rb
@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 require_relative '../../instrumentation/gateway'
-require_relative '../../reactive/operation'
+require_relative '../../reactive/engine'
 require_relative '../reactive/set_user'
 
 module Datadog
@@ -19,32 +19,29 @@ def watch
 
             def watch_user_id(gateway = Instrumentation.gateway)
               gateway.watch('identity.set_user', :appsec) do |stack, user|
-                block = false
                 event = nil
                 scope = Datadog::AppSec.active_scope
-
-                AppSec::Reactive::Operation.new('identity.set_user') do |op|
-                  Monitor::Reactive::SetUser.subscribe(op, scope.processor_context) do |result|
-                    if result.status == :match
-                      # TODO: should this hash be an Event instance instead?
-                      event = {
-                        waf_result: result,
-                        trace: scope.trace,
-                        span: scope.service_entry_span,
-                        user: user,
-                        actions: result.actions
-                      }
-
-                      # We want to keep the trace in case of security event
-                      scope.trace.keep! if scope.trace
-                      Datadog::AppSec::Event.tag_and_keep!(scope, result)
-                      scope.processor_context.events << event
-                    end
+                engine = AppSec::Reactive::Engine.new
+
+                Monitor::Reactive::SetUser.subscribe(engine, scope.processor_context) do |result|
+                  if result.status == :match
+                    # TODO: should this hash be an Event instance instead?
+                    event = {
+                      waf_result: result,
+                      trace: scope.trace,
+                      span: scope.service_entry_span,
+                      user: user,
+                      actions: result.actions
+                    }
+
+                    # We want to keep the trace in case of security event
+                    scope.trace.keep! if scope.trace
+                    Datadog::AppSec::Event.tag_and_keep!(scope, result)
+                    scope.processor_context.events << event
                   end
-
-                  block = Monitor::Reactive::SetUser.publish(op, user)
                 end
 
+                block = Monitor::Reactive::SetUser.publish(engine, user)
                 throw(Datadog::AppSec::Ext::INTERRUPT, [nil, [[:block, event]]]) if block
 
                 ret, res = stack.call(user)
diff --git a/lib/datadog/appsec/monitor/reactive/set_user.rb b/lib/datadog/appsec/monitor/reactive/set_user.rb
index f3c43883bd5..ea3128c6031 100644
--- a/lib/datadog/appsec/monitor/reactive/set_user.rb
+++ b/lib/datadog/appsec/monitor/reactive/set_user.rb
@@ -11,16 +11,16 @@ module SetUser
           ].freeze
           private_constant :ADDRESSES
 
-          def self.publish(op, user)
+          def self.publish(engine, user)
             catch(:block) do
-              op.publish('usr.id', user.id)
+              engine.publish('usr.id', user.id)
 
               nil
             end
           end
 
-          def self.subscribe(op, waf_context)
-            op.subscribe(*ADDRESSES) do |*values|
+          def self.subscribe(engine, waf_context)
+            engine.subscribe(*ADDRESSES) do |*values|
               Datadog.logger.debug { "reacted to #{ADDRESSES.inspect}: #{values.inspect}" }
 
               user_id = values[0]
diff --git a/lib/datadog/appsec/reactive/operation.rb b/lib/datadog/appsec/reactive/operation.rb
deleted file mode 100644
index f7696cdbcec..00000000000
--- a/lib/datadog/appsec/reactive/operation.rb
+++ /dev/null
@@ -1,68 +0,0 @@
-# frozen_string_literal: true
-
-require_relative 'engine'
-
-module Datadog
-  module AppSec
-    module Reactive
-      # Reactive Engine nested operation tracking
-      class Operation
-        attr_reader :reactive,
-          :parent,
-          :name
-
-        def initialize(name, parent = nil, reactive_engine = nil)
-          Datadog.logger.debug { "operation: #{name} initialize" }
-          @name = name
-          @parent = parent
-          @reactive = select_reactive_engine(reactive_engine, parent)
-
-          # TODO: concurrent store
-          # TODO: constant
-          Thread.current[:datadog_security_active_operation] = self
-
-          yield self if block_given?
-        ensure
-          finalize
-        end
-
-        # TODO: use structs instead of an arg splat
-        def subscribe(*addresses, &block)
-          reactive.subscribe(*addresses, &block)
-        end
-
-        def publish(address, data)
-          reactive.publish(address, data)
-        end
-
-        def finalize
-          Datadog.logger.debug { "operation: #{name} finalize" }
-          Thread.current[:datadog_security_active_operation] = parent
-        end
-
-        private
-
-        def select_reactive_engine(reactive, parent)
-          return reactive if reactive
-
-          return parent.reactive unless parent.nil?
-
-          Reactive::Engine.new
-        end
-
-        class << self
-          def active
-            Thread.current[:datadog_security_active_operation]
-          end
-
-          private
-
-          # For testing only.
-          def reset!
-            Thread.current[:datadog_security_active_operation] = nil
-          end
-        end
-      end
-    end
-  end
-end
diff --git a/lib/datadog/di/error.rb b/lib/datadog/di/error.rb
index d5f305617cd..c1785f71ba7 100644
--- a/lib/datadog/di/error.rb
+++ b/lib/datadog/di/error.rb
@@ -27,6 +27,11 @@ class AgentCommunicationError < Error
       class DITargetNotDefined < Error
       end
 
+      # Attempting to instrument a line and the file containing the line
+      # was loaded prior to code tracking being enabled.
+      class DITargetNotInRegistry < Error
+      end
+
       # Raised when trying to install a probe whose installation failed
       # earlier in the same process. This exception should contain the
       # original exception report from initial installation attempt.
diff --git a/lib/datadog/di/instrumenter.rb b/lib/datadog/di/instrumenter.rb
index b9238c9ad74..b1f8f0f599a 100644
--- a/lib/datadog/di/instrumenter.rb
+++ b/lib/datadog/di/instrumenter.rb
@@ -246,11 +246,12 @@ def hook_line(probe, &block)
               #
               # If the requested file is not in code tracker's registry,
               # or the code tracker does not exist at all,
-              # do not attempt to instrumnet now.
+              # do not attempt to instrument now.
               # The caller should add the line to the list of pending lines
               # to instrument and install the hook when the file in
               # question is loaded (and hopefully, by then code tracking
               # is active, otherwise the line will never be instrumented.)
+              raise_if_probe_in_loaded_features(probe)
               raise Error::DITargetNotDefined, "File not in code tracker registry: #{probe.file}"
             end
           end
@@ -258,6 +259,7 @@ def hook_line(probe, &block)
           # Same as previous comment, if untargeted trace points are not
           # explicitly defined, and we do not have code tracking, do not
           # instrument the method.
+          raise_if_probe_in_loaded_features(probe)
           raise Error::DITargetNotDefined, "File not in code tracker registry: #{probe.file}"
         end
 
@@ -374,6 +376,26 @@ def unhook(probe)
 
       attr_reader :lock
 
+      def raise_if_probe_in_loaded_features(probe)
+        return unless probe.file
+
+        # If the probe file is in the list of loaded files
+        # (as per $LOADED_FEATURES, using either exact or suffix match),
+        # raise an error indicating that
+        # code tracker is missing the loaded file because the file
+        # won't be loaded again (DI only works in production environments
+        # that do not normally reload code).
+        if $LOADED_FEATURES.include?(probe.file)
+          raise Error::DITargetNotInRegistry, "File loaded but is not in code tracker registry: #{probe.file}"
+        end
+        # Ths is an expensive check
+        $LOADED_FEATURES.each do |path|
+          if Utils.path_matches_suffix?(path, probe.file)
+            raise Error::DITargetNotInRegistry, "File matching probe path (#{probe.file}) was loaded and is not in code tracker registry: #{path}"
+          end
+        end
+      end
+
       # TODO test that this resolves qualified names e.g. A::B
       def symbolize_class_name(cls_name)
         Object.const_get(cls_name)
diff --git a/lib/datadog/tracing/span_event.rb b/lib/datadog/tracing/span_event.rb
index 79f2ff18783..4a0685e7b15 100644
--- a/lib/datadog/tracing/span_event.rb
+++ b/lib/datadog/tracing/span_event.rb
@@ -19,23 +19,143 @@ class SpanEvent
       #   @return [Integer]
       attr_reader :time_unix_nano
 
+      # TODO: Accept {Time} as the time_unix_nano parameter, possibly in addition to the current nano integer.
       def initialize(
         name,
         attributes: nil,
         time_unix_nano: nil
       )
         @name = name
-        @attributes = attributes || {}
+
+        @attributes = attributes.dup || {}
+        validate_attributes!(@attributes)
+        @attributes.transform_keys!(&:to_s)
+
         # OpenTelemetry SDK stores span event timestamps in nanoseconds (not seconds).
         # We will do the same here to avoid unnecessary conversions and inconsistencies.
         @time_unix_nano = time_unix_nano || (Time.now.to_r * 1_000_000_000).to_i
       end
 
+      # Converts the span event into a hash to be used by with the span tag serialization
+      # (`span.set_tag('events) = [event.to_hash]`). This serialization format has the drawback
+      # of being limiting span events to the size limit of a span tag.
+      # All Datadog agents support this format.
       def to_hash
-        h = { name: @name, time_unix_nano: @time_unix_nano }
-        h[:attributes] = attributes unless @attributes.empty?
+        h = { 'name' => @name, 'time_unix_nano' => @time_unix_nano }
+        h['attributes'] = @attributes unless @attributes.empty?
+        h
+      end
+
+      # Converts the span event into a hash to be used by the MessagePack serialization as a
+      # top-level span field (span.span_events = [event.to_native_format]).
+      # This serialization format removes the serialization limitations of the `span.set_tag('events)` approach,
+      # but is only supported by newer version of the Datadog agent.
+      def to_native_format
+        h = { 'name' => @name, 'time_unix_nano' => @time_unix_nano }
+
+        attr = {}
+        @attributes.each do |key, value|
+          attr[key] = if value.is_a?(Array)
+                        { type: ARRAY_TYPE, array_value: value.map { |v| serialize_native_attribute(v) } }
+                      else
+                        serialize_native_attribute(value)
+                      end
+        end
+
+        h['attributes'] = attr unless @attributes.empty?
+
         h
       end
+
+      private
+
+      MIN_INT64_SIGNED = -2**63
+      MAX_INT64_SIGNED = 2 << 63 - 1
+
+      # Checks the attributes hash to ensure it only contains serializable values.
+      # Invalid values are removed from the hash.
+      def validate_attributes!(attributes)
+        attributes.select! do |key, value|
+          case value
+          when Array
+            next true if value.empty?
+
+            first = value.first
+            case first
+            when String, Integer, Float
+              first_type = first.class
+              if value.all? { |v| v.is_a?(first_type) }
+                value.all? { |v| validate_scalar_attribute!(key, v) }
+              else
+                Datadog.logger.warn("Attribute #{key} array must be homogenous: #{value}.")
+                false
+              end
+            when TrueClass, FalseClass
+              if value.all? { |v| v.is_a?(TrueClass) || v.is_a?(FalseClass) }
+                value.all? { |v| validate_scalar_attribute!(key, v) }
+              else
+                Datadog.logger.warn("Attribute #{key} array must be homogenous: #{value}.")
+                false
+              end
+            else
+              Datadog.logger.warn("Attribute #{key} must be a string, number, or boolean: #{value}.")
+              false
+            end
+          when String, Numeric, TrueClass, FalseClass
+            validate_scalar_attribute!(key, value)
+          else
+            Datadog.logger.warn("Attribute #{key} must be a string, number, boolean, or array: #{value}.")
+            false
+          end
+        end
+      end
+
+      def validate_scalar_attribute!(key, value)
+        case value
+        when String, TrueClass, FalseClass
+          true
+        when Integer
+          # Cannot be larger than signed 64-bit integer
+          if value < MIN_INT64_SIGNED || value > MAX_INT64_SIGNED
+            Datadog.logger.warn("Attribute #{key} must be within the range of a signed 64-bit integer: #{value}.")
+            false
+          else
+            true
+          end
+        when Float
+          # Has to be finite
+          return true if value.finite?
+
+          Datadog.logger.warn("Attribute #{key} must be a finite number: #{value}.")
+          false
+        else
+          Datadog.logger.warn("Attribute #{key} must be a string, number, or boolean: #{value}.")
+          false
+        end
+      end
+
+      STRING_TYPE = 0
+      BOOLEAN_TYPE = 1
+      INTEGER_TYPE = 2
+      DOUBLE_TYPE = 3
+      ARRAY_TYPE = 4
+
+      # Serializes individual scalar attributes into the native format.
+      def serialize_native_attribute(value)
+        case value
+        when String
+          { type: STRING_TYPE, string_value: value }
+        when TrueClass, FalseClass
+          { type: BOOLEAN_TYPE, bool_value: value }
+        when Integer
+          { type: INTEGER_TYPE, int_value: value }
+        when Float
+          { type: DOUBLE_TYPE, double_value: value }
+        else
+          # This is technically unreachable due to the validation in #initialize.
+          raise ArgumentError, "Attribute must be a string, number, or boolean: #{value}."
+        end
+      end
     end
   end
 end
diff --git a/lib/datadog/tracing/span_operation.rb b/lib/datadog/tracing/span_operation.rb
index d6fbde553fc..52a8fc469af 100644
--- a/lib/datadog/tracing/span_operation.rb
+++ b/lib/datadog/tracing/span_operation.rb
@@ -11,6 +11,8 @@
 require_relative 'metadata'
 require_relative 'metadata/ext'
 require_relative 'span'
+require_relative 'span_event'
+require_relative 'span_link'
 require_relative 'utils'
 
 module Datadog
diff --git a/lib/datadog/tracing/transport/serializable_trace.rb b/lib/datadog/tracing/transport/serializable_trace.rb
index 850e84f7707..380f8623954 100644
--- a/lib/datadog/tracing/transport/serializable_trace.rb
+++ b/lib/datadog/tracing/transport/serializable_trace.rb
@@ -12,8 +12,11 @@ class SerializableTrace
         attr_reader \
           :trace
 
-        def initialize(trace)
+        # @param trace [Datadog::Trace] the trace to serialize
+        # @param native_events_supported [Boolean] whether the agent supports span events as a top-level field
+        def initialize(trace, native_events_supported = false)
           @trace = trace
+          @native_events_supported = native_events_supported
         end
 
         # MessagePack serializer interface. Making this object
@@ -26,13 +29,13 @@ def initialize(trace)
         # @param packer [MessagePack::Packer] serialization buffer, can be +nil+ with JRuby
         def to_msgpack(packer = nil)
           # As of 1.3.3, JRuby implementation doesn't pass an existing packer
-          trace.spans.map { |s| SerializableSpan.new(s) }.to_msgpack(packer)
+          trace.spans.map { |s| SerializableSpan.new(s, @native_events_supported) }.to_msgpack(packer)
         end
 
         # JSON serializer interface.
         # Used by older version of the transport.
         def to_json(*args)
-          trace.spans.map { |s| SerializableSpan.new(s).to_hash }.to_json(*args)
+          trace.spans.map { |s| SerializableSpan.new(s, @native_events_supported).to_hash }.to_json(*args)
         end
       end
 
@@ -41,9 +44,12 @@ class SerializableSpan
         attr_reader \
           :span
 
-        def initialize(span)
+        # @param span [Datadog::Span] the span to serialize
+        # @param native_events_supported [Boolean] whether the agent supports span events as a top-level field
+        def initialize(span, native_events_supported)
           @span = span
           @trace_id = Tracing::Utils::TraceId.to_low_order(span.trace_id)
+          @native_events_supported = native_events_supported
         end
 
         # MessagePack serializer interface. Making this object
@@ -55,11 +61,14 @@ def initialize(span)
         #
         # @param packer [MessagePack::Packer] serialization buffer, can be +nil+ with JRuby
         # rubocop:disable Metrics/AbcSize
+        # rubocop:disable Metrics/MethodLength
         def to_msgpack(packer = nil)
           packer ||= MessagePack::Packer.new
 
           number_of_elements_to_write = 11
 
+          number_of_elements_to_write += 1 if span.events.any? && @native_events_supported
+
           if span.stopped?
             packer.write_map_header(number_of_elements_to_write + 2) # Set header with how many elements in the map
 
@@ -72,8 +81,16 @@ def to_msgpack(packer = nil)
             packer.write_map_header(number_of_elements_to_write) # Set header with how many elements in the map
           end
 
-          # serialize span events as meta tags
-          span.set_tag('events', span.events.map(&:to_hash).to_json) if span.events.any?
+          if span.events.any?
+            if @native_events_supported
+              # Use top-level field for native events
+              packer.write('span_events')
+              packer.write(span.events.map(&:to_native_format))
+            else
+              # Serialize span events as meta tags
+              span.set_tag('events', span.events.map(&:to_hash).to_json)
+            end
+          end
 
           # DEV: We use strings as keys here, instead of symbols, as
           # DEV: MessagePack will ultimately convert them to strings.
@@ -103,6 +120,7 @@ def to_msgpack(packer = nil)
           packer
         end
         # rubocop:enable Metrics/AbcSize
+        # rubocop:enable Metrics/MethodLength
 
         # JSON serializer interface.
         # Used by older version of the transport.
diff --git a/sig/datadog/appsec/contrib/graphql/reactive/multiplex.rbs b/sig/datadog/appsec/contrib/graphql/reactive/multiplex.rbs
index 7095e94da50..d0cbad649d8 100644
--- a/sig/datadog/appsec/contrib/graphql/reactive/multiplex.rbs
+++ b/sig/datadog/appsec/contrib/graphql/reactive/multiplex.rbs
@@ -4,11 +4,11 @@ module Datadog
       module GraphQL
         module Reactive
           module Multiplex
-            ADDRESSES: ::Array[String]
+            ADDRESSES: ::Array[::String]
 
-            def self.publish: (Datadog::AppSec::Reactive::Operation op, Datadog::AppSec::Contrib::GraphQL::Gateway::Multiplex gateway_multiplex) -> boolish
+            def self.publish: (AppSec::Reactive::Engine engine, AppSec::Contrib::GraphQL::Gateway::Multiplex gateway_multiplex) -> boolish
 
-            def self.subscribe: (Datadog::AppSec::Reactive::Operation op, Datadog::AppSec::Contrib::GraphQL::Gateway::Multiplex waf_context) -> void
+            def self.subscribe: (AppSec::Reactive::Engine engine, AppSec::Contrib::GraphQL::Gateway::Multiplex waf_context) -> void
           end
         end
       end
diff --git a/sig/datadog/appsec/contrib/rack/reactive/request.rbs b/sig/datadog/appsec/contrib/rack/reactive/request.rbs
index 7163eff15e3..d55f00a447a 100644
--- a/sig/datadog/appsec/contrib/rack/reactive/request.rbs
+++ b/sig/datadog/appsec/contrib/rack/reactive/request.rbs
@@ -6,9 +6,9 @@ module Datadog
           module Request
             ADDRESSES: ::Array[::String]
 
-            def self.publish: (untyped op, Datadog::AppSec::Contrib::Rack::Gateway::Request request) -> untyped
+            def self.publish: (AppSec::Reactive::Engine engine, AppSec::Contrib::Rack::Gateway::Request request) -> untyped
 
-            def self.subscribe: (untyped op, untyped waf_context) { (untyped) -> untyped } -> untyped
+            def self.subscribe: (AppSec::Reactive::Engine engine, untyped waf_context) { (untyped) -> untyped } -> untyped
           end
         end
       end
diff --git a/sig/datadog/appsec/contrib/rack/reactive/request_body.rbs b/sig/datadog/appsec/contrib/rack/reactive/request_body.rbs
index 3d04cb7c5ed..7747040eb6a 100644
--- a/sig/datadog/appsec/contrib/rack/reactive/request_body.rbs
+++ b/sig/datadog/appsec/contrib/rack/reactive/request_body.rbs
@@ -6,9 +6,9 @@ module Datadog
           module RequestBody
             ADDRESSES: ::Array[::String]
 
-            def self.publish: (untyped op, Datadog::AppSec::Contrib::Rack::Gateway::Response request) -> untyped
+            def self.publish: (AppSec::Reactive::Engine engine, AppSec::Contrib::Rack::Gateway::Response request) -> untyped
 
-            def self.subscribe: (untyped op, untyped waf_context) { (untyped) -> untyped } -> untyped
+            def self.subscribe: (AppSec::Reactive::Engine engine, untyped waf_context) { (untyped) -> untyped } -> untyped
           end
         end
       end
diff --git a/sig/datadog/appsec/contrib/rack/reactive/response.rbs b/sig/datadog/appsec/contrib/rack/reactive/response.rbs
index 2d2a9edddc1..a79a98b0da6 100644
--- a/sig/datadog/appsec/contrib/rack/reactive/response.rbs
+++ b/sig/datadog/appsec/contrib/rack/reactive/response.rbs
@@ -6,9 +6,9 @@ module Datadog
           module Response
             ADDRESSES: ::Array[::String]
 
-            def self.publish: (untyped op, Datadog::AppSec::Contrib::Rack::Gateway::Response response) -> untyped
+            def self.publish: (AppSec::Reactive::Engine engine, AppSec::Contrib::Rack::Gateway::Response response) -> untyped
 
-            def self.subscribe: (untyped op, untyped waf_context) { (untyped) -> untyped } -> untyped
+            def self.subscribe: (AppSec::Reactive::Engine engine, untyped waf_context) { (untyped) -> untyped } -> untyped
           end
         end
       end
diff --git a/sig/datadog/appsec/contrib/rails/reactive/action.rbs b/sig/datadog/appsec/contrib/rails/reactive/action.rbs
index f24f3c62ff5..e442469cb1b 100644
--- a/sig/datadog/appsec/contrib/rails/reactive/action.rbs
+++ b/sig/datadog/appsec/contrib/rails/reactive/action.rbs
@@ -6,9 +6,9 @@ module Datadog
           module Action
             ADDRESSES: ::Array[::String]
 
-            def self.publish: (untyped op, Datadog::AppSec::Contrib::Rails::Gateway::Request request) -> untyped
+            def self.publish: (AppSec::Reactive::Engine engine, AppSec::Contrib::Rails::Gateway::Request request) -> untyped
 
-            def self.subscribe: (untyped op, untyped waf_context) { (untyped) -> untyped } -> untyped
+            def self.subscribe: (AppSec::Reactive::Engine engine, untyped waf_context) { (untyped) -> untyped } -> untyped
           end
         end
       end
diff --git a/sig/datadog/appsec/contrib/sinatra/reactive/routed.rbs b/sig/datadog/appsec/contrib/sinatra/reactive/routed.rbs
index de3fa0a0919..a711268f7d4 100644
--- a/sig/datadog/appsec/contrib/sinatra/reactive/routed.rbs
+++ b/sig/datadog/appsec/contrib/sinatra/reactive/routed.rbs
@@ -6,9 +6,9 @@ module Datadog
           module Routed
             ADDRESSES: ::Array[::String]
 
-            def self.publish: (untyped op, ::Array[Datadog::AppSec::Instrumentation::Gateway::Argument] data) -> untyped
+            def self.publish: (AppSec::Reactive::Engine engine, ::Array[AppSec::Instrumentation::Gateway::Argument] data) -> untyped
 
-            def self.subscribe: (untyped op, untyped waf_context) { (untyped) -> untyped } -> untyped
+            def self.subscribe: (AppSec::Reactive::Engine engine, untyped waf_context) { (untyped) -> untyped } -> untyped
           end
         end
       end
diff --git a/sig/datadog/appsec/monitor/reactive/set_user.rbs b/sig/datadog/appsec/monitor/reactive/set_user.rbs
index 49fa18ff860..1efd5ff8277 100644
--- a/sig/datadog/appsec/monitor/reactive/set_user.rbs
+++ b/sig/datadog/appsec/monitor/reactive/set_user.rbs
@@ -5,9 +5,9 @@ module Datadog
         module SetUser
           ADDRESSES: ::Array[::String]
 
-          def self.publish: (untyped op, Datadog::AppSec::Instrumentation::Gateway::User user) -> untyped
+          def self.publish: (AppSec::Reactive::Engine engine, AppSec::Instrumentation::Gateway::User user) -> untyped
 
-          def self.subscribe: (untyped op, untyped waf_context) { (untyped) -> untyped } -> untyped
+          def self.subscribe: (AppSec::Reactive::Engine engine, untyped waf_context) { (untyped) -> untyped } -> untyped
         end
       end
     end
diff --git a/sig/datadog/appsec/reactive/operation.rbs b/sig/datadog/appsec/reactive/operation.rbs
deleted file mode 100644
index 854e7c9dc6b..00000000000
--- a/sig/datadog/appsec/reactive/operation.rbs
+++ /dev/null
@@ -1,25 +0,0 @@
-module Datadog
-  module AppSec
-    module Reactive
-      class Operation
-        attr_reader reactive: Engine
-        attr_reader parent: Operation?
-        attr_reader name: ::String
-
-        def initialize: (::String name, ?Operation? parent, ?Engine? reactive_engine) ?{ (Operation) -> void } -> void
-        def logger: () -> ::Logger
-        def subscribe: (*::String addresses) { (*untyped values) -> void } -> void
-        def publish: (::String address, untyped data) -> untyped
-        def finalize: () -> void
-
-        private
-
-        def select_reactive_engine: (Engine? reactive, Operation? parent) -> Engine
-
-        def self.active: () -> Operation?
-
-        def self.reset!: () -> untyped
-      end
-    end
-  end
-end
diff --git a/sig/datadog/di/error.rbs b/sig/datadog/di/error.rbs
index 351fa7bb85a..4812de578a7 100644
--- a/sig/datadog/di/error.rbs
+++ b/sig/datadog/di/error.rbs
@@ -7,6 +7,8 @@ module Datadog
       end
       class DITargetNotDefined < Error
       end
+      class DITargetNotInRegistry < Error
+      end
       class ProbePreviouslyFailed < Error
       end
       class MultiplePathsMatch < Error
diff --git a/sig/datadog/di/instrumenter.rbs b/sig/datadog/di/instrumenter.rbs
index cfaa56cac64..f87c2818cf4 100644
--- a/sig/datadog/di/instrumenter.rbs
+++ b/sig/datadog/di/instrumenter.rbs
@@ -48,6 +48,7 @@ module Datadog
 
       attr_reader lock: untyped
       def symbolize_class_name: (untyped cls_name) -> untyped
+      def raise_if_probe_in_loaded_features: (Probe probe) -> void
     end
   end
 end
diff --git a/sig/datadog/tracing/span_event.rbs b/sig/datadog/tracing/span_event.rbs
index eff90ad371b..643c3062fc1 100644
--- a/sig/datadog/tracing/span_event.rbs
+++ b/sig/datadog/tracing/span_event.rbs
@@ -1,27 +1,47 @@
 module Datadog
   module Tracing
-    # SpanEvent represents an annotation on a span.
-    # @public_api
     class SpanEvent
-      @name: untyped
+      type attributes = Hash[String,attributeValue]
+      type attributeValue = String | Integer | Float | bool | Array[String] | Array[Integer] | Array[Float] | Array[bool]
 
-      @attributes: untyped
+      MIN_INT64_SIGNED: Integer
+      MAX_INT64_SIGNED: Integer
+      STRING_TYPE: Integer
+      BOOLEAN_TYPE: Integer
+      INTEGER_TYPE: Integer
+      DOUBLE_TYPE: Integer
+      ARRAY_TYPE: Integer
 
-      @time_unix_nano: untyped
+      attr_reader name: untyped # TODO: Typing this makes to_hash internal typecheck fail
+      attr_reader attributes: attributes
+      attr_reader time_unix_nano: untyped # TODO: Typing this also makes to_hash internal typecheck fail
 
-      # @!attribute [r] name
-      #   @return [String]
-      attr_reader name: untyped
+      def initialize: (String name, ?attributes: attributes, ?time_unix_nano: Integer) -> void
 
-      # @!attribute [r] attributes
-      #   @return [Hash<String,String>]
-      attr_reader attributes: untyped
+      def to_hash: -> Hash[String, untyped]
+      # TODO: Steep does not track Hash keys when they are added with `hash[:key] = val`.
+      #   {
+      #   name: String,
+      #   time_unix_nano: Integer,
+      #   ?attributes: attributes,
+      # }
 
-      # @!attribute [r] time_unix_nano
-      #   @return [Integer]
-      attr_reader time_unix_nano: untyped
+      def to_native_format: -> Hash[String, untyped]
+      # TODO: Steep does not track Hash keys when they are added with `hash[:key] = val`.
+      # {
+      #   name: String,
+      #   time_unix_nano: Integer,
+      #   ?attributes: Hash[String, nativeAttributeValue],
+      # }
+      # type nativeAttributeValue = { type: Integer, string_value: String } | { type: Integer, int_value: Integer } | { type: Integer, double_value: Float } | { type: Integer, bool_value: bool } | { type: Integer, string_array_value: Array[String] } | { type: Integer, int_array_value: Array[Integer] } | { type: Integer, double_array_value: Array[Float] } | { type: Integer, bool_array_value: Array[bool] }
 
-      def initialize: (untyped name, ?attributes: untyped?, ?time_unix_nano: untyped?) -> void
+      private
+
+      def serialize_native_attribute: (attributeValue value)-> ({ type: Integer, string_value: String } | { type: Integer, int_value: Integer } | { type: Integer, double_value: Float } | { type: Integer, bool_value: bool })
+
+      def validate_attributes!: (attributes attributes)-> void
+
+      def validate_scalar_attribute!: (String key, attributeValue value)-> bool
     end
   end
 end
\ No newline at end of file
diff --git a/sig/datadog/tracing/transport/serializable_trace.rbs b/sig/datadog/tracing/transport/serializable_trace.rbs
index b318bed50f3..ab84fcfc23d 100644
--- a/sig/datadog/tracing/transport/serializable_trace.rbs
+++ b/sig/datadog/tracing/transport/serializable_trace.rbs
@@ -2,9 +2,11 @@ module Datadog
   module Tracing
     module Transport
       class SerializableTrace
-        attr_reader trace: untyped
+        @native_events_supported: bool
 
-        def initialize: (untyped trace) -> void
+        attr_reader trace: Span
+
+        def initialize: (untyped trace, bool native_events_supported) -> void
 
         def to_msgpack: (?untyped? packer) -> untyped
 
@@ -12,9 +14,12 @@ module Datadog
       end
 
       class SerializableSpan
-        attr_reader span: untyped
+        @native_events_supported: bool
+        @trace_id: Integer
+
+        attr_reader span: Span
 
-        def initialize: (untyped span) -> void
+        def initialize: (untyped span, bool native_events_supported) -> void
 
         def to_msgpack: (?untyped? packer) -> untyped
 
diff --git a/spec/datadog/appsec/contrib/graphql/reactive/multiplex_spec.rb b/spec/datadog/appsec/contrib/graphql/reactive/multiplex_spec.rb
index 519904cfc5d..21df7b7b6d5 100644
--- a/spec/datadog/appsec/contrib/graphql/reactive/multiplex_spec.rb
+++ b/spec/datadog/appsec/contrib/graphql/reactive/multiplex_spec.rb
@@ -4,14 +4,15 @@
 require 'datadog/tracing/contrib/graphql/support/application'
 
 require 'datadog/appsec/spec_helper'
-require 'datadog/appsec/reactive/operation'
 require 'datadog/appsec/contrib/graphql/gateway/multiplex'
 require 'datadog/appsec/contrib/graphql/reactive/multiplex'
+require 'datadog/appsec/reactive/engine'
 require 'datadog/appsec/reactive/shared_examples'
 
 RSpec.describe Datadog::AppSec::Contrib::GraphQL::Reactive::Multiplex do
   include_context 'with GraphQL multiplex'
 
+  let(:engine) { Datadog::AppSec::Reactive::Engine.new }
   let(:expected_arguments) do
     {
       'user' => [{ 'id' => 1 }, { 'id' => 10 }],
@@ -20,10 +21,10 @@
   end
 
   describe '.publish' do
-    it 'propagates multiplex attributes to the operation' do
-      expect(operation).to receive(:publish).with('graphql.server.all_resolvers', expected_arguments)
+    it 'propagates multiplex attributes to the engine' do
+      expect(engine).to receive(:publish).with('graphql.server.all_resolvers', expected_arguments)
       gateway_multiplex = Datadog::AppSec::Contrib::GraphQL::Gateway::Multiplex.new(multiplex)
-      described_class.publish(operation, gateway_multiplex)
+      described_class.publish(engine, gateway_multiplex)
     end
   end
 
@@ -32,17 +33,17 @@
 
     context 'not all addresses have been published' do
       it 'does not call the waf context' do
-        expect(operation).to receive(:subscribe).with(
+        expect(engine).to receive(:subscribe).with(
           'graphql.server.all_resolvers'
         ).and_call_original
         expect(waf_context).to_not receive(:run)
-        described_class.subscribe(operation, waf_context)
+        described_class.subscribe(engine, waf_context)
       end
     end
 
     context 'all addresses have been published' do
       it 'does call the waf context with the right arguments' do
-        expect(operation).to receive(:subscribe).and_call_original
+        expect(engine).to receive(:subscribe).and_call_original
 
         waf_result = double(:waf_result, status: :ok, timeout: false)
         expect(waf_context).to receive(:run).with(
@@ -50,9 +51,9 @@
           {},
           Datadog.configuration.appsec.waf_timeout
         ).and_return(waf_result)
-        described_class.subscribe(operation, waf_context)
+        described_class.subscribe(engine, waf_context)
         gateway_multiplex = Datadog::AppSec::Contrib::GraphQL::Gateway::Multiplex.new(multiplex)
-        result = described_class.publish(operation, gateway_multiplex)
+        result = described_class.publish(engine, gateway_multiplex)
         expect(result).to be_nil
       end
     end
diff --git a/spec/datadog/appsec/contrib/rack/reactive/request_body_spec.rb b/spec/datadog/appsec/contrib/rack/reactive/request_body_spec.rb
index 7b5ce95a3a2..75cf9aae480 100644
--- a/spec/datadog/appsec/contrib/rack/reactive/request_body_spec.rb
+++ b/spec/datadog/appsec/contrib/rack/reactive/request_body_spec.rb
@@ -1,15 +1,15 @@
 # frozen_string_literal: true
 
 require 'datadog/appsec/spec_helper'
-require 'datadog/appsec/reactive/operation'
 require 'datadog/appsec/contrib/rack/gateway/request'
 require 'datadog/appsec/contrib/rack/reactive/request_body'
+require 'datadog/appsec/reactive/engine'
 require 'datadog/appsec/reactive/shared_examples'
 
 require 'rack'
 
 RSpec.describe Datadog::AppSec::Contrib::Rack::Reactive::RequestBody do
-  let(:operation) { Datadog::AppSec::Reactive::Operation.new('test') }
+  let(:engine) { Datadog::AppSec::Reactive::Engine.new }
   let(:request) do
     Datadog::AppSec::Contrib::Rack::Gateway::Request.new(
       Rack::MockRequest.env_for(
@@ -20,10 +20,10 @@
   end
 
   describe '.publish' do
-    it 'propagates request body attributes to the operation' do
-      expect(operation).to receive(:publish).with('request.body', { 'foo' => 'bar' })
+    it 'propagates request body attributes to the engine' do
+      expect(engine).to receive(:publish).with('request.body', { 'foo' => 'bar' })
 
-      described_class.publish(operation, request)
+      described_class.publish(engine, request)
     end
   end
 
@@ -32,15 +32,15 @@
 
     context 'not all addresses have been published' do
       it 'does not call the waf context' do
-        expect(operation).to receive(:subscribe).with('request.body').and_call_original
+        expect(engine).to receive(:subscribe).with('request.body').and_call_original
         expect(waf_context).to_not receive(:run)
-        described_class.subscribe(operation, waf_context)
+        described_class.subscribe(engine, waf_context)
       end
     end
 
     context 'all addresses have been published' do
       it 'does call the waf context with the right arguments' do
-        expect(operation).to receive(:subscribe).and_call_original
+        expect(engine).to receive(:subscribe).and_call_original
 
         expected_waf_arguments = { 'server.request.body' => { 'foo' => 'bar' } }
 
@@ -50,8 +50,8 @@
           {},
           Datadog.configuration.appsec.waf_timeout
         ).and_return(waf_result)
-        described_class.subscribe(operation, waf_context)
-        result = described_class.publish(operation, request)
+        described_class.subscribe(engine, waf_context)
+        result = described_class.publish(engine, request)
         expect(result).to be_nil
       end
     end
diff --git a/spec/datadog/appsec/contrib/rack/reactive/request_spec.rb b/spec/datadog/appsec/contrib/rack/reactive/request_spec.rb
index b9927ce8f87..f22a3964b1d 100644
--- a/spec/datadog/appsec/contrib/rack/reactive/request_spec.rb
+++ b/spec/datadog/appsec/contrib/rack/reactive/request_spec.rb
@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 require 'datadog/appsec/spec_helper'
-require 'datadog/appsec/reactive/operation'
+require 'datadog/appsec/reactive/engine'
 require 'datadog/appsec/contrib/rack/gateway/request'
 require 'datadog/appsec/contrib/rack/reactive/request'
 require 'datadog/appsec/reactive/shared_examples'
@@ -9,7 +9,7 @@
 require 'rack'
 
 RSpec.describe Datadog::AppSec::Contrib::Rack::Reactive::Request do
-  let(:operation) { Datadog::AppSec::Reactive::Operation.new('test') }
+  let(:engine) { Datadog::AppSec::Reactive::Engine.new }
   let(:request) do
     Datadog::AppSec::Contrib::Rack::Gateway::Request.new(
       Rack::MockRequest.env_for(
@@ -34,15 +34,15 @@
   end
 
   describe '.publish' do
-    it 'propagates request attributes to the operation' do
-      expect(operation).to receive(:publish).with('server.request.method', 'GET')
-      expect(operation).to receive(:publish).with('request.query', { 'a' => ['foo'] })
-      expect(operation).to receive(:publish).with('request.headers', expected_headers_with_cookies)
-      expect(operation).to receive(:publish).with('request.uri.raw', '/?a=foo')
-      expect(operation).to receive(:publish).with('request.cookies', { 'foo' => 'bar' })
-      expect(operation).to receive(:publish).with('request.client_ip', '10.10.10.10')
+    it 'propagates request attributes to the engine' do
+      expect(engine).to receive(:publish).with('server.request.method', 'GET')
+      expect(engine).to receive(:publish).with('request.query', { 'a' => ['foo'] })
+      expect(engine).to receive(:publish).with('request.headers', expected_headers_with_cookies)
+      expect(engine).to receive(:publish).with('request.uri.raw', '/?a=foo')
+      expect(engine).to receive(:publish).with('request.cookies', { 'foo' => 'bar' })
+      expect(engine).to receive(:publish).with('request.client_ip', '10.10.10.10')
 
-      described_class.publish(operation, request)
+      described_class.publish(engine, request)
     end
   end
 
@@ -51,7 +51,7 @@
 
     context 'not all addresses have been published' do
       it 'does not call the waf context' do
-        expect(operation).to receive(:subscribe).with(
+        expect(engine).to receive(:subscribe).with(
           'request.headers',
           'request.uri.raw',
           'request.query',
@@ -60,13 +60,13 @@
           'server.request.method',
         ).and_call_original
         expect(waf_context).to_not receive(:run)
-        described_class.subscribe(operation, waf_context)
+        described_class.subscribe(engine, waf_context)
       end
     end
 
     context 'all addresses have been published' do
       it 'does call the waf context with the right arguments' do
-        expect(operation).to receive(:subscribe).and_call_original
+        expect(engine).to receive(:subscribe).and_call_original
 
         expected_waf_arguments = {
           'server.request.cookies' => { 'foo' => 'bar' },
@@ -84,8 +84,8 @@
           {},
           Datadog.configuration.appsec.waf_timeout
         ).and_return(waf_result)
-        described_class.subscribe(operation, waf_context)
-        result = described_class.publish(operation, request)
+        described_class.subscribe(engine, waf_context)
+        result = described_class.publish(engine, request)
         expect(result).to be_nil
       end
     end
diff --git a/spec/datadog/appsec/contrib/rack/reactive/response_spec.rb b/spec/datadog/appsec/contrib/rack/reactive/response_spec.rb
index 8ea92c75e78..ca51117f5e1 100644
--- a/spec/datadog/appsec/contrib/rack/reactive/response_spec.rb
+++ b/spec/datadog/appsec/contrib/rack/reactive/response_spec.rb
@@ -2,13 +2,13 @@
 
 require 'datadog/appsec/spec_helper'
 require 'datadog/appsec/scope'
-require 'datadog/appsec/reactive/operation'
 require 'datadog/appsec/contrib/rack/gateway/response'
 require 'datadog/appsec/contrib/rack/reactive/response'
+require 'datadog/appsec/reactive/engine'
 require 'datadog/appsec/reactive/shared_examples'
 
 RSpec.describe Datadog::AppSec::Contrib::Rack::Reactive::Response do
-  let(:operation) { Datadog::AppSec::Reactive::Operation.new('test') }
+  let(:engine) { Datadog::AppSec::Reactive::Engine.new }
   let(:processor_context) { instance_double(Datadog::AppSec::Processor::Context) }
   let(:scope) { instance_double(Datadog::AppSec::Scope, processor_context: processor_context) }
   let(:body) { ['Ok'] }
@@ -24,31 +24,31 @@
   end
 
   describe '.publish' do
-    it 'propagates response attributes to the operation' do
-      expect(operation).to receive(:publish).with('response.status', 200)
-      expect(operation).to receive(:publish).with(
+    it 'propagates response attributes to the engine' do
+      expect(engine).to receive(:publish).with('response.status', 200)
+      expect(engine).to receive(:publish).with(
         'response.headers',
         headers,
       )
-      described_class.publish(operation, response)
+      described_class.publish(engine, response)
     end
   end
 
   describe '.subscribe' do
     context 'not all addresses have been published' do
       it 'does not call the waf context' do
-        expect(operation).to receive(:subscribe).with(
+        expect(engine).to receive(:subscribe).with(
           'response.status',
           'response.headers',
         ).and_call_original
         expect(processor_context).to_not receive(:run)
-        described_class.subscribe(operation, processor_context)
+        described_class.subscribe(engine, processor_context)
       end
     end
 
     context 'waf arguments' do
       before do
-        expect(operation).to receive(:subscribe).and_call_original
+        expect(engine).to receive(:subscribe).and_call_original
       end
 
       let(:waf_result) { double(:waf_result, status: :ok, timeout: false) }
@@ -73,8 +73,8 @@
             {},
             Datadog.configuration.appsec.waf_timeout
           ).and_return(waf_result)
-          described_class.subscribe(operation, processor_context)
-          result = described_class.publish(operation, response)
+          described_class.subscribe(engine, processor_context)
+          result = described_class.publish(engine, response)
           expect(result).to be_nil
         end
       end
diff --git a/spec/datadog/appsec/contrib/rails/integration_test_spec.rb b/spec/datadog/appsec/contrib/rails/integration_test_spec.rb
index b564e477c9f..1c1b7eecaad 100644
--- a/spec/datadog/appsec/contrib/rails/integration_test_spec.rb
+++ b/spec/datadog/appsec/contrib/rails/integration_test_spec.rb
@@ -1,3 +1,5 @@
+# frozen_string_literal: true
+
 require 'datadog/tracing/contrib/rails/rails_helper'
 require 'datadog/appsec/contrib/support/integration/shared_examples'
 require 'datadog/appsec/spec_helper'
diff --git a/spec/datadog/appsec/contrib/rails/reactive/action_spec.rb b/spec/datadog/appsec/contrib/rails/reactive/action_spec.rb
index 75c302e2751..fad066ef7cf 100644
--- a/spec/datadog/appsec/contrib/rails/reactive/action_spec.rb
+++ b/spec/datadog/appsec/contrib/rails/reactive/action_spec.rb
@@ -1,13 +1,15 @@
+# frozen_string_literal: true
+
 require 'datadog/appsec/spec_helper'
-require 'datadog/appsec/reactive/operation'
 require 'datadog/appsec/contrib/rails/reactive/action'
 require 'datadog/appsec/contrib/rails/gateway/request'
+require 'datadog/appsec/reactive/engine'
 require 'datadog/appsec/reactive/shared_examples'
 
 require 'action_dispatch'
 
 RSpec.describe Datadog::AppSec::Contrib::Rails::Reactive::Action do
-  let(:operation) { Datadog::AppSec::Reactive::Operation.new('test') }
+  let(:engine) { Datadog::AppSec::Reactive::Engine.new }
   let(:request) do
     request_env = Rack::MockRequest.env_for(
       'http://example.com:8080/?a=foo',
@@ -24,11 +26,11 @@
   end
 
   describe '.publish' do
-    it 'propagates request attributes to the operation' do
-      expect(operation).to receive(:publish).with('rails.request.body', { 'foo' => 'bar' })
-      expect(operation).to receive(:publish).with('rails.request.route_params', { id: '1234' })
+    it 'propagates request attributes to the engine' do
+      expect(engine).to receive(:publish).with('rails.request.body', { 'foo' => 'bar' })
+      expect(engine).to receive(:publish).with('rails.request.route_params', { id: '1234' })
 
-      described_class.publish(operation, request)
+      described_class.publish(engine, request)
     end
   end
 
@@ -37,15 +39,15 @@
 
     context 'not all addresses have been published' do
       it 'does not call the waf context' do
-        expect(operation).to receive(:subscribe).with('rails.request.body', 'rails.request.route_params').and_call_original
+        expect(engine).to receive(:subscribe).with('rails.request.body', 'rails.request.route_params').and_call_original
         expect(waf_context).to_not receive(:run)
-        described_class.subscribe(operation, waf_context)
+        described_class.subscribe(engine, waf_context)
       end
     end
 
     context 'all addresses have been published' do
       it 'does call the waf context with the right arguments' do
-        expect(operation).to receive(:subscribe).and_call_original
+        expect(engine).to receive(:subscribe).and_call_original
 
         expected_waf_arguments = {
           'server.request.body' => { 'foo' => 'bar' },
@@ -58,8 +60,8 @@
           {},
           Datadog.configuration.appsec.waf_timeout
         ).and_return(waf_result)
-        described_class.subscribe(operation, waf_context)
-        result = described_class.publish(operation, request)
+        described_class.subscribe(engine, waf_context)
+        result = described_class.publish(engine, request)
         expect(result).to be_nil
       end
     end
diff --git a/spec/datadog/appsec/contrib/sinatra/reactive/routed_spec.rb b/spec/datadog/appsec/contrib/sinatra/reactive/routed_spec.rb
index b30ca74c3f6..690d1609ecb 100644
--- a/spec/datadog/appsec/contrib/sinatra/reactive/routed_spec.rb
+++ b/spec/datadog/appsec/contrib/sinatra/reactive/routed_spec.rb
@@ -1,16 +1,16 @@
 # frozen_string_literal: true
 
 require 'datadog/appsec/spec_helper'
-require 'datadog/appsec/reactive/operation'
 require 'datadog/appsec/contrib/sinatra/reactive/routed'
 require 'datadog/appsec/contrib/rack/gateway/request'
 require 'datadog/appsec/contrib/sinatra/gateway/route_params'
+require 'datadog/appsec/reactive/engine'
 require 'datadog/appsec/reactive/shared_examples'
 
 require 'rack'
 
 RSpec.describe Datadog::AppSec::Contrib::Sinatra::Reactive::Routed do
-  let(:operation) { Datadog::AppSec::Reactive::Operation.new('test') }
+  let(:engine) { Datadog::AppSec::Reactive::Engine.new }
   let(:request) do
     Datadog::AppSec::Contrib::Rack::Gateway::Request.new(
       Rack::MockRequest.env_for(
@@ -22,10 +22,10 @@
   let(:routed_params) { Datadog::AppSec::Contrib::Sinatra::Gateway::RouteParams.new({ id: '1234' }) }
 
   describe '.publish' do
-    it 'propagates routed params attributes to the operation' do
-      expect(operation).to receive(:publish).with('sinatra.request.route_params', { id: '1234' })
+    it 'propagates routed params attributes to the engine' do
+      expect(engine).to receive(:publish).with('sinatra.request.route_params', { id: '1234' })
 
-      described_class.publish(operation, [request, routed_params])
+      described_class.publish(engine, [request, routed_params])
     end
   end
 
@@ -34,15 +34,15 @@
 
     context 'not all addresses have been published' do
       it 'does not call the waf context' do
-        expect(operation).to receive(:subscribe).with('sinatra.request.route_params').and_call_original
+        expect(engine).to receive(:subscribe).with('sinatra.request.route_params').and_call_original
         expect(waf_context).to_not receive(:run)
-        described_class.subscribe(operation, waf_context)
+        described_class.subscribe(engine, waf_context)
       end
     end
 
     context 'all addresses have been published' do
       it 'does call the waf context with the right arguments' do
-        expect(operation).to receive(:subscribe).and_call_original
+        expect(engine).to receive(:subscribe).and_call_original
 
         expected_waf_arguments = {
           'server.request.path_params' => { id: '1234' }
@@ -54,8 +54,8 @@
           {},
           Datadog.configuration.appsec.waf_timeout
         ).and_return(waf_result)
-        described_class.subscribe(operation, waf_context)
-        result = described_class.publish(operation, [request, routed_params])
+        described_class.subscribe(engine, waf_context)
+        result = described_class.publish(engine, [request, routed_params])
         expect(result).to be_nil
       end
     end
diff --git a/spec/datadog/appsec/monitor/reactive/set_user_spec.rb b/spec/datadog/appsec/monitor/reactive/set_user_spec.rb
index e29b076ffa0..5e5472f6b2a 100644
--- a/spec/datadog/appsec/monitor/reactive/set_user_spec.rb
+++ b/spec/datadog/appsec/monitor/reactive/set_user_spec.rb
@@ -1,19 +1,19 @@
 # frozen_string_literal: true
 
 require 'datadog/appsec/spec_helper'
-require 'datadog/appsec/reactive/operation'
+require 'datadog/appsec/reactive/engine'
 require 'datadog/appsec/monitor/reactive/set_user'
 require 'datadog/appsec/reactive/shared_examples'
 
 RSpec.describe Datadog::AppSec::Monitor::Reactive::SetUser do
-  let(:operation) { Datadog::AppSec::Reactive::Operation.new('test') }
+  let(:engine) { Datadog::AppSec::Reactive::Engine.new }
   let(:user) { double(:user, id: 1) }
 
   describe '.publish' do
-    it 'propagates request body attributes to the operation' do
-      expect(operation).to receive(:publish).with('usr.id', 1)
+    it 'propagates request body attributes to the engine' do
+      expect(engine).to receive(:publish).with('usr.id', 1)
 
-      described_class.publish(operation, user)
+      described_class.publish(engine, user)
     end
   end
 
@@ -22,15 +22,15 @@
 
     context 'not all addresses have been published' do
       it 'does not call the waf context' do
-        expect(operation).to receive(:subscribe).with('usr.id').and_call_original
+        expect(engine).to receive(:subscribe).with('usr.id').and_call_original
         expect(waf_context).to_not receive(:run)
-        described_class.subscribe(operation, waf_context)
+        described_class.subscribe(engine, waf_context)
       end
     end
 
     context 'all addresses have been published' do
       it 'does call the waf context with the right arguments' do
-        expect(operation).to receive(:subscribe).and_call_original
+        expect(engine).to receive(:subscribe).and_call_original
 
         expected_waf_persisted_data = { 'usr.id' => 1 }
 
@@ -40,8 +40,8 @@
           {},
           Datadog.configuration.appsec.waf_timeout
         ).and_return(waf_result)
-        described_class.subscribe(operation, waf_context)
-        result = described_class.publish(operation, user)
+        described_class.subscribe(engine, waf_context)
+        result = described_class.publish(engine, user)
         expect(result).to be_nil
       end
     end
diff --git a/spec/datadog/appsec/reactive/operation_spec.rb b/spec/datadog/appsec/reactive/operation_spec.rb
deleted file mode 100644
index 81c0aff0f04..00000000000
--- a/spec/datadog/appsec/reactive/operation_spec.rb
+++ /dev/null
@@ -1,86 +0,0 @@
-# frozen_string_literal: true
-
-require 'datadog/appsec/spec_helper'
-require 'datadog/appsec/reactive/operation'
-
-RSpec.describe Datadog::AppSec::Reactive::Operation do
-  after do
-    described_class.send(:reset!)
-  end
-
-  describe '#initialize' do
-    it 'sets active to yield operation for the duration of the block' do
-      active_operation = described_class.active
-      expect(active_operation).to be_nil
-      described_class.new('test') do |op|
-        expect(described_class.active).to eq(op)
-      end
-      expect(described_class.active).to be_nil
-    end
-
-    it 'sets active to parent' do
-      parent_operation = described_class.new('parent_test')
-      described_class.new('test', parent_operation) do |op|
-        expect(described_class.active).to eq(op)
-      end
-      expect(described_class.active).to eq(parent_operation)
-    end
-
-    it 'creates a new Reactive instance when no reactive instance provided' do
-      described_class.new('test') do |op|
-        expect(op.reactive).to be_a Datadog::AppSec::Reactive::Engine
-      end
-    end
-
-    it 'uses provided reactive instance' do
-      reactive_instance = Datadog::AppSec::Reactive::Engine.new
-      described_class.new('test', nil, reactive_instance) do |op|
-        expect(op.reactive).to eq(reactive_instance)
-      end
-    end
-
-    it 'uses reactive instance from parent' do
-      parent_operation = described_class.new('parent_test')
-      described_class.new('test', parent_operation) do |op|
-        expect(op.reactive).to eq(parent_operation.reactive)
-      end
-    end
-
-    it 'uses reactive instance over parent engine' do
-      reactive_instance = Datadog::AppSec::Reactive::Engine.new
-      parent_operation = described_class.new('parent_test')
-      described_class.new('test', parent_operation, reactive_instance) do |op|
-        expect(op.reactive).to eq(reactive_instance)
-      end
-    end
-  end
-
-  describe '#subscribe' do
-    it 'delegates to reactive engine' do
-      operation = described_class.new('test')
-      expect(operation.reactive).to receive(:subscribe).with([:a, :b, :c])
-      operation.subscribe([:a, :b, :c]) do
-        1 + 1
-      end
-    end
-  end
-
-  describe '#publish' do
-    it 'delegates to reactive engine' do
-      operation = described_class.new('test')
-      expect(operation.reactive).to receive(:publish).with(:a, 'hello world')
-      operation.publish(:a, 'hello world')
-    end
-  end
-
-  describe '#finalize' do
-    it 'sets active to parent' do
-      parent_operation = described_class.new('parent_test')
-      described_class.new('test', parent_operation)
-      expect(described_class.active).to eq(parent_operation)
-      parent_operation.finalize
-      # The parent of parent_operation is nil because is the top operation
-      expect(described_class.active).to be_nil
-    end
-  end
-end
diff --git a/spec/datadog/appsec/reactive/shared_examples.rb b/spec/datadog/appsec/reactive/shared_examples.rb
index 84715c3b95f..352ae336781 100644
--- a/spec/datadog/appsec/reactive/shared_examples.rb
+++ b/spec/datadog/appsec/reactive/shared_examples.rb
@@ -3,98 +3,98 @@
 RSpec.shared_examples 'waf result' do
   context 'is a match' do
     it 'yields result and no blocking action' do
-      expect(operation).to receive(:subscribe).and_call_original
+      expect(engine).to receive(:subscribe).and_call_original
 
       waf_result = double(:waf_result, status: :match, timeout: false, actions: [])
       expect(waf_context).to receive(:run).and_return(waf_result)
-      described_class.subscribe(operation, waf_context) do |result|
+      described_class.subscribe(engine, waf_context) do |result|
         expect(result).to eq(waf_result)
       end
-      result = described_class.publish(operation, gateway)
+      result = described_class.publish(engine, gateway)
       expect(result).to be_nil
     end
 
     it 'yields result and blocking action. The publish method catches the resul as well' do
-      expect(operation).to receive(:subscribe).and_call_original
+      expect(engine).to receive(:subscribe).and_call_original
 
       waf_result = double(:waf_result, status: :match, timeout: false, actions: ['block'])
       expect(waf_context).to receive(:run).and_return(waf_result)
-      described_class.subscribe(operation, waf_context) do |result|
+      described_class.subscribe(engine, waf_context) do |result|
         expect(result).to eq(waf_result)
       end
-      block = described_class.publish(operation, gateway)
+      block = described_class.publish(engine, gateway)
       expect(block).to eq(true)
     end
   end
 
   context 'is ok' do
     it 'does not yield' do
-      expect(operation).to receive(:subscribe).and_call_original
+      expect(engine).to receive(:subscribe).and_call_original
 
       waf_result = double(:waf_result, status: :ok, timeout: false)
       expect(waf_context).to receive(:run).and_return(waf_result)
-      expect { |b| described_class.subscribe(operation, waf_context, &b) }.not_to yield_control
-      result = described_class.publish(operation, gateway)
+      expect { |b| described_class.subscribe(engine, waf_context, &b) }.not_to yield_control
+      result = described_class.publish(engine, gateway)
       expect(result).to be_nil
     end
   end
 
   context 'is invalid_call' do
     it 'does not yield' do
-      expect(operation).to receive(:subscribe).and_call_original
+      expect(engine).to receive(:subscribe).and_call_original
 
       waf_result = double(:waf_result, status: :invalid_call, timeout: false)
       expect(waf_context).to receive(:run).and_return(waf_result)
-      expect { |b| described_class.subscribe(operation, waf_context, &b) }.not_to yield_control
-      result = described_class.publish(operation, gateway)
+      expect { |b| described_class.subscribe(engine, waf_context, &b) }.not_to yield_control
+      result = described_class.publish(engine, gateway)
       expect(result).to be_nil
     end
   end
 
   context 'is invalid_rule' do
     it 'does not yield' do
-      expect(operation).to receive(:subscribe).and_call_original
+      expect(engine).to receive(:subscribe).and_call_original
 
       waf_result = double(:waf_result, status: :invalid_rule, timeout: false)
       expect(waf_context).to receive(:run).and_return(waf_result)
-      expect { |b| described_class.subscribe(operation, waf_context, &b) }.not_to yield_control
-      result = described_class.publish(operation, gateway)
+      expect { |b| described_class.subscribe(engine, waf_context, &b) }.not_to yield_control
+      result = described_class.publish(engine, gateway)
       expect(result).to be_nil
     end
   end
 
   context 'is invalid_flow' do
     it 'does not yield' do
-      expect(operation).to receive(:subscribe).and_call_original
+      expect(engine).to receive(:subscribe).and_call_original
 
       waf_result = double(:waf_result, status: :invalid_flow, timeout: false)
       expect(waf_context).to receive(:run).and_return(waf_result)
-      expect { |b| described_class.subscribe(operation, waf_context, &b) }.not_to yield_control
-      result = described_class.publish(operation, gateway)
+      expect { |b| described_class.subscribe(engine, waf_context, &b) }.not_to yield_control
+      result = described_class.publish(engine, gateway)
       expect(result).to be_nil
     end
   end
 
   context 'is no_rule' do
     it 'does not yield' do
-      expect(operation).to receive(:subscribe).and_call_original
+      expect(engine).to receive(:subscribe).and_call_original
 
       waf_result = double(:waf_result, status: :no_rule, timeout: false)
       expect(waf_context).to receive(:run).and_return(waf_result)
-      expect { |b| described_class.subscribe(operation, waf_context, &b) }.not_to yield_control
-      result = described_class.publish(operation, gateway)
+      expect { |b| described_class.subscribe(engine, waf_context, &b) }.not_to yield_control
+      result = described_class.publish(engine, gateway)
       expect(result).to be_nil
     end
   end
 
   context 'is unknown' do
     it 'does not yield' do
-      expect(operation).to receive(:subscribe).and_call_original
+      expect(engine).to receive(:subscribe).and_call_original
 
       waf_result = double(:waf_result, status: :foo, timeout: false)
       expect(waf_context).to receive(:run).and_return(waf_result)
-      expect { |b| described_class.subscribe(operation, waf_context, &b) }.not_to yield_control
-      result = described_class.publish(operation, gateway)
+      expect { |b| described_class.subscribe(engine, waf_context, &b) }.not_to yield_control
+      result = described_class.publish(engine, gateway)
       expect(result).to be_nil
     end
   end
diff --git a/spec/datadog/di/instrumenter_spec.rb b/spec/datadog/di/instrumenter_spec.rb
index 821e6f685df..695d8567386 100644
--- a/spec/datadog/di/instrumenter_spec.rb
+++ b/spec/datadog/di/instrumenter_spec.rb
@@ -983,6 +983,20 @@
         expect(observed_calls.length).to eq 1
         expect(observed_calls.first).to be_a(Hash)
       end
+
+      context 'when instrumenting a line in loaded but not tracked file' do
+        let(:probe) do
+          Datadog::DI::Probe.new(file: 'hook_line.rb', line_no: 3,
+            id: 1, type: :log)
+        end
+
+        it 'raises DITargetNotInRegistry' do
+          expect do
+            instrumenter.hook_line(probe) do |payload|
+            end
+          end.to raise_error(Datadog::DI::Error::DITargetNotInRegistry, /File matching probe path.*was loaded and is not in code tracker registry/)
+        end
+      end
     end
 
     context 'when method is recursive' do
diff --git a/spec/datadog/tracing/contrib/graphql/support/application.rb b/spec/datadog/tracing/contrib/graphql/support/application.rb
index 88ab68a45fc..ba403ada0d4 100644
--- a/spec/datadog/tracing/contrib/graphql/support/application.rb
+++ b/spec/datadog/tracing/contrib/graphql/support/application.rb
@@ -39,7 +39,6 @@
     TestGraphQL.send(:remove_const, :UserType) if defined?(TestGraphQL::UserType)
     load 'spec/datadog/tracing/contrib/graphql/support/application_schema.rb'
   end
-  let(:operation) { Datadog::AppSec::Reactive::Operation.new('test') }
   let(:schema) { TestGraphQL::Schema }
 end
 
diff --git a/spec/datadog/tracing/span_event_spec.rb b/spec/datadog/tracing/span_event_spec.rb
index 2d68d409379..0bd7ff9a63f 100644
--- a/spec/datadog/tracing/span_event_spec.rb
+++ b/spec/datadog/tracing/span_event_spec.rb
@@ -19,11 +19,40 @@
         expect(span_event.attributes).to eq({})
         expect(span_event.time_unix_nano / 1e9).to be_within(1).of(Time.now.to_f)
       end
+
+      context 'with invalid attributes' do
+        let(:attributes) do
+          {
+            'int' => 1,
+            'invalid_arr1' => [1, 'a'],
+            'invalid_arr2' => [[1]],
+            'invalid_int1' => 2 << 65,
+            'invalid_int2' => -2 << 65,
+            'invalid_float1' => Float::NAN,
+            'invalid_float2' => Float::INFINITY,
+            'string' => 'bar',
+          }
+        end
+
+        it 'skips invalid values' do
+          expect(Datadog.logger).to receive(:warn).with(/Attribute invalid_.*/).exactly(6).times
+
+          expect(span_event.attributes).to eq('int' => 1, 'string' => 'bar',)
+        end
+      end
+
+      context 'with attributes with non-string keys' do
+        let(:attributes) { { 1 => 'val1', sym: 'val2' } }
+
+        it 'converts keys to strings' do
+          expect(span_event.attributes).to eq('1' => 'val1', 'sym' => 'val2')
+        end
+      end
     end
 
     context 'given' do
       context ':attributes' do
-        let(:attributes) { { tag: 'value' } }
+        let(:attributes) { { 'tag' => 'value' } }
         it { is_expected.to have_attributes(attributes: attributes) }
       end
 
@@ -39,21 +68,62 @@
     let(:name) { 'Another Event!' }
 
     context 'with required fields' do
-      it { is_expected.to eq({ name: name, time_unix_nano: span_event.time_unix_nano }) }
+      it { is_expected.to eq({ 'name' => name, 'time_unix_nano' => span_event.time_unix_nano }) }
+    end
+
+    context 'with timestamp' do
+      let(:time_unix_nano) { 25 }
+      it { is_expected.to include('time_unix_nano' => 25) }
+    end
+
+    context 'when attributes is set' do
+      let(:attributes) { { 'event.name' => 'test_event', 'event.id' => 1, 'nested' => [2, 3] } }
+      it { is_expected.to include('attributes' => attributes) }
+    end
+  end
+
+  describe '#to_native_format' do
+    subject(:to_native_format) { span_event.to_native_format }
+    let(:name) { 'Another Event!' }
+
+    context 'with required fields' do
+      it { is_expected.to eq({ 'name' => name, 'time_unix_nano' => span_event.time_unix_nano }) }
     end
 
     context 'with timestamp' do
       let(:time_unix_nano) { 25 }
-      it { is_expected.to include(time_unix_nano: 25) }
+      it { is_expected.to include('time_unix_nano' => 25) }
     end
 
     context 'when attributes is set' do
-      let(:attributes) { { 'event.name' => 'test_event', 'event.id' => 1, 'nested' => [true, [2, 3], 'val'] } }
-      it {
-        is_expected.to include(
-          attributes: attributes
+      let(:attributes) do
+        {
+          'string' => 'value',
+          'bool' => true,
+          'int' => 1,
+          'float' => 1.0,
+          'string_arr' => %w[ab cd],
+          'bool_arr' => [true, false],
+          'int_arr' => [1, 2],
+          'float_arr' => [1.0, 2.0]
+        }
+      end
+
+      it do
+        expect(to_native_format['attributes']).to eq(
+          'string' => { type: 0, string_value: 'value' },
+          'bool' => { type: 1, bool_value: true },
+          'int' => { type: 2, int_value: 1 },
+          'float' => { type: 3, double_value: 1.0 },
+          'string_arr' => { type: 4,
+                            array_value: [{ type: 0, string_value: 'ab' }, { type: 0, string_value: 'cd' }] },
+          'bool_arr' => { type: 4,
+                          array_value: [{ type: 1, bool_value: true }, { type: 1, bool_value: false }] },
+          'int_arr' => { type: 4, array_value: [{ type: 2, int_value: 1 }, { type: 2, int_value: 2 }] },
+          'float_arr' => { type: 4,
+                           array_value: [{ type: 3, double_value: 1.0 }, { type: 3, double_value: 2.0 }] }
         )
-      }
+      end
     end
   end
 end
diff --git a/spec/datadog/tracing/transport/serializable_trace_spec.rb b/spec/datadog/tracing/transport/serializable_trace_spec.rb
index c1e63ae9ac5..de88a519225 100644
--- a/spec/datadog/tracing/transport/serializable_trace_spec.rb
+++ b/spec/datadog/tracing/transport/serializable_trace_spec.rb
@@ -8,9 +8,10 @@
 require 'datadog/tracing/transport/serializable_trace'
 
 RSpec.describe Datadog::Tracing::Transport::SerializableTrace do
-  subject(:serializable_trace) { described_class.new(trace) }
+  subject(:serializable_trace) { described_class.new(trace, native_events_supported) }
 
   let(:trace) { Datadog::Tracing::TraceSegment.new(spans) }
+  let(:native_events_supported) { false }
   let(:spans) do
     Array.new(3) do |i|
       span = Datadog::Tracing::Span.new(
@@ -26,12 +27,38 @@
     end
   end
 
-  describe '#to_msgpack' do
-    subject(:to_msgpack) { serializable_trace.to_msgpack }
+  shared_examples 'serialize all fields' do |include_duration: false, include_native_events: false|
+    it 'contains all fields' do
+      unpacked_trace.each do |span|
+        expected = [
+          'span_id',
+          'parent_id',
+          'trace_id',
+          'name',
+          'service',
+          'resource',
+          'type',
+          'meta',
+          'metrics',
+          'span_links',
+          'error',
+        ]
+        if include_duration
+          expected << 'start'
+          expected << 'duration'
+        end
+        expected << 'span_events' if include_native_events
 
-    context 'when packed then upacked' do
-      subject(:unpacked_trace) { MessagePack.unpack(to_msgpack) }
+        expect(span.keys).to match_array(expected)
+      end
+    end
+  end
 
+  describe '#to_msgpack' do
+    subject(:unpacked_trace) { MessagePack.unpack(to_msgpack) }
+    let(:to_msgpack) { serializable_trace.to_msgpack }
+
+    context 'when packed then unpacked' do
       let(:original_spans) do
         spans.map do |span|
           Hash[span.to_hash.map { |k, v| [k.to_s, v] }]
@@ -44,8 +71,6 @@
     end
 
     context 'when given trace_id' do
-      subject(:unpacked_trace) { MessagePack.unpack(to_msgpack) }
-
       let(:spans) do
         Array.new(3) do |_i|
           Datadog::Tracing::Span.new(
@@ -73,8 +98,6 @@
     end
 
     context 'when given span links' do
-      subject(:unpacked_trace) { MessagePack.unpack(to_msgpack) }
-
       let(:spans) do
         Array.new(3) do |_i|
           Datadog::Tracing::Span.new(
@@ -131,8 +154,6 @@
     end
 
     context 'when given span events' do
-      subject(:unpacked_trace) { MessagePack.unpack(to_msgpack) }
-
       let(:spans) do
         Array.new(2) do |i|
           Datadog::Tracing::Span.new(
@@ -140,11 +161,11 @@
             events: [
               Datadog::Tracing::SpanEvent.new(
                 'First Event',
-                time_unix_nano: 1_000_000_000
+                time_unix_nano: 123
               ),
               Datadog::Tracing::SpanEvent.new(
                 "Another Event #{i}!",
-                time_unix_nano: 2_000_000_000,
+                time_unix_nano: 456,
                 attributes: { id: i, required: (i == 1) },
               ),
             ],
@@ -159,14 +180,47 @@
           end
         ).to eq(
           [
-            '[{"name":"First Event","time_unix_nano":1000000000},{"name":"Another Event 0!","time_unix_nano":2000000000,' \
+            '[{"name":"First Event","time_unix_nano":123},{"name":"Another Event 0!","time_unix_nano":456,' \
             '"attributes":{"id":0,"required":false}}]',
-            '[{"name":"First Event","time_unix_nano":1000000000},{"name":"Another Event 1!","time_unix_nano":2000000000,' \
+            '[{"name":"First Event","time_unix_nano":123},{"name":"Another Event 1!","time_unix_nano":456,' \
             '"attributes":{"id":1,"required":true}}]',
           ]
         )
       end
+
+      it_behaves_like 'serialize all fields'
+
+      context 'when native events are supported' do
+        let(:native_events_supported) { true }
+
+        it 'serializes span events as top-level field' do
+          expect(
+            unpacked_trace.map do |s|
+              s['span_events']
+            end
+          ).to eq(
+            [
+              [
+                { 'name' => 'First Event', 'time_unix_nano' => 123 },
+                { 'name' => 'Another Event 0!', 'time_unix_nano' => 456, 'attributes' => {
+                  'id' => { 'int_value' => 0, 'type' => 2 }, 'required' => { 'bool_value' => false, 'type' => 1 }
+                } }
+              ],
+              [
+                { 'name' => 'First Event', 'time_unix_nano' => 123 },
+                { 'name' => 'Another Event 1!', 'time_unix_nano' => 456, 'attributes' => {
+                  'id' => { 'int_value' => 1, 'type' => 2 }, 'required' => { 'bool_value' => true, 'type' => 1 }
+                } }
+              ],
+            ]
+          )
+        end
+
+        it_behaves_like 'serialize all fields', include_native_events: true
+      end
     end
+
+    it_behaves_like 'serialize all fields'
   end
 
   describe '#to_json' do