feat: stage transmit effects with an ordered drain

Add Actor#transmit and SolidObjects.register_transmit, the staging
side of the transmit family and the counterpart of the ingest in the
previous commit. transmit returns the same fluent dispatcher schedule
returns and stages a solid-objects.transmit effect in the actor
commit; a raw emit with explicit actorType/actorId targets a
different actor, matching the JS staging surface.

register_transmit wraps register_effect with envelope construction
and the ordered drain ported from solid-objects-js: a claimed
transmit effect delivers every undelivered sibling for its actor up
to its own mailbox sequence, oldest first. Per-actor order therefore
survives a failed delivery, and redelivery is safe because the
receiving side dedups on transmit:<effectId>. A raised delivery
retries with backoff and dead-letters on exhaustion, like any other
effect; a malformed staged effect raises InvalidTransmission and is
skipped by sibling drains rather than blocking them.

The drain-ordering test was verified against a neutered drain that
delivers only the claimed effect: it fails with [2] where [1, 2] is
expected, so the test observes the guarantee it claims.

Closes #48

Author: cardmagic

CHANGELOG.md

@@ -14,6 +14,16 @@
   controller boundary. Golden fixtures in
   `compatibility/transmit-envelopes.json` pin the wire contract shared with
   the JS runtime.
+- Add `Actor#transmit` and `SolidObjects.register_transmit`, the staging
+  side of the transmit family. `transmit.increment(amount:)` stages a
+  `solid-objects.transmit` effect in the same commit as the state change;
+  `register_transmit` drains staged effects into camelCase envelopes and
+  hands each to the delivery block, which raises to retry. A claimed
+  transmit effect delivers every undelivered sibling for its actor up to
+  its own mailbox sequence, oldest first, so per-actor order survives a
+  failed delivery, and the receiving side dedups on `transmit:<effectId>`.
+  A raw `emit "solid-objects.transmit"` with explicit `actorType` and
+  `actorId` targets a different actor, matching the JS staging surface.
 
 ## 0.13.3 - 2026-08-18
 

docs/roadmap.md

@@ -72,13 +72,17 @@
   Rails 7.1 and 7.2 is unmeasured against those servers. Rails 7.0 is out of
   range because its SQLite adapter requires `sqlite3 ~> 1.4`, and this gem needs
   the busy-handler control that arrived in `sqlite3` 2.x
-- `SolidObjects::Transmission.receive`, the server ingest for browser
-  transmit envelopes from solid-objects-js: envelope validation, actor type
-  resolution with a per-call `resolve_actor_type:` escape hatch, and an
-  internal idempotent enqueue keyed `transmit:<effectId>`, with the wire
-  contract pinned by golden fixtures in
-  `compatibility/transmit-envelopes.json`. Ingest only; the staging side in
-  Ruby and an engine-mounted route are not implemented
+- The transmit family, both sides. `SolidObjects::Transmission.receive` is
+  the ingest: envelope validation, actor type resolution with a per-call
+  `resolve_actor_type:` escape hatch, and an internal idempotent enqueue
+  keyed `transmit:<effectId>`. `Actor#transmit` and
+  `SolidObjects.register_transmit` are the staging side: a transactional
+  `solid-objects.transmit` effect and a drain that delivers every
+  undelivered sibling for the actor up to the claimed effect's mailbox
+  sequence, oldest first, so per-actor order survives a failed delivery.
+  The wire contract is pinned by golden fixtures in
+  `compatibility/transmit-envelopes.json`. An engine-mounted ingest route
+  is not implemented
 - A JavaScript suite covering every browser module, run in CI with Node's test
   runner and jsdom, plus a browser suite running the same modules against real
   Chromium and a real Turbo build, with every GitHub Actions reference pinned to

docs/transmission.md

