diff --git a/lib/datadog/di/probe_notification_builder.rb b/lib/datadog/di/probe_notification_builder.rb
new file mode 100644
index 00000000000..57baa1bccc3
--- /dev/null
+++ b/lib/datadog/di/probe_notification_builder.rb
@@ -0,0 +1,216 @@
+# frozen_string_literal: true
+
+module Datadog
+  module DI
+    # Builds probe status notification and snapshot payloads.
+    #
+    # @api private
+    class ProbeNotificationBuilder
+      def initialize(settings, serializer)
+        @settings = settings
+        @serializer = serializer
+      end
+
+      attr_reader :settings
+      attr_reader :serializer
+
+      def build_received(probe)
+        build_status(probe,
+          message: "Probe #{probe.id} has been received correctly",
+          status: 'RECEIVED',)
+      end
+
+      def build_installed(probe)
+        build_status(probe,
+          message: "Probe #{probe.id} has been instrumented correctly",
+          status: 'INSTALLED',)
+      end
+
+      def build_emitting(probe)
+        build_status(probe,
+          message: "Probe #{probe.id} is emitting",
+          status: 'EMITTING',)
+      end
+
+      def build_executed(probe,
+        trace_point: nil, rv: nil, duration: nil, callers: nil,
+        args: nil, kwargs: nil, serialized_entry_args: nil)
+        snapshot = if probe.line? && probe.capture_snapshot?
+          if trace_point.nil?
+            raise "Cannot create snapshot because there is no trace point"
+          end
+          get_local_variables(trace_point)
+        end
+        if callers
+          callers = callers[0..9]
+        end
+        build_snapshot(probe, rv: rv, snapshot: snapshot,
+          duration: duration, callers: callers, args: args, kwargs: kwargs,
+          serialized_entry_args: serialized_entry_args)
+      end
+
+      def build_snapshot(probe, rv: nil, snapshot: nil,
+        duration: nil, callers: nil, args: nil, kwargs: nil,
+        serialized_entry_args: nil)
+        # TODO also verify that non-capturing probe does not pass
+        # snapshot or vars/args into this method
+        captures = if probe.capture_snapshot?
+          if probe.method?
+            {
+              entry: {
+                # standard:disable all
+                arguments: if serialized_entry_args
+                  serialized_entry_args
+                else
+                  (args || kwargs) && serializer.serialize_args(args, kwargs)
+                end,
+                throwable: nil,
+                # standard:enable all
+              },
+              return: {
+                arguments: {
+                  "@return": serializer.serialize_value(rv),
+                },
+                throwable: nil,
+              },
+            }
+          elsif probe.line?
+            {
+              lines: snapshot && {
+                probe.line_no => {locals: serializer.serialize_vars(snapshot)},
+              },
+            }
+          end
+        end
+
+        location = if probe.line?
+          actual_file = if probe.file
+            # Normally callers should always be filled for a line probe
+            # but in the test suite we don't always provide all arguments.
+            callers&.detect do |caller|
+              File.basename(caller.sub(/:.*/, '')) == File.basename(probe.file)
+            end&.sub(/:.*/, '') || probe.file
+          end
+          {
+            file: actual_file,
+            lines: [probe.line_no],
+          }
+        elsif probe.method?
+          {
+            method: probe.method_name,
+            type: probe.type_name,
+          }
+        end
+
+        stack = if callers
+          format_callers(callers)
+        end
+
+        timestamp = timestamp_now
+        {
+          service: settings.service,
+          "debugger.snapshot": {
+            id: SecureRandom.uuid,
+            timestamp: timestamp,
+            evaluationErrors: [],
+            probe: {
+              id: probe.id,
+              version: 0,
+              location: location,
+            },
+            language: 'ruby',
+            # TODO add test coverage for callers being nil
+            stack: stack,
+            captures: captures,
+          },
+          # In python tracer duration is under debugger.snapshot,
+          # but UI appears to expect it here at top level.
+          duration: duration ? (duration * 10**9).to_i : nil,
+          host: nil,
+          logger: {
+            name: probe.file,
+            method: probe.method_name || 'no_method',
+            thread_name: Thread.current.name,
+            thread_id: thread_id,
+            version: 2,
+          },
+          "dd.trace_id": 136035165280417366521542318182735500431,
+          "dd.span_id": 17576285113343575026,
+          ddsource: 'dd_debugger',
+          message: probe.template && evaluate_template(probe.template,
+            duration: duration ? duration * 1000 : nil),
+          timestamp: timestamp,
+        }
+      end
+
+      def build_status(probe, message:, status:)
+        {
+          service: settings.service,
+          timestamp: timestamp_now,
+          message: message,
+          ddsource: 'dd_debugger',
+          debugger: {
+            diagnostics: {
+              probeId: probe.id,
+              probeVersion: 0,
+              runtimeId: Core::Environment::Identity.id,
+              parentId: nil,
+              status: status,
+            },
+          },
+        }
+      end
+
+      def format_callers(callers)
+        callers.map do |caller|
+          if caller =~ /\A([^:]+):(\d+):in `([^']+)'\z/
+            {
+              fileName: $1, function: $3, lineNumber: Integer($2),
+            }
+          else
+            {
+              fileName: 'unknown', function: 'unknown', lineNumber: 0,
+            }
+          end
+        end
+      end
+
+      def evaluate_template(template, **vars)
+        message = template.dup
+        vars.each do |key, value|
+          message.gsub!("{@#{key}}", value.to_s)
+        end
+        message
+      end
+
+      def timestamp_now
+        (Time.now.to_f * 1000).to_i
+      end
+
+      def get_local_variables(trace_point)
+        # binding appears to be constructed on access, therefore
+        # 1) we should attempt to cache it and
+        # 2) we should not call +binding+ until we actually need variable values.
+        binding = trace_point.binding
+
+        # steep hack - should never happen
+        return {} unless binding
+
+        binding.local_variables.each_with_object({}) do |name, map|
+          value = binding.local_variable_get(name)
+          map[name] = value
+        end
+      end
+
+      def thread_id
+        thread = Thread.current
+        if thread.respond_to?(:native_thread_id)
+          # Ruby 3.1+
+          thread.native_thread_id
+        else
+          thread.object_id
+        end
+      end
+    end
+  end
+end
diff --git a/sig/datadog/di/probe_notification_builder.rbs b/sig/datadog/di/probe_notification_builder.rbs
new file mode 100644
index 00000000000..094cf4db1c8
--- /dev/null
+++ b/sig/datadog/di/probe_notification_builder.rbs
@@ -0,0 +1,36 @@
+module Datadog
+  module DI
+    class ProbeNotificationBuilder
+      @serializer: Serializer
+
+      def initialize: (untyped settings, Serializer serializer) -> void
+
+      attr_reader settings: untyped
+      attr_reader serializer: Serializer
+
+      def build_received: (Probe probe) -> untyped
+
+      def build_installed: (Probe probe) -> untyped
+
+      def build_emitting: (Probe probe) -> untyped
+
+      def build_executed: (Probe probe, ?trace_point: untyped?, ?rv: untyped?, ?duration: untyped?, ?callers: untyped?, ?args: untyped?, ?kwargs: untyped?, ?serialized_entry_args: untyped?) -> untyped
+
+      def build_snapshot: (Probe probe, ?rv: untyped?, ?snapshot: untyped?, ?duration: untyped?, ?callers: untyped?, ?args: untyped?, ?kwargs: untyped?, ?serialized_entry_args: untyped?) -> { service: untyped, :"debugger.snapshot" => { id: untyped, timestamp: untyped, evaluationErrors: ::Array[untyped], probe: { id: untyped, version: 0, location: untyped }, language: "ruby", stack: untyped, captures: untyped }, duration: untyped, host: nil, logger: { name: untyped, method: untyped, thread_name: untyped, thread_id: untyped, version: 2 }, :"dd.trace_id" => 136035165280417366521542318182735500431, :"dd.span_id" => 17576285113343575026, ddsource: "dd_debugger", message: untyped, timestamp: untyped }
+
+      def build_status: (Probe probe, message: untyped, status: untyped) -> (nil | { service: untyped, timestamp: untyped, message: untyped, ddsource: "dd_debugger", debugger: { diagnostics: { probeId: untyped, probeVersion: 0, runtimeId: untyped, parentId: nil, status: untyped } } })
+
+      def format_callers: (untyped callers) -> untyped
+
+      def evaluate_template: (untyped template, **untyped vars) -> untyped
+
+      def timestamp_now: () -> untyped
+
+      def get_local_variables: (TracePoint trace_point) -> untyped
+      
+      private
+      
+      def thread_id: -> Integer
+    end
+  end
+end
diff --git a/sig/datadog/di/serializer.rbs b/sig/datadog/di/serializer.rbs
index aaaf674148e..eb428098bc1 100644
--- a/sig/datadog/di/serializer.rbs
+++ b/sig/datadog/di/serializer.rbs
@@ -13,9 +13,9 @@ module Datadog
       
       def serialize_args: (untyped args, untyped kwargs) -> untyped
       def serialize_vars: (untyped vars) -> untyped
+      def serialize_value: (untyped value, ?name: String, ?depth: Integer) -> untyped
 
       private
-      def serialize_value: (untyped value, ?name: String, ?depth: untyped) -> ({ type: untyped, notCapturedReason: "redactedType" } | { type: untyped, notCapturedReason: "redactedIdent" } | untyped)
       def class_name: (untyped cls) -> untyped
     end
   end
diff --git a/spec/datadog/di/integration/probe_notification_builder_spec.rb b/spec/datadog/di/integration/probe_notification_builder_spec.rb
new file mode 100644
index 00000000000..4a052b95564
--- /dev/null
+++ b/spec/datadog/di/integration/probe_notification_builder_spec.rb
@@ -0,0 +1,108 @@
+require 'datadog/di/serializer'
+require 'datadog/di/probe'
+require 'datadog/di/probe_notification_builder'
+
+RSpec.describe Datadog::DI::ProbeNotificationBuilder do
+  describe 'log probe' do
+    let(:redactor) { Datadog::DI::Redactor.new(settings) }
+    let(:serializer) { Datadog::DI::Serializer.new(settings, redactor) }
+
+    let(:builder) { described_class.new(settings, serializer) }
+
+    let(:instrumenter) do
+      Datadog::DI::Instrumenter.new(settings)
+    end
+
+    let(:settings) do
+      double('settings').tap do |settings|
+        allow(settings).to receive(:dynamic_instrumentation).and_return(di_settings)
+        allow(settings).to receive(:service).and_return('fake service')
+      end
+    end
+
+    let(:di_settings) do
+      double('di settings').tap do |settings|
+        allow(settings).to receive(:enabled).and_return(true)
+        allow(settings).to receive(:untargeted_trace_points).and_return(false)
+        allow(settings).to receive(:max_capture_depth).and_return(2)
+        allow(settings).to receive(:max_capture_string_length).and_return(20)
+        allow(settings).to receive(:max_capture_collection_size).and_return(20)
+        allow(settings).to receive(:redacted_type_names).and_return([])
+        allow(settings).to receive(:redacted_identifiers).and_return([])
+      end
+    end
+
+    context 'line probe' do
+      let(:probe) do
+        Datadog::DI::Probe.new(id: '123', type: :log, file: 'X', line_no: 1, capture_snapshot: true)
+      end
+
+      context 'with snapshot' do
+        let(:vars) do
+          {hello: 42, hash: {hello: 42, password: 'redacted'}, array: [true]}
+        end
+
+        let(:captures) do
+          {lines: {1 => {
+            locals: {
+              hello: {type: 'Integer', value: '42'},
+              hash: {type: 'Hash', entries: [
+                [{type: 'Symbol', value: 'hello'}, {type: 'Integer', value: '42'}],
+                [{type: 'Symbol', value: 'password'}, {type: 'String', notCapturedReason: 'redactedIdent'}],
+              ]},
+              array: {type: 'Array', elements: [
+                {type: 'TrueClass', value: 'true'},
+              ]},
+            },
+          }}}
+        end
+
+        it 'builds expected payload' do
+          payload = builder.build_snapshot(probe, snapshot: vars)
+          expect(payload).to be_a(Hash)
+          expect(payload.fetch(:"debugger.snapshot").fetch(:captures)).to eq(captures)
+        end
+      end
+    end
+
+    context 'method probe' do
+      let(:probe) do
+        Datadog::DI::Probe.new(id: '123', type: :log, type_name: 'X', method_name: 'y', capture_snapshot: true)
+      end
+
+      context 'with snapshot' do
+        let(:args) do
+          [1, 'hello']
+        end
+
+        let(:kwargs) do
+          {foo: 42}
+        end
+
+        let(:expected_captures) do
+          {entry: {
+            arguments: {
+              arg1: {type: 'Integer', value: '1'},
+              arg2: {type: 'String', value: 'hello'},
+              foo: {type: 'Integer', value: '42'},
+            }, throwable: nil,
+          }, return: {
+            arguments: {
+              :@return => {
+                type: 'NilClass',
+                isNull: true,
+              },
+            }, throwable: nil,
+          }}
+        end
+
+        it 'builds expected payload' do
+          payload = builder.build_snapshot(probe, args: args, kwargs: kwargs)
+          expect(payload).to be_a(Hash)
+          captures = payload.fetch(:"debugger.snapshot").fetch(:captures)
+          expect(captures).to eq(expected_captures)
+        end
+      end
+    end
+  end
+end
diff --git a/spec/datadog/di/probe_notification_builder_spec.rb b/spec/datadog/di/probe_notification_builder_spec.rb
new file mode 100644
index 00000000000..8a5ef1be30d
--- /dev/null
+++ b/spec/datadog/di/probe_notification_builder_spec.rb
@@ -0,0 +1,101 @@
+require 'datadog/di/probe_notification_builder'
+require 'datadog/di/serializer'
+require 'datadog/di/probe'
+
+# Notification builder is primarily tested via integration tests for
+# dynamic instrumentation overall, since the generated payloads depend
+# heavily on probe attributes and parameters.
+#
+# The unit tests here are only meant to catch grave errors in the implementaton,
+# not comprehensively verify correctness.
+
+RSpec.describe Datadog::DI::ProbeNotificationBuilder do
+  let(:settings) do
+    double("settings").tap do |settings|
+      allow(settings).to receive(:dynamic_instrumentation).and_return(di_settings)
+      allow(settings).to receive(:service).and_return('test service')
+    end
+  end
+
+  let(:di_settings) do
+    double("di settings").tap do |settings|
+      allow(settings).to receive(:enabled).and_return(true)
+      allow(settings).to receive(:propagate_all_exceptions).and_return(false)
+      allow(settings).to receive(:redacted_identifiers).and_return([])
+      allow(settings).to receive(:redacted_type_names).and_return(%w[])
+      allow(settings).to receive(:max_capture_collection_size).and_return(10)
+      allow(settings).to receive(:max_capture_attribute_count).and_return(10)
+      allow(settings).to receive(:max_capture_depth).and_return(2)
+      allow(settings).to receive(:max_capture_string_length).and_return(100)
+    end
+  end
+
+  let(:redactor) { Datadog::DI::Redactor.new(settings) }
+  let(:serializer) { Datadog::DI::Serializer.new(settings, redactor) }
+
+  let(:builder) { described_class.new(settings, serializer) }
+
+  let(:probe) do
+    Datadog::DI::Probe.new(id: '123', type: :log, file: 'X', line_no: 1)
+  end
+
+  describe '#build_received' do
+    it 'returns a hash' do
+      expect(builder.build_received(probe)).to be_a(Hash)
+    end
+  end
+
+  describe '#build_installed' do
+    it 'returns a hash' do
+      expect(builder.build_installed(probe)).to be_a(Hash)
+    end
+  end
+
+  describe '#build_emitting' do
+    it 'returns a hash' do
+      expect(builder.build_emitting(probe)).to be_a(Hash)
+    end
+  end
+
+  describe '#build_executed' do
+    context 'with template' do
+      let(:probe) do
+        Datadog::DI::Probe.new(id: '123', type: :log, file: 'X', line_no: 1,
+          template: 'hello world')
+      end
+
+      it 'returns a hash' do
+        expect(builder.build_executed(probe)).to be_a(Hash)
+      end
+    end
+
+    context 'without snapshot capture' do
+      let(:probe) do
+        Datadog::DI::Probe.new(id: '123', type: :log, file: 'X', line_no: 1,
+          capture_snapshot: false)
+      end
+
+      it 'returns a hash' do
+        expect(builder.build_executed(probe)).to be_a(Hash)
+      end
+    end
+
+    context 'with snapshot capture' do
+      let(:probe) do
+        Datadog::DI::Probe.new(id: '123', type: :log, file: 'X', line_no: 1,
+          capture_snapshot: true,)
+      end
+
+      let(:trace_point) do
+        instance_double(TracePoint).tap do |tp|
+          # Returns an empty binding
+          expect(tp).to receive(:binding).and_return(binding)
+        end
+      end
+
+      it 'returns a hash' do
+        expect(builder.build_executed(probe, trace_point: trace_point)).to be_a(Hash)
+      end
+    end
+  end
+end