fix: Detect replaced work in projections

Compare complete staged work around projections so draining one intent
and staging another cannot bypass the read-only contract. Cover effect
and commit-action replacement with terminal rollback regressions.

Author: cardmagic

CHANGELOG.md

@@ -4,7 +4,8 @@
 
 - Reject query and observable state mutation and staged durable work with
   terminal `QueryMutatedState` errors. Cover individual snapshot projections and
-  preserve ordinary operations' already-staged work while reading projections.
+  preserve ordinary operations' already-staged work while reading projections,
+  including replacements that leave the intent count unchanged.
 - Pin reserved JSON property names with shared Ruby/JS fixtures. Document the
   reminder-name limit difference and the authorized dead-transmit retry API.
 

lib/solid_objects/actor.rb

@@ -556,15 +556,22 @@ def intent_count
     # @rbs (String) { () -> untyped } -> untyped
     def read_projection(operation)
       state_before = state.to_h
-      intents_before = intent_count
+      intents_before = intent_snapshot
       result = guard_application_writes(operation) { yield }
-      unless state.to_h == state_before && intent_count == intents_before
+      unless state.to_h == state_before && intent_snapshot == intents_before
         raise QueryMutatedState, "observables must not mutate actor state or stage durable work"
       end
 
       result
     end
 
+    # @rbs () -> Array[Array[Hash[Symbol, untyped]]]
+    def intent_snapshot
+      [ effect_intents, effect_recovery_intents, commit_action_intents, reminder_intents, outbound_message_intents ].map do |intents|
+        intents.map { |intent| intent.to_h.deep_dup }
+      end
+    end
+
     # @rbs (String) { () -> untyped } -> untyped
     def guard_application_writes(operation, &block)
       ApplicationWriteGuard.call(

sig/generated/lib/solid_objects/actor.rbs

@@ -366,6 +366,9 @@ module SolidObjects
     # @rbs (String) { () -> untyped } -> untyped
     def read_projection: (String) { () -> untyped } -> untyped
 
+    # @rbs () -> Array[Array[Hash[Symbol, untyped]]]
+    def intent_snapshot: () -> Array[Array[Hash[Symbol, untyped]]]
+
     # @rbs (String) { () -> untyped } -> untyped
     def guard_application_writes: (String) { () -> untyped } -> untyped
 

test/integration/read_only_actor_test.rb

@@ -26,6 +26,11 @@ def append
       items
     end
 
+    def append_with_work(action:)
+      perform_action(action)
+      append
+    end
+
     private
 
     def perform_action(action)
@@ -36,6 +41,12 @@ def perform_action(action)
       when "reminder" then schedule(at: Time.now + 60).append
       when "outbound" then send_to(self.class.ref("other")).append
       when "state" then items << "unexpected"
+      when "replace_effect", "replace_commit_action"
+        return unless intent_count.positive?
+
+        discard_intents
+        emit(:replacement) if action == "replace_effect"
+        commit_action(:replacement) if action == "replace_commit_action"
       end
     end
   end
@@ -45,6 +56,7 @@ def perform_action(action)
     SolidObjects.configuration.max_attempts = 3
     SolidObjects.configuration.retry_delay = ->(_) { 0 }
     SolidObjects.register_commit_action(:unexpected) { SolidObjectsTestDomainRecord.create!(name: "unexpected") }
+    SolidObjects.register_commit_action(:replacement) { SolidObjectsTestDomainRecord.create!(name: "replacement") }
   end
 
   %w[effect recovery commit_action reminder outbound state].each do |action|
@@ -75,6 +87,17 @@ def perform_action(action)
     assert_empty SolidObjects::Instance.all
   end
 
+  %w[effect commit_action].each do |action|
+    test "observables cannot replace staged #{action} with the same intent count" do
+      Reader.projection_action = "replace_#{action}"
+
+      error = assert_raises(SolidObjects::MessageFailed) { Reader.ref("one").sync.append_with_work(action:) }
+
+      assert_equal "SolidObjects::QueryMutatedState", error.details.fetch("class")
+      assert_no_committed_work
+    end
+  end
+
   test "pure projections preserve effects already staged by an operation" do
     reader = Reader.new(actor_id: "local", state: SolidObjects::State.new(Reader.definition.state_definition))
     reader.emit(:expected)