diff --git a/Gemfile b/Gemfile
index 17463e74e..e7ee0d08a 100644
--- a/Gemfile
+++ b/Gemfile
@@ -32,7 +32,6 @@ group :test do
 end
 
 group :spec do
-  # Using feature branch until https://github.com/Shopify/liquid-spec/pull/144 is merged
-  gem 'liquid-spec', github: 'Shopify/liquid-spec', branch: 'self-drop-env-lookup-specs'
+  gem 'liquid-spec', github: 'Shopify/liquid-spec'
   gem 'activesupport', require: false
 end
diff --git a/docs/template_recorder.md b/docs/template_recorder.md
new file mode 100644
index 000000000..451a3ffed
--- /dev/null
+++ b/docs/template_recorder.md
@@ -0,0 +1,72 @@
+# Recording and replaying renders
+
+`Liquid::TemplateRecorder` captures successful template renders so they can be
+replayed without the application's file system or Drop implementations.
+Recording does not wrap or replace assigns, so the recorded render has the same
+semantics as a normal render.
+
+```ruby
+Liquid::TemplateRecorder.record("render.json") do
+  template = Liquid::Template.parse(source)
+  template.render!(assigns)
+end
+
+replayer = Liquid::TemplateRecorder.replay_from("render.json", mode: :verify)
+replayer.render # raises if the output changed
+```
+
+A recording contains the root template, every parsed partial, partial contents,
+plain Hash/Array values resolved by the template, properties actually read from `Liquid::Drop` objects,
+filter-call diagnostics, engine options, and the rendered output. Drop instance
+variables are never inspected. An unsupported Ruby object raises
+`Liquid::TemplateRecorder::SerializationError` rather than silently producing a
+recording that cannot be replayed.
+
+## Storage formats
+
+A `.json` destination is written atomically after the recording block succeeds.
+It contains a session with every render performed by the block.
+
+A `.jsonl` destination is append-only. Each successful top-level render is one
+compact, self-contained JSON line. This is the recommended format for production
+sampling: a process failure can lose at most the render being written, writers
+are serialized with `flock`, and a recording can be replayed by index.
+
+A destination may instead be any writer object responding to `write(record)`. The
+writer receives one self-contained recording Hash per successful render. Liquid
+does not own or close injected writers, so applications can publish records to
+Kafka, object storage, or another transport without coupling that transport to
+the recorder.
+Pass `on_error:` to keep serialization or sink failures out of the render path;
+the callback receives the error and should not raise.
+
+```ruby
+Liquid::TemplateRecorder.record(kafka_writer) do
+  template.render!(assigns)
+end
+```
+
+```ruby
+Liquid::TemplateRecorder.replay_from("renders.jsonl")           # last render
+Liquid::TemplateRecorder.replay_from("renders.jsonl", index: 0) # first render
+Liquid::TemplateRecorder.records("renders.jsonl")               # inspect all
+```
+
+Compression is intentionally separate from the schema. In particular, one
+long-lived compressed stream makes appending, recovery, and selecting a render
+harder. Compress rotated `.jsonl` files with the storage system of your choice;
+a future compressed writer can use one independent frame per record without a
+schema change.
+
+Recording sessions are thread-local. Nested sessions in the same thread are
+rejected. Existing application register names and the one-argument
+`FileSystem#read_template_file` API remain unchanged.
+
+## Replay modes
+
+* `:compute` runs filters normally. Pass application filters with
+  `replayer.render(filters: MyFilters)`.
+* `:strict` returns each exact recorded filter result and rejects a changed
+  filter sequence. This can replay application-specific or nondeterministic
+  filters without loading their implementations.
+* `:verify` computes normally and raises when the final output differs.
diff --git a/lib/liquid.rb b/lib/liquid.rb
index dce089772..b6034c233 100644
--- a/lib/liquid.rb
+++ b/lib/liquid.rb
@@ -90,3 +90,4 @@ module Liquid
 require 'liquid/usage'
 require 'liquid/registers'
 require 'liquid/template_factory'
+require "liquid/template_recorder"
diff --git a/lib/liquid/block_body.rb b/lib/liquid/block_body.rb
index 90e7c8237..9c6aef518 100644
--- a/lib/liquid/block_body.rb
+++ b/lib/liquid/block_body.rb
@@ -84,10 +84,15 @@ def self.raise_missing_variable_terminator(token, parse_context)
 
     # @api private
     def self.render_node(context, output, node)
+      recorder = TemplateRecorder.current
+      tag_call = recorder&.begin_tag_render(node, context)
+      output_start = output.length
       node.render_to_output_buffer(context, output)
     rescue => exc
       blank_tag = !node.instance_of?(Variable) && node.blank?
       rescue_render_node(context, output, node.line_number, exc, blank_tag)
+    ensure
+      recorder&.finish_tag_render(tag_call, output[output_start..]) if output_start
     end
 
     # @api private
diff --git a/lib/liquid/context.rb b/lib/liquid/context.rb
index 30492578e..ce398f0ef 100644
--- a/lib/liquid/context.rb
+++ b/lib/liquid/context.rb
@@ -228,6 +228,8 @@ def find_variable(key, raise_on_not_found: true)
 
       liquid_variable.context = self if variable != liquid_variable && liquid_variable.respond_to?(:context=)
 
+      recorder = @registers[TemplateRecorder::REGISTER_KEY] if defined?(TemplateRecorder)
+      recorder&.emit_variable_read(key, liquid_variable)
       liquid_variable
     end
 
diff --git a/lib/liquid/drop.rb b/lib/liquid/drop.rb
index d13398ec4..ebd1f05ca 100644
--- a/lib/liquid/drop.rb
+++ b/lib/liquid/drop.rb
@@ -37,11 +37,15 @@ def liquid_method_missing(method)
 
     # called by liquid to invoke a drop
     def invoke_drop(method_or_key)