@@ -1,14 +1,64 @@
-# Transmit ingest
-
-`SolidObjects::Transmission.receive(envelope)` is the server side of the
-browser transmit family in
-[solid-objects-js](https://github.com/cardmagic/solid-objects-js). A
-solid-objects-js actor in a browser stages a transmit intent with
-`this.transmit().increment({ amount })` in the same transaction as its state
-change. The browser's effect worker drains that outbox with at-least-once
-delivery, per-actor order, and retry backoff. It posts one JSON envelope per
-effect to a route the host application owns. `Transmission.receive` replays
-that envelope onto a server actor.
+# The transmit family
+
+The transmit family replays one runtime's actor operations onto another
+runtime over a shared wire contract. The Ruby gem holds both sides:
+
+- `Actor#transmit` and `SolidObjects.register_transmit` stage and deliver
+  envelopes. This is the sending side.
+- `SolidObjects::Transmission.receive(envelope)` ingests envelopes. This is
+  the receiving side.
+
+[solid-objects-js](https://github.com/cardmagic/solid-objects-js) holds the
+same two sides for the browser and Node. A browser actor stages a transmit
+intent with `this.transmit().increment({ amount })` in the same transaction
+as its state change; its effect worker drains that outbox with
+at-least-once delivery, per-actor order, and retry backoff, and posts one
+JSON envelope per effect to a route the host application owns. A Rails
+actor does the same with `transmit.increment(amount:)`. Either ingest
+accepts either sender, so Rails-to-Rails, Rails-to-Node, Node-to-Rails,
+and browser-to-Rails replication all ride one contract.
+
+## The sending side
+
+```ruby
+class Counter < SolidObjects::Actor
+  actor_type "counters"
+
+  attribute :count, default: 0
+
+  def increment(amount: 1)
+    self.count += amount
+    transmit.increment(amount:)
+  end
+end
+
+SolidObjects.register_transmit do |envelope|
+  DeliverToUpstream.call(envelope)
+end
+```
+
+`transmit` returns the same fluent dispatcher `schedule` returns. It stages
+one `solid-objects.transmit` effect in the same commit as the state change,
+targeting the same operation on the same actor in the receiving runtime. For
+a different target, stage the effect directly:
+
+```ruby
+emit "solid-objects.transmit",
+  operation: "increment",
+  arguments: { amount: 2 },
+  actorType: "other-counters",
+  actorId: "counter-1"
+```
+
+`SolidObjects.register_transmit(&deliver)` registers the drain handler for
+that effect. The block receives one camelCase envelope per staged effect.
+Raise inside the block while the upstream is unreachable; the effect
+retries with backoff and dead-letters on exhaustion, like any other effect.
+
+The drain keeps per-actor order across failures: a claimed transmit effect
+delivers every undelivered sibling for its actor up to its own mailbox
+sequence, oldest first. The receiving side dedups on `transmit:<effectId>`,
+so a redelivered envelope applies once.
 
 ## Wire contract
 
@@ -79,7 +129,6 @@ SolidObjects::Transmission.receive(
 
 ## Scope
 
-`receive` is ingest only. The staging side in Ruby, an `actor.transmit` for
-Ruby-to-Ruby replication, is a separate feature. An engine-mounted route
-with an authentication hook is a possible follow-up; it stays out because
-it carries authentication, CSRF, and rate-limit decisions of its own.
+An engine-mounted route with an authentication hook is a possible
+follow-up; it stays out because it carries authentication, CSRF, and
+rate-limit decisions of its own.

lib/solid_objects.rb

@@ -101,6 +101,20 @@ def register_effect(name, &handler)
       effect_registry.register(name, handler)
     end
 
+    # @rbs (?effect_name: String | Symbol) { (Hash[String, untyped]) -> untyped } -> Proc
+    def register_transmit(effect_name: Transmission::EFFECT_NAME, &deliver)
+      raise ArgumentError, "register_transmit requires a delivery block" unless deliver
+
+      register_effect(effect_name) do |arguments, context|
+        Transmission.deliver_through(
+          effect_name: effect_name.to_s,
+          arguments:,
+          context:,
+          deliver:
+        )
+      end
+    end
+
     # @rbs () -> CommitActionRegistry
     def commit_action_registry
       @commit_action_registry ||= CommitActionRegistry.new

lib/solid_objects/actor.rb

@@ -190,6 +190,17 @@ def emit(name, on_success: nil, on_failure: nil, **arguments)
       nil
     end
 
+    # @rbs () -> OperationDispatcher
+    def transmit
+      OperationDispatcher.new(
+        actor_type: self.class.actor_type,
+        handlers: self.class.definition.messages
+      ) do |operation, arguments|
+        emit(Transmission::EFFECT_NAME, operation: operation.to_s, arguments:)
+        nil
+      end
+    end
+
     # @rbs (Symbol | String, **untyped) -> nil
     def commit_action(name, **arguments)
       CommitActionIntent.new(

lib/solid_objects/transmission.rb

@@ -4,6 +4,8 @@ module SolidObjects
   module Transmission
     REQUIRED_FIELDS = %w[effectId actorType actorId operation].freeze
     IDEMPOTENCY_PREFIX = "transmit:"
+    EFFECT_NAME = "solid-objects.transmit"
+    UNDELIVERED_STATUSES = %w[pending processing].freeze
 
     class << self
       # @rbs (untyped envelope, ?resolve_actor_type: ^(String) -> (String | Symbol), ?mailbox: Mailbox) -> MessageReference
@@ -26,8 +28,79 @@ def receive(envelope, resolve_actor_type: :itself.to_proc, mailbox: Mailbox.new)
         )
       end
 
+      # @rbs (effect_name: String, arguments: Hash[String, untyped], context: EffectContext, deliver: Proc) -> nil
+      def deliver_through(effect_name:, arguments:, context:, deliver:)
+        staged_envelope(
+          arguments,
+          effect_id: context.id,
+          actor_type: context.actor_type,
+          actor_id: context.actor_id
+        )
+        undelivered_envelopes_through(effect_name:, context:).each do |envelope|
+          deliver.call(envelope)
+        end
+        nil
+      end
+
       private
 
+      # @rbs (Hash[String, untyped], effect_id: String, actor_type: String, actor_id: String) -> Hash[String, untyped]
+      def staged_envelope(arguments, effect_id:, actor_type:, actor_id:)
+        operation = arguments["operation"]
+        unless operation.is_a?(String) && !operation.empty?
+          raise InvalidTransmission, "transmit effect arguments require a non-empty operation"
+        end
+
+        target_arguments = arguments["arguments"]
+        target_arguments = {} if target_arguments.nil?
+        unless target_arguments.is_a?(Hash)
+          raise InvalidTransmission, %(transmit effect arguments must hold a JSON object in "arguments")
+        end
+
+        target_type = arguments["actorType"].nil? ? actor_type : arguments["actorType"]
+        target_id = arguments["actorId"].nil? ? actor_id : arguments["actorId"]
+        { "actorType" => target_type, "actorId" => target_id }.each do |field, value|
+          next if value.is_a?(String) && !value.empty?
+
+          raise InvalidTransmission, "transmit effect #{field} must be a non-empty string"
+        end
+
+        {
+          "effectId" => effect_id,
+          "actorType" => target_type,
+          "actorId" => target_id,
+          "operation" => operation,
+          "arguments" => target_arguments
+        }
+      end
+
+      # @rbs (effect_name: String, context: EffectContext) -> Array[Hash[String, untyped]]
+      def undelivered_envelopes_through(effect_name:, context:)
+        source_sequence = Message.find(context.source_message_id).sequence
+        effects = Effect
+          .joins(:message)
+          .where(name: effect_name, status: UNDELIVERED_STATUSES)
+          .merge(
+            Message.where(
+              actor_type: context.actor_type,
+              actor_id: context.actor_id,
+              sequence: ..source_sequence
+            )
+          )
+          .order(Message.arel_table[:sequence].asc, :id)
+
+        effects.filter_map do |effect|
+          staged_envelope(
+            effect.arguments,
+            effect_id: effect.effect_id,
+            actor_type: context.actor_type,
+            actor_id: context.actor_id
+          )
+        rescue InvalidTransmission
+          nil
+        end
+      end
+
       # @rbs (untyped) -> void
       def validate!(envelope)
         raise InvalidTransmission, "envelope must be a JSON object" unless envelope.is_a?(Hash)

sig/generated/lib/solid_objects.rbs

@@ -18,6 +18,9 @@ module SolidObjects
   # @rbs (String | Symbol) { (Hash[String, untyped], EffectContext) -> untyped } -> Proc
   def self.register_effect: (String | Symbol) { (Hash[String, untyped], EffectContext) -> untyped } -> Proc
 
+  # @rbs (?effect_name: String | Symbol) { (Hash[String, untyped]) -> untyped } -> Proc
+  def self.register_transmit: (?effect_name: String | Symbol) { (Hash[String, untyped]) -> untyped } -> Proc
+
   # @rbs () -> CommitActionRegistry
   def self.commit_action_registry: () -> CommitActionRegistry
 

sig/generated/lib/solid_objects/actor.rbs

@@ -161,6 +161,9 @@ module SolidObjects
     # @rbs (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, **untyped) -> nil
     def emit: (Symbol | String, ?on_success: Symbol | String?, ?on_failure: Symbol | String?, **untyped) -> nil
 
+    # @rbs () -> OperationDispatcher
+    def transmit: () -> OperationDispatcher
+
     # @rbs (Symbol | String, **untyped) -> nil
     def commit_action: (Symbol | String, **untyped) -> nil
 

sig/generated/lib/solid_objects/transmission.rbs

@@ -6,9 +6,22 @@ module SolidObjects
 
     IDEMPOTENCY_PREFIX: ::String
 
+    EFFECT_NAME: ::String
+
+    UNDELIVERED_STATUSES: untyped
+
     # @rbs (untyped envelope, ?resolve_actor_type: ^(String) -> (String | Symbol), ?mailbox: Mailbox) -> MessageReference
     def self.receive: (untyped envelope, ?resolve_actor_type: ^(String) -> (String | Symbol), ?mailbox: Mailbox) -> MessageReference
 
+    # @rbs (effect_name: String, arguments: Hash[String, untyped], context: EffectContext, deliver: Proc) -> nil
+    def self.deliver_through: (effect_name: String, arguments: Hash[String, untyped], context: EffectContext, deliver: Proc) -> nil
+
+    # @rbs (Hash[String, untyped], effect_id: String, actor_type: String, actor_id: String) -> Hash[String, untyped]
+    private def self.staged_envelope: (Hash[String, untyped], effect_id: String, actor_type: String, actor_id: String) -> Hash[String, untyped]
+
+    # @rbs (effect_name: String, context: EffectContext) -> Array[Hash[String, untyped]]
+    private def self.undelivered_envelopes_through: (effect_name: String, context: EffectContext) -> Array[Hash[String, untyped]]
+
     # @rbs (untyped) -> void
     private def self.validate!: (untyped) -> void
   end