-      if self.class.invokable?(method_or_key)
+      result = if self.class.invokable?(method_or_key)
         send(method_or_key)
       else
         liquid_method_missing(method_or_key)
       end
+
+      recorder = @context&.registers&.[](TemplateRecorder::REGISTER_KEY) if defined?(TemplateRecorder)
+      recorder&.emit_drop_read(self, method_or_key, result)
+      result
     end
 
     def key?(_name)
diff --git a/lib/liquid/partial_cache.rb b/lib/liquid/partial_cache.rb
index f49d14d90..e99a6cb32 100644
--- a/lib/liquid/partial_cache.rb
+++ b/lib/liquid/partial_cache.rb
@@ -10,6 +10,8 @@ def self.load(template_name, context:, parse_context:)
 
       file_system = context.registers[:file_system]
       source      = file_system.read_template_file(template_name)
+      recorder = context.registers[TemplateRecorder::REGISTER_KEY] if defined?(TemplateRecorder)
+      recorder&.emit_file_read(template_name, source)
 
       parse_context.partial = true
 
diff --git a/lib/liquid/strainer_template.rb b/lib/liquid/strainer_template.rb
index ca0626dda..4e6f662e1 100644
--- a/lib/liquid/strainer_template.rb
+++ b/lib/liquid/strainer_template.rb
@@ -48,13 +48,17 @@ def filter_methods
     end
 
     def invoke(method, *args)
-      if self.class.invokable?(method)
+      result = if self.class.invokable?(method)
         send(method, *args)
       elsif @context.strict_filters
         raise Liquid::UndefinedFilter, "undefined filter #{method}"
       else
         args.first
       end
+
+      recorder = @context&.registers&.[](TemplateRecorder::REGISTER_KEY) if defined?(TemplateRecorder)
+      recorder&.emit_filter_call(method, args.first, args.drop(1), result)
+      result
     rescue ::ArgumentError => e
       raise Liquid::ArgumentError, e.message, e.backtrace
     end
diff --git a/lib/liquid/template.rb b/lib/liquid/template.rb
index 2190e928d..17352b268 100644
--- a/lib/liquid/template.rb
+++ b/lib/liquid/template.rb
@@ -105,6 +105,7 @@ def parse(source, options = {})
 
       tokenizer     = parse_context.new_tokenizer(source, start_line_number: @line_numbers && 1)
       @root         = Document.parse(tokenizer, parse_context)
+      @template_recorder_source = source.dup.freeze if defined?(TemplateRecorder) && TemplateRecorder.current
       self
     end
 
@@ -141,6 +142,8 @@ def errors
     def render(*args)
       return '' if @root.nil?
 
+      recording_session = TemplateRecorder.current if defined?(TemplateRecorder)
+      recording_assigns = args.first
       context = case args.first
       when Liquid::Context
         c = args.shift
@@ -180,6 +183,13 @@ def render(*args)
         context.add_filters(args.pop)
       end
 
+      recording = recording_session&.begin_render(self, recording_assigns, context)
+      if recording
+        recorder_registers = context.registers.static
+        previous_recorder = recorder_registers[TemplateRecorder::REGISTER_KEY]
+        recorder_registers[TemplateRecorder::REGISTER_KEY] = recording
+      end
+
       # Retrying a render resets resource usage
       context.resource_limits.reset
 
@@ -192,17 +202,31 @@ def render(*args)
       previous_error_mode = context.registers.static[:template_error_mode]
       context.registers.static[:template_error_mode] = @error_mode
 
+      rendered_output = nil
+      render_succeeded = false
       begin
         # render the nodelist.
-        @root.render_to_output_buffer(context, output || +'')
+        rendered_output = @root.render_to_output_buffer(context, output || +'')
+        render_succeeded = true
+        rendered_output
       rescue Liquid::MemoryError => e
-        context.handle_error(e)
+        rendered_output = context.handle_error(e)
+        render_succeeded = true
+        rendered_output
       ensure
         if previous_error_mode
           context.registers.static[:template_error_mode] = previous_error_mode
         else
           context.registers.static.delete(:template_error_mode)
         end
+        if recording
+          if previous_recorder
+            recorder_registers[TemplateRecorder::REGISTER_KEY] = previous_recorder
+          else
+            recorder_registers.delete(TemplateRecorder::REGISTER_KEY)
+          end
+          recording_session.finish_render(recording, rendered_output, context, success: render_succeeded)
+        end
         @errors = context.errors
       end
     end
diff --git a/lib/liquid/template_recorder.rb b/lib/liquid/template_recorder.rb
new file mode 100644
index 000000000..73c0e49cb
--- /dev/null
+++ b/lib/liquid/template_recorder.rb
@@ -0,0 +1,550 @@
+# frozen_string_literal: true
+
+require 'English'
+require "digest/sha2"
+require "json"
+require "securerandom"
+require "tempfile"
+require "time"
+
+module Liquid
+  # Records complete Liquid renders without changing the objects being rendered.
+  # A .json file contains one session; a .jsonl file is an append-only sequence
+  # of independently replayable renders.
+  class TemplateRecorder
+    FORMAT = "liquid-render"
+    SCHEMA_VERSION = 1
+    REGISTER_KEY = :__liquid_template_recorder
+    REPLAYER_REGISTER_KEY = :__liquid_template_recorder_replayer
+
+    class Error < StandardError; end
+    class ReplayError < Error; end
+    class SerializationError < Error; end
+
+    class << self
+      def record(destination, on_error: nil)
+        raise ArgumentError, "a block is required" unless block_given?
+
+        previous_session = current
+        raise Error, "nested recording sessions are not supported" if previous_session
+
+        session = Session.new(destination, on_error: on_error)
+        Thread.current[thread_key] = session
+        yield
+      ensure
+        if session
+          Thread.current[thread_key] = previous_session
+          session.close if $ERROR_INFO.nil?
+        end
+      end
+
+      def current
+        Thread.current[thread_key]
+      end
+
+      def replay_from(path, mode: :compute, index: -1)
+        records = Store.read(path)
+        raise ReplayError, "recording contains no renders" if records.empty?
+
+        record = records.fetch(index)
+        Replayer.new(record, mode: mode)
+      rescue IndexError
+        raise ReplayError, "render index #{index} does not exist"
+      end
+
+      def records(path)
+        Store.read(path)
+      end
+
+      private
+
+      def thread_key
+        :__liquid_template_recorder_session
+      end
+    end
+
+    class Session
+      def initialize(destination, on_error:)
+        @path = destination.to_s if destination.is_a?(String) || destination.respond_to?(:to_path)
+        @writer = destination unless @path
+        @on_error = on_error
+        @records = []
+        @active = nil
+        @pending_files = {}
+      end
+
+      def begin_render(template, _assigns, context)
+        if @active
+          @active.add_template(template)
+          @active.nesting += 1
+          return @active
+        end
+
+        @active = Render.new(template, context)
+        @pending_files.each { |path, source| @active.emit_file_read(path, source) }
+        @pending_files.clear
+        @active
+      end
+
+      def emit_file_read(path, source)
+        if @active
+          @active.emit_file_read(path, source)
+        else
+          @pending_files[path.to_s] = source
+        end
+      end
+
+      def emit_variable_output(output)
+        @active&.emit_variable_output(output)
+      end
+
+      def begin_tag_render(node, context)
+        @active&.begin_tag_render(node, context)
+      end
+
+      def finish_tag_render(call, output)
+        @active&.finish_tag_render(call, output)
+      end
+
+      def finish_render(render, output, context, success:)
+        return unless render.equal?(@active)
+
+        if render.nesting.positive?
+          render.nesting -= 1
+          return
+        end
+
+        if success
+          begin
+            record = render.finish(output, context)
+            if @writer
+              @writer.write(record)
+            elsif Store.jsonl?(@path)
+              Store.append(@path, record)
+            else
+              @records << record
+            end
+          rescue => error
+            handle_error(error)
+          end
+        end
+        @active = nil
+      end
+
+      def close
+        return if @writer || Store.jsonl?(@path)
+
+        Store.write_session(@path, @records)
+      rescue => error
+        handle_error(error)
+      end
+
+      private
+
+      def handle_error(error)
+        raise error unless @on_error
+
+        @on_error.call(error)
+      end
+    end
+
+    class Render
+      attr_accessor :nesting
+
+      def initialize(template, _context)
+        @nesting = 0
+        @templates = []
+        @files = {}
+        @filter_calls = []
+        @tag_calls = []
+        @tag_render_depth = 0
+        @variables = {}
+        @variable_outputs = []
+        @drop_values = {}
+        @bindings = {}.compare_by_identity
+        @root_template = template
+        add_template(template)
+      end
+
+      def add_template(template)
+        source = template.instance_variable_get(:@template_recorder_source)
+        return unless source
+
+        entrypoint = template.name
+        digest = Digest::SHA256.hexdigest(source)
+        return if @templates.any? { |item| item["sha256"] == digest && item["entrypoint"] == entrypoint }
+
+        @templates << { "source" => source, "entrypoint" => entrypoint, "sha256" => digest }
+      end
+
+      def emit_variable_output(output)
+        return if @tag_render_depth.positive?
+
+        @variable_outputs << output
+      end
+
+      def begin_tag_render(node, context)
+        name = context.environment.tags.key(node.class)
+        return unless name
+
+        @tag_render_depth += 1
+        return :nested if @tag_render_depth > 1
+
+        call = { "name" => name.to_s, "output" => nil }
+        @tag_calls << call
+        call
+      end
+
+      def finish_tag_render(call, output)
+        return unless call
+
+        @tag_render_depth -= 1
+        call["output"] = output unless call == :nested
+      end
+
+      def emit_variable_read(name, value)
+        path = [name.to_s]
+        @variables[name.to_s] = serialize(value, path, bind: true)
+      rescue SerializationError
+        # Unsupported values must not affect the render being observed.
+      end
+
+      def emit_drop_read(drop, key, value)
+        base = @bindings[drop]
+        return unless base
+
+        path = base + [key.to_s]
+        set_path(@drop_values, path, serialize(value, path, bind: true))
+      rescue SerializationError
+        # Unsupported values must not affect the render being observed.
+      end
+
+      def emit_file_read(path, source)
+        @files[path.to_s] = source.to_s
+      end
+
+      def emit_filter_call(name, input, arguments, output)
+        return if @tag_render_depth.positive?
+
+        @filter_calls << {
+          "name" => name.to_s,
+          "input" => serialize(input, ["filters", @filter_calls.length, "input"]),
+          "arguments" => serialize(arguments, ["filters", @filter_calls.length, "arguments"]),
+          "output" => serialize(output, ["filters", @filter_calls.length, "output"]),
+        }
+      rescue SerializationError
+        # Filter diagnostics must never make an otherwise replayable render fail.
+      end
+
+      def finish(output, context)
+        variables = deep_merge(@variables, @drop_values)
+        source = @root_template.instance_variable_get(:@template_recorder_source)
+        raise Error, "the rendered template was parsed outside the recording block" unless source
+
+        {
+          "format" => FORMAT,
+          "schema_version" => SCHEMA_VERSION,
+          "id" => SecureRandom.uuid,
+          "recorded_at" => Time.now.utc.iso8601,
+          "engine" => {
+            "liquid_version" => Liquid::VERSION,
+            "ruby_version" => RUBY_VERSION,
+            "strict_variables" => !!context.strict_variables,
+            "strict_filters" => !!context.strict_filters,
+          },
+          "template" => @templates.first,
+          "templates" => @templates,
+          "assigns" => variables,
+          "variable_outputs" => @variable_outputs,
+          "file_system" => @files,
+          "filter_calls" => @filter_calls,
+          "tag_calls" => @tag_calls,
+          "output" => output.to_s,
+        }
+      end
+
+      private
+
+      def serialize(value, path, seen = {}.compare_by_identity, bind: false)
+        case value
+        when nil, true, false, String, Integer, Float
+          value
+        when Symbol
+          value.to_s
+        when Liquid::Drop
+          @bindings[value] ||= path if bind
+          existing = value_at(@drop_values, @bindings[value])
+          existing || {}
+        when Hash
+          raise SerializationError, "circular value at #{format_path(path)}" if seen.key?(value)
+
+          seen[value] = true
+          result = value.each_with_object({}) do |(key, child), hash|
+            string_key = key.to_s
+            hash[string_key] = serialize(child, path + [string_key], seen, bind: bind)
+          end
+          seen.delete(value)
+          result
+        when Array
+          raise SerializationError, "circular value at #{format_path(path)}" if seen.key?(value)
+
+          seen[value] = true
+          result = value.each_with_index.map { |child, index| serialize(child, path + [index], seen, bind: bind) }
+          seen.delete(value)
+          result
+        else
+          raise SerializationError, "cannot record #{value.class} at #{format_path(path)}"
+        end
+      end
+
+      def set_path(root, path, value)
+        return root.replace(value) if path.empty? && value.is_a?(Hash)
+
+        cursor = root
+        path.each_with_index do |segment, index|
+          last = index == path.length - 1
+          if segment.is_a?(Integer)
+            break unless cursor.is_a?(Array)
+
+          end
+          cursor[segment] = last ? value : (cursor[segment] ||= container_for(path[index + 1]))
+          cursor = cursor[segment] unless last
+        end
+      end
+
+      def value_at(root, path)
+        return unless path
+
+        path.reduce(root) { |value, segment| value.respond_to?(:[]) ? value[segment] : nil }
+      end
+
+      def container_for(segment)
+        segment.is_a?(Integer) ? [] : {}
+      end
+
+      def deep_merge(left, right)
+        return right unless left.is_a?(Hash) && right.is_a?(Hash)
+
+        left.merge(right) { |_key, a, b| deep_merge(a, b) }
+      end
+
+      def format_path(path)
+        path.empty? ? "<root>" : path.join(".")
+      end
+    end
+
+    class Store
+      class << self
+        def jsonl?(path)
+          path.end_with?(".jsonl")
+        end
+
+        def append(path, record)
+          line = JSON.generate(record) << "\n"
+          File.open(path, File::WRONLY | File::CREAT | File::APPEND, 0o600) do |file|
+            file.flock(File::LOCK_EX)
+            file.write(line)
+            file.flush
+          end
+        end
+
+        def write_session(path, records)
+          payload = JSON.pretty_generate(
+            "format" => "liquid-recording-session",
+            "schema_version" => SCHEMA_VERSION,
+            "renders" => records,
+          ) << "\n"
+          directory = File.dirname(File.expand_path(path))
+          Tempfile.create([".liquid-recording", ".tmp"], directory, mode: File::RDWR, perm: 0o600) do |file|
+            file.write(payload)
+            file.flush
+            file.fsync
+            File.rename(file.path, path)
+          end
+        end
+
+        def read(path)
+          content = File.binread(path)
+          records = if jsonl?(path)
+            read_jsonl(content)
+          else
+            parsed = JSON.parse(content)
+            parsed["renders"] || [parsed]
+          end
+          records.each { |record| validate!(record) }
+          records
+        rescue Errno::ENOENT
+          raise ReplayError, "recording file not found: #{path}"
+        rescue JSON::ParserError => error
+          raise ReplayError, "invalid recording JSON: #{error.message}"
+        end
+
+        def read_jsonl(content)
+          lines = content.lines
+          lines.filter_map.with_index do |line, index|
+            next if line.strip.empty?
+
+            JSON.parse(line)
+          rescue JSON::ParserError
+            last_truncated_line = index == lines.length - 1 && !content.end_with?("\n")
+            raise unless last_truncated_line
+          end
+        end
+
+        def validate!(record)
+          raise ReplayError, "unsupported recording format" unless record["format"] == FORMAT
+          raise ReplayError, "unsupported schema version #{record["schema_version"].inspect}" unless record["schema_version"] == SCHEMA_VERSION
+
+          ['template', 'assigns', 'file_system', 'output'].each do |key|
+            raise ReplayError, "recording is missing #{key}" unless record.key?(key)
+          end
+          template = record["template"]
+          expected = Digest::SHA256.hexdigest(template.fetch("source"))
+          raise ReplayError, "template checksum does not match" unless template["sha256"] == expected
+        rescue KeyError, TypeError => error
+          raise ReplayError, "invalid recording schema: #{error.message}"
+        end
+      end
+    end
+
+    class MemoryFileSystem
+      def initialize(files)
+        @files = files
+      end
+
+      def read_template_file(path)
+        @files.fetch(path.to_s) { raise FileSystemError, "No such template '#{path}'" }
+      end
+    end
+
+    class Replayer
+      def initialize(record, mode: :compute, environment: nil)
+        @record = record
+        @mode = mode.to_sym
+        @environment = environment
+        unless [:compute, :strict, :verify].include?(@mode)
+          raise ReplayError, "mode must be :compute, :strict, or :verify"
+        end
+      end
+
+      def render(to: nil, filters: nil)
+        @filter_index = 0
+        @tag_index = 0
+        @variable_index = 0
+        parse_options = {}
+        parse_options[:environment] = strict_environment if @mode == :strict
+        template = Liquid::Template.parse(@record.dig("template", "source"), parse_options)
+        registers = { file_system: MemoryFileSystem.new(@record["file_system"]) }
+        registers[REPLAYER_REGISTER_KEY] = self if @mode == :strict && @record.key?("variable_outputs")
+        options = {
+          registers: registers,
+          strict_variables: @record.dig("engine", "strict_variables"),
+          strict_filters: @record.dig("engine", "strict_filters"),
+        }
+        options[:filters] = filters if filters
+        output = template.render!(@record["assigns"], options)
+        if @mode == :strict
+          verify_filter_count! unless @record.key?("variable_outputs")
+          verify_tag_count!
+          verify_variable_count!
+        end
+        if [:strict, :verify].include?(@mode) && output != @record["output"]
+          raise ReplayError, "replayed output does not match the recording"
+        end
+
+        File.binwrite(to, output) if to
+        output
+      end
+
+      def replay_filter(name)
+        call = @record["filter_calls"].fetch(@filter_index) do
+          raise ReplayError, "unexpected filter call #{name}"
+        end
+        if call["name"] != name.to_s
+          raise ReplayError, "expected filter #{call["name"]}, got #{name}"
+        end
+
+        @filter_index += 1
+        JSON.parse(JSON.generate(call["output"]))
+      end
+
+      def replay_variable
+        value = @record.fetch("variable_outputs").fetch(@variable_index) do
+          raise ReplayError, "unexpected variable render"
+        end
+        @variable_index += 1
+        value
+      end
+
+      def replay_tag(name)
+        call = @record.fetch("tag_calls", []).fetch(@tag_index) do
+          raise ReplayError, "unexpected tag call #{name}"
+        end
+        if call["name"] != name.to_s
+          raise ReplayError, "expected tag #{call["name"]}, got #{name}"
+        end
+
+        @tag_index += 1
+        call["output"]
+      end
+
+      def recorded_output
+        @record["output"]
+      end
+
+      def templates
+        @record["templates"]
+      end
+
+      private
+
+      def strict_environment
+        replayer = self
+        strainer = Class.new(Liquid::StrainerTemplate) do
+          define_method(:invoke) do |name, *_args|
+            replayer.replay_filter(name)
+          end
+        end
+        tags = (@environment || Liquid::Environment.default).tags.dup
+        tags&.each do |name, tag_class|
+          tags[name] = replay_tag_class(tag_class, name)
+        end
+        Liquid::Environment.build(tags: tags) do |environment|
+          environment.strainer_template = strainer
+        end
+      end
+
+      def replay_tag_class(tag_class, name)
+        replayer = self
+        Class.new(tag_class) do
+          define_method(:render_to_output_buffer) do |_context, output|
+            output << replayer.replay_tag(name)
+          end
+        end
+      end
+
+      def verify_variable_count!
+        return unless @record.key?("variable_outputs")
+
+        expected = @record["variable_outputs"].length
+        return if @variable_index == expected
+
+        raise ReplayError, "expected #{expected} variable renders, got #{@variable_index}"
+      end
+
+      def verify_tag_count!
+        expected = @record.fetch("tag_calls", []).length
+        return if @tag_index == expected
+
+        raise ReplayError, "expected #{expected} tag calls, got #{@tag_index}"
+      end
+
+      def verify_filter_count!
+        expected = @record["filter_calls"].length
+        return if @filter_index == expected
+
+        raise ReplayError, "expected #{expected} filter calls, got #{@filter_index}"
+      end
+    end
+  end
+end
diff --git a/lib/liquid/variable.rb b/lib/liquid/variable.rb
index 12b8d9f28..689b256c7 100644
--- a/lib/liquid/variable.rb
+++ b/lib/liquid/variable.rb
@@ -109,8 +109,14 @@ def render(context)
     end
 
     def render_to_output_buffer(context, output)
+      if (replayer = context.registers[TemplateRecorder::REPLAYER_REGISTER_KEY])
+        return output << replayer.replay_variable
+      end
+
+      output_start = output.length
       obj = render(context)
       render_obj_to_output(obj, output)
+      TemplateRecorder.current&.emit_variable_output(output[output_start..])
       output
     end
 
diff --git a/test/integration/blank_body_error_handling_test.rb b/test/integration/blank_body_error_handling_test.rb
index c428891e4..c395f867e 100644
--- a/test/integration/blank_body_error_handling_test.rb
+++ b/test/integration/blank_body_error_handling_test.rb
@@ -14,60 +14,60 @@ def assert_render_raises(source, error_mode:, assigns: {}, message: nil)
     error = assert_raises(Liquid::ArgumentError) do
       Liquid::Template.parse(source, line_numbers: true, error_mode: error_mode).render!(assigns)
     end
-    assert_includes error.message, message if message
+    assert_includes(error.message, message) if message
   end
 
   def test_blank_if_body_suppresses_inline_error_text_in_lax_and_strict
     [:lax, :strict].each do |mode|
-      assert_equal '', render_inline('{% if 5 > "x" %}{% endif %}', error_mode: mode)
+      assert_equal('', render_inline('{% if 5 > "x" %}{% endif %}', error_mode: mode))
     end
   end
 
   def test_blank_unless_body_suppresses_inline_error_text_in_lax_and_strict
     [:lax, :strict].each do |mode|
-      assert_equal '', render_inline('{% unless 5 > "x" %} {% endunless %}', error_mode: mode)
+      assert_equal('', render_inline('{% unless 5 > "x" %} {% endunless %}', error_mode: mode))
     end
   end
 
   def test_blank_for_body_suppresses_inline_error_text_in_lax_and_strict
     [:lax, :strict].each do |mode|
-      assert_equal '', render_inline('{% for i in (1..3) offset: xs %}{% endfor %}', error_mode: mode, assigns: { 'xs' => 'bad' })
+      assert_equal('', render_inline('{% for i in (1..3) offset: xs %}{% endfor %}', error_mode: mode, assigns: { 'xs' => 'bad' }))
     end
   end
 
   def test_strict2_blank_if_body_shows_inline_error_text
-    assert_equal COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% endif %}', error_mode: :strict2)
+    assert_equal(COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% endif %}', error_mode: :strict2))
   end
 
   def test_strict2_whitespace_if_body_shows_inline_error_text
-    assert_equal COMPARISON_ERROR, render_inline('{% if 5 > "x" %}   {% endif %}', error_mode: :strict2)
+    assert_equal(COMPARISON_ERROR, render_inline('{% if 5 > "x" %}   {% endif %}', error_mode: :strict2))
   end
 
   def test_strict2_assign_if_body_shows_inline_error_text
-    assert_equal COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% assign a = 1 %}{% endif %}', error_mode: :strict2)
+    assert_equal(COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% assign a = 1 %}{% endif %}', error_mode: :strict2))
   end
 
   def test_strict2_comment_if_body_shows_inline_error_text
-    assert_equal COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% comment %}c{% endcomment %}{% endif %}', error_mode: :strict2)
+    assert_equal(COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% comment %}c{% endcomment %}{% endif %}', error_mode: :strict2))
   end
 
   def test_strict2_capture_if_body_shows_inline_error_text
-    assert_equal COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% capture c %}text{% endcapture %}{% endif %}', error_mode: :strict2)
+    assert_equal(COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% capture c %}text{% endcapture %}{% endif %}', error_mode: :strict2))
   end
 
   def test_strict2_blank_unless_body_shows_inline_error_text
-    assert_equal COMPARISON_ERROR, render_inline('{% unless 5 > "x" %} {% endunless %}', error_mode: :strict2)
+    assert_equal(COMPARISON_ERROR, render_inline('{% unless 5 > "x" %} {% endunless %}', error_mode: :strict2))
   end
 
   def test_strict2_blank_for_body_shows_inline_error_text
-    assert_equal INVALID_INTEGER_ERROR, render_inline('{% for i in (1..3) offset: xs %}{% endfor %}', error_mode: :strict2, assigns: { 'xs' => 'bad' })
+    assert_equal(INVALID_INTEGER_ERROR, render_inline('{% for i in (1..3) offset: xs %}{% endfor %}', error_mode: :strict2, assigns: { 'xs' => 'bad' }))
   end
 
   def test_nonblank_bodies_show_inline_error_text_in_all_modes
     [:lax, :strict, :strict2].each do |mode|
-      assert_equal COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% echo 1 %}{% endif %}', error_mode: mode)
-      assert_equal COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{{ "" }}{% endif %}', error_mode: mode)
-      assert_equal COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% else %}E{% endif %}', error_mode: mode)
+      assert_equal(COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% echo 1 %}{% endif %}', error_mode: mode))
+      assert_equal(COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{{ "" }}{% endif %}', error_mode: mode))
+      assert_equal(COMPARISON_ERROR, render_inline('{% if 5 > "x" %}{% else %}E{% endif %}', error_mode: mode))
     end
   end
 
diff --git a/test/integration/error_handling_test.rb b/test/integration/error_handling_test.rb
index 0fda83ca5..6980fd197 100644
--- a/test/integration/error_handling_test.rb
+++ b/test/integration/error_handling_test.rb
@@ -269,7 +269,8 @@ def test_bug_compatible_silencing_of_errors_in_blank_nodes
     assert_equal("Liquid error: comparison of Integer with String failed0", output)
 
     output = Liquid::Template.parse("{% assign x = 0 %}{% if 1 < '2' %}{% assign x = 3 %}{% endif %}{{ x }}").render
-    assert_equal("0", output)
+    expected = ENV["LIQUID_PARSER_MODE"] == "strict2" ? "Liquid error: comparison of Integer with String failed0" : "0"
+    assert_equal(expected, output)
   end
 
   def test_syntax_error_is_raised_with_template_name
diff --git a/test/integration/template_recorder_test.rb b/test/integration/template_recorder_test.rb
new file mode 100644
index 000000000..c1ed8a6d3
--- /dev/null
+++ b/test/integration/template_recorder_test.rb
@@ -0,0 +1,360 @@
+# frozen_string_literal: true
+
+require_relative "../test_helper"
+require "tmpdir"
+
+class TemplateRecorderTest < Minitest::Test
+  class ProductDrop < Liquid::Drop
+    def initialize(title, secret)
+      super()
+      @title = title
+      @secret = secret
+    end
+
+    attr_reader :title
+
+    def details
+      DetailsDrop.new
+    end
+  end
+
+  class DetailsDrop < Liquid::Drop
+    def count
+      3
+    end
+  end
+
+  class LegacyFileSystem
+    attr_reader :reads
+
+    def initialize
+      @reads = []
+    end
+
+    def read_template_file(name)
+      @reads << name
+      "partial={{ product.title }}"
+    end
+  end
+
+  class UnsupportedLiquidValue
+    def to_liquid
+      self
+    end
+
+    def to_s
+      "unsupported"
+    end
+  end
+
+  class WrapperTag < Liquid::Block
+  end
+
+  class MarkerTag < Liquid::Tag
+    def render(_context)
+      "custom"
+    end
+  end
+
+  class CollectingWriter
+    attr_reader :records
+
+    def initialize
+      @records = []
+    end
+
+    def write(record)
+      @records << record
+    end
+  end
+
+  def setup
+    @directory = Dir.mktmpdir
+  end
+
+  def teardown
+    FileUtils.remove_entry(@directory)
+  end
+
+  def path(name = "recording.json")
+    File.join(@directory, name)
+  end
+
+  def test_records_and_verifies_a_render_without_changing_drop_behavior
+    product = ProductDrop.new("Computed", "must not be recorded")
+    template_source = "{{ product.title }} ({{ product.details.count }})"
+
+    output = Liquid::TemplateRecorder.record(path) do
+      Liquid::Template.parse(template_source).render!("product" => product)
+    end
+
+    assert_equal("Computed (3)", output)
+    record = Liquid::TemplateRecorder.records(path).first
+    assert_equal({ "title" => "Computed", "details" => { "count" => 3 } }, record.dig("assigns", "product"))
+    refute_includes(File.read(path), "must not be recorded")
+    assert_equal(output, Liquid::TemplateRecorder.replay_from(path, mode: :verify).render)
+  end
+
+  def test_preserves_the_legacy_one_argument_file_system_contract
+    file_system = LegacyFileSystem.new
+    environment = Liquid::Environment.build { |env| env.file_system = file_system }
+
+    output = Liquid::TemplateRecorder.record(path) do
+      Liquid::Template.parse("before {% include 'card' %}", environment: environment)
+        .render!("product" => ProductDrop.new("Hat", "secret"))
+    end
+
+    assert_equal("before partial=Hat", output)
+    assert_equal(["card"], file_system.reads)
+    record = Liquid::TemplateRecorder.records(path).first
+    assert_equal({ "card" => "partial={{ product.title }}" }, record["file_system"])
+    assert_equal(2, record["templates"].length)
+    assert_equal(output, Liquid::TemplateRecorder.replay_from(path, mode: :verify).render)
+  end
+
+  def test_jsonl_appends_one_self_contained_record_per_render
+    recording = path("renders.jsonl")
+
+    2.times do |sequence|
+      Liquid::TemplateRecorder.record(recording) do
+        Liquid::Template.parse("value={{ value }}").render!("value" => sequence)
+      end
+    end
+
+    assert_equal(2, File.readlines(recording).length)
+    assert_equal(["value=0", "value=1"], Liquid::TemplateRecorder.records(recording).map { |item| item["output"] })
+    assert_equal("value=0", Liquid::TemplateRecorder.replay_from(recording, index: 0).render)
+    assert_equal("value=1", Liquid::TemplateRecorder.replay_from(recording).render)
+  end
+
+  def test_json_session_supports_multiple_renders
+    Liquid::TemplateRecorder.record(path) do
+      Liquid::Template.parse("one={{ value }}").render!("value" => 1)
+      Liquid::Template.new.parse("two={{ value }}").render!("value" => 2)
+    end
+
+    records = Liquid::TemplateRecorder.records(path)
+    assert_equal(["one=1", "two=2"], records.map { |item| item["output"] })
+    assert_equal("two=2", Liquid::TemplateRecorder.replay_from(path).render)
+  end
+
+  def test_failed_recording_does_not_delete_an_existing_json_file
+    File.write(path, "existing")
+
+    assert_raises(RuntimeError) do
+      Liquid::TemplateRecorder.record(path) { raise "boom" }
+    end
+
+    assert_equal("existing", File.read(path))
+  end
+
+  def test_render_failure_is_not_written_to_jsonl
+    recording = path("renders.jsonl")
+
+    assert_raises(Liquid::UndefinedVariable) do
+      Liquid::TemplateRecorder.record(recording) do
+        Liquid::Template.parse("{{ missing }}").render!(nil, strict_variables: true)
+      end
+    end
+
+    refute_path_exists(recording)
+  end
+
+  def test_tampered_template_is_rejected
+    Liquid::TemplateRecorder.record(path) { Liquid::Template.parse("safe").render! }
+    session = JSON.parse(File.read(path))
+    session["renders"][0]["template"]["source"] = "changed"
+    File.write(path, JSON.generate(session))
+
+    error = assert_raises(Liquid::TemplateRecorder::ReplayError) do
+      Liquid::TemplateRecorder.replay_from(path)
+    end
+    assert_match(/checksum/, error.message)
+  end
+
+  def test_recordings_are_thread_local
+    paths = [path("a.json"), path("b.json")]
+    ready = Queue.new
+    release = Queue.new
+    threads = Array.new(2) do |index|
+      Thread.new do
+        Liquid::TemplateRecorder.record(paths[index]) do
+          ready << true
+          release.pop
+          Liquid::Template.parse("thread={{ value }}").render!("value" => index)
+        end
+      end
+    end
+    2.times { ready.pop }
+    2.times { release << true }
+    threads.each(&:join)
+
+    assert_equal("thread=0", Liquid::TemplateRecorder.records(paths[0]).first["output"])
+    assert_equal("thread=1", Liquid::TemplateRecorder.records(paths[1]).first["output"])
+  end
+
+  def test_nested_sessions_fail_without_corrupting_outer_session
+    error = nil
+    Liquid::TemplateRecorder.record(path) do
+      error = assert_raises(Liquid::TemplateRecorder::Error) do
+        Liquid::TemplateRecorder.record(path("inner.json")) { flunk }
+      end
+      Liquid::Template.parse("outer").render!
+    end
+
+    assert_match(/nested/, error.message)
+    assert_equal("outer", Liquid::TemplateRecorder.replay_from(path).render)
+  end
+
+  def test_supported_render_argument_forms_keep_working
+    filter = Module.new do
+      def decorate(input)
+        "[#{input}]"
+      end
+    end
+    template = nil
+    context = Liquid::Context.new([{ "value" => "context" }])
+
+    Liquid::TemplateRecorder.record(path) do
+      template = Liquid::Template.parse("{{ value | decorate }}")
+      assert_equal("[hash]", template.render({ "value" => "hash" }, filter))
+      context.add_filters(filter)
+      assert_equal("[context]", template.render(context))
+    end
+
+    assert_equal(2, Liquid::TemplateRecorder.records(path).length)
+  end
+
+  def test_strict_replay_uses_exact_recorded_filter_outputs
+    filter = Module.new do
+      def external_lookup(_input)
+        "x" * 150
+      end
+    end
+
+    output = Liquid::TemplateRecorder.record(path) do
+      Liquid::Template.parse("{{ key | external_lookup }}").render!({ "key" => "a" }, filter)
+    end
+
+    assert_equal("x" * 150, output)
+    assert_equal(output, Liquid::TemplateRecorder.replay_from(path, mode: :strict).render)
+    assert_equal("a", Liquid::TemplateRecorder.replay_from(path, mode: :compute).render)
+  end
+
+  def test_jsonl_reader_ignores_only_a_truncated_final_record
+    recording = path("renders.jsonl")
+    Liquid::TemplateRecorder.record(recording) { Liquid::Template.parse("complete").render! }
+    File.open(recording, "ab") { |file| file.write('{"format":') }
+
+    assert_equal(["complete"], Liquid::TemplateRecorder.records(recording).map { |item| item["output"] })
+  end
+
+  def test_accepts_a_pluggable_writer
+    writer = CollectingWriter.new
+
+    output = Liquid::TemplateRecorder.record(writer) do
+      Liquid::Template.parse("Hello {{ name }}").render!("name" => "Shopify")
+    end
+
+    assert_equal("Hello Shopify", output)
+    assert_equal(["Hello Shopify"], writer.records.map { |record| record["output"] })
+  end
+
+  def test_records_only_variables_resolved_by_the_template
+    unused = Object.new
+    assigns = { "visible" => "yes", "unused" => unused }
+
+    Liquid::TemplateRecorder.record(path) do
+      Liquid::Template.parse("{{ visible }}").render!(assigns)
+    end
+
+    assert_equal({ "visible" => "yes" }, Liquid::TemplateRecorder.records(path).first["assigns"])
+  end
+
+  def test_recording_scope_is_fiber_local
+    writer = CollectingWriter.new
+    ordinary_output = nil
+
+    Liquid::TemplateRecorder.record(writer) do
+      Fiber.new do
+        ordinary_output = Liquid::Template.parse("ordinary").render!
+      end.resume
+      Liquid::Template.parse("recorded").render!
+    end
+
+    assert_equal("ordinary", ordinary_output)
+    assert_equal(["recorded"], writer.records.map { |record| record["output"] })
+  end
+
+  def test_on_error_keeps_recording_failures_out_of_the_render_path
+    writer = Object.new
+    writer.define_singleton_method(:write) { |_record| raise "sink unavailable" }
+    errors = []
+
+    output = Liquid::TemplateRecorder.record(writer, on_error: errors.method(:<<)) do
+      Liquid::Template.parse("still rendered").render!
+    end
+
+    assert_equal("still rendered", output)
+    assert_equal(["sink unavailable"], errors.map(&:message))
+  end
+
+  def test_unsupported_accessed_values_do_not_affect_the_render
+    value = UnsupportedLiquidValue.new
+    writer = CollectingWriter.new
+
+    output = Liquid::TemplateRecorder.record(writer) do
+      Liquid::Template.parse("{{ value }}").render!("value" => value)
+    end
+
+    assert_equal(value.to_s, output)
+    assert_equal({}, writer.records.first["assigns"])
+  end
+
+  def test_strict_replay_accepts_the_application_environment
+    environment = Liquid::Environment.build(
+      tags: Liquid::Environment.default.tags.merge("marker" => MarkerTag),
+    )
+    writer = CollectingWriter.new
+    Liquid::TemplateRecorder.record(writer) do
+      Liquid::Template.parse("{% marker %}", environment: environment).render!
+    end
+
+    replay = Liquid::TemplateRecorder::Replayer.new(
+      writer.records.first,
+      mode: :strict,
+      environment: environment,
+    )
+
+    assert_equal("custom", replay.render)
+  end
+
+  def test_captures_file_reads_that_happen_before_the_template_starts_rendering
+    writer = CollectingWriter.new
+    Liquid::TemplateRecorder.record(writer) do
+      Liquid::TemplateRecorder.current.emit_file_read("card", "Card")
+      file_system = LegacyFileSystem.new
+      Liquid::Template.parse("{% render 'card' %}").render!({}, registers: { file_system: file_system })
+    end
+
+    replay = Liquid::TemplateRecorder::Replayer.new(writer.records.first, mode: :strict)
+
+    assert_equal("partial=", replay.render)
+  end
+
+  def test_strict_replay_skips_nested_custom_tag_calls
+    environment = Liquid::Environment.build(
+      tags: Liquid::Environment.default.tags.merge("wrapper" => WrapperTag, "marker" => MarkerTag),
+    )
+    writer = CollectingWriter.new
+    Liquid::TemplateRecorder.record(writer) do
+      Liquid::Template.parse("{% wrapper %}{% marker %}{% endwrapper %}", environment: environment).render!
+    end
+
+    record = writer.records.first
+    replay = Liquid::TemplateRecorder::Replayer.new(record, mode: :strict, environment: environment)
+
+    assert_equal(["wrapper"], record["tag_calls"].map { |call| call["name"] })
+    assert_equal("custom", replay.render)
+  end
+end