Add fluent actor dispatch

Author: cardmagic

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## Unreleased
+
+- Replace positional actor dispatch with fluent operation selection. Direct
+  committed calls remain `account.disable(...)`; configured committed calls
+  use `account.sync(timeout: ...).status`; asynchronous calls use
+  `account.async(...).disable(...)`; actor outbox delivery uses
+  `send_to(account, ...).disable(...)`; and reminders use
+  `schedule(at: ...).evaluate(...)`. Delivery and reminder options are now
+  unambiguously separate from actor message arguments. The former positional
+  `async`, `sync`, `send_to`, and `schedule` forms are removed.
+- Validate fluent operations before enqueueing or staging them. Direct and
+  configured synchronous calls accept messages and queries, while `async`,
+  `send_to`, and `schedule` accept public actor messages only.
+- Keep SQLite reconnect failures inside a synchronous lock deadline. Active
+  Record can reconnect after a lock error while the lock is still held, and
+  configuring WAL then raises `SQLite3::CantOpenException`; it now retries
+  within the original deadline and surfaces the documented sync timeout.
+
 ## 0.11.0 - 2026-08-11
 
 - Add `SolidObjects.configuration.register_component`. An extension gem can now

README.md

@@ -27,7 +27,7 @@ current_count = counter.value
 current_snapshot = counter.snapshot.value
 
 # Durable fire-and-forget delivery. A worker processes it later.
-message = counter.async(:increment, amount: 5)
+message = counter.async.increment(amount: 5)
 ```
 
 `Counter / global` is a logical identity. Like a Durable Object named with
@@ -41,10 +41,10 @@ The invocation model is the first adoption decision:
 | Call | Returns | Worker fleet required? |
 | --- | --- | --- |
 | `counter.increment(amount: 5)` | Committed handler result | No |
-| `counter.sync(:increment, amount: 5)` | Committed handler result | No |
+| `counter.sync(timeout: 5.seconds).increment(amount: 5)` | Committed handler result | No |
 | `counter.value` | Ordered, committed query result | No |
 | `counter.snapshot.value` | Current committed state without a mailbox message | No |
-| `counter.async(:increment, amount: 5)` | `MessageReference` immediately | Yes |
+| `counter.async.increment(amount: 5)` | `MessageReference` immediately | Yes |
 
 Direct methods and `sync` durably enqueue the call, then the Rails caller helps
 execute the actor through the same mailbox, lease, and fencing path as a
@@ -155,7 +155,7 @@ durable reminders owned by one logical identity:
 
 ```ruby
 def schedule_expiration
-  schedule :expire, at: 30.minutes.from_now, arguments: {}
+  schedule(at: 30.minutes.from_now).expire
 end
 ```
 
@@ -536,7 +536,7 @@ cart.add_item(product_id: "shirt-123", quantity: 2)
 items = cart.items
 ```
 
-Use `cart.async(:add_item, product_id: "shirt-123", quantity: 2)` to enqueue
+Use `cart.async.add_item(product_id: "shirt-123", quantity: 2)` to enqueue
 without waiting; that call returns a `SolidObjects::MessageReference`. `items`
 is a deeply frozen JSON snapshot, so mutating it cannot bypass the actor
 mailbox. State changes must go through public actor methods or explicit
@@ -638,26 +638,34 @@ Use `async` for durable fire-and-forget work. It returns a
 
 ```ruby
 message = order.async(
-  :submit,
-  idempotency_key: "submit-order-123"
-)
+  idempotency_key: "submit-order-123",
+  authorization_context: Current.user
+).submit
 ```
 
 Use `available_at:` to spread bulk work or delay one message:
 
 ```ruby
-order.async(:evaluate, available_at: 10.minutes.from_now)
+order.async(available_at: 10.minutes.from_now).evaluate
 ```
 
 ### `sync`
 
-Use explicit `sync` when the operation name is dynamic or collides with a
-reference method:
+Use explicit `sync` when the invocation needs a timeout, idempotency key, or
+authorization context different from the defaults:
 
 ```ruby
-status = order.sync(:status, timeout: 5.seconds)
+status = order.sync(
+  timeout: 5.seconds,
+  authorization_context: Current.user
+).status
 ```
 
+Delivery configuration belongs on `async(...)` or `sync(...)` before the
+operation. Keywords on the final method call are always actor message
+arguments, so `order.sync(timeout: 5.seconds).record(timeout: "payload")`
+keeps the invocation timeout separate from the payload value.
+
 Direct calls and `sync` use the same caller-assisted execution path. A healthy
 actor normally needs no worker round trip, making this path suitable for HTTP
 and MCP request/response boundaries when the handler itself fits the
@@ -677,7 +685,7 @@ recover its eventual result through the durable message identity:
 
 ```ruby
 begin
-  order.submit(timeout: 250.milliseconds)
+  order.sync(timeout: 250.milliseconds).submit
 rescue SolidObjects::SyncTimeout => error
   result = error.message_reference.wait(
     timeout: 5.seconds,
@@ -700,6 +708,17 @@ Actor code cannot use direct calls or `sync` on another actor; synchronous
 actor-to-actor waits can deadlock in cycles. Use `async` or `send_to` and a
 result message.
 
+```ruby
+send_to(
+  audit_log,
+  available_at: 5.minutes.from_now,
+  idempotency_key: event_id
+).record(event_id:, event_name: "account_disabled")
+```
+
+Actor-to-actor delivery is staged with the current turn, returns `nil`, and is
+discarded if that turn does not commit. It accepts messages, not queries.
+
 ### Domain rejection
 
 Reject invalid input without retrying or creating a dead letter:
@@ -831,7 +850,11 @@ API. One-shot and recurring alarms are actor-owned database records:
 
 ```ruby
 def schedule_evaluation
-  schedule :evaluate, at: 1.hour.from_now, every: 1.hour, missed: :latest
+  schedule(
+    at: 1.hour.from_now,
+    every: 1.hour,
+    missed: :latest
+  ).evaluate(account_id:)
 end
 ```
 
@@ -852,7 +875,7 @@ once. It also means this is a data-loss bug:
 # Wrong. Every entry overwrites the previous entry's alarm.
 def add(entry:)
   self.entries = entries + [ entry ]
-  schedule :deliver, at: entry.fetch("wait_until"), arguments: {}
+  schedule(at: entry.fetch("wait_until")).deliver
 end
 ```
 
@@ -882,7 +905,7 @@ def arm_next
   earliest = entries.first
   return unless earliest
 
-  schedule :deliver, at: Time.at(earliest.fetch("wait_until")), arguments: {}
+  schedule(at: Time.at(earliest.fetch("wait_until"))).deliver
 end
 ```
 

benchmark/support.rb

@@ -98,13 +98,13 @@ def measure(name)
     def enqueue
       reference = CounterActor.ref("enqueue")
       measure("enqueue #{count} messages") do
-        count.times { reference.async(:increment) }
+        count.times { reference.async.increment }
       end
     end
 
     # @rbs () -> void
     def claim
-      count.times { |index| CounterActor.ref("claim-#{index}").async(:increment) }
+      count.times { |index| CounterActor.ref("claim-#{index}").async.increment }
       process_registry = SolidObjects::ProcessRegistry.new
       owner_id = process_registry.register.id
       activation_manager = SolidObjects::ActivationManager.new(owner_id:)
@@ -137,7 +137,7 @@ def processing
 
     # @rbs () -> void
     def cold_actors
-      count.times { |index| CounterActor.ref("cold-#{index}").async(:increment) }
+      count.times { |index| CounterActor.ref("cold-#{index}").async.increment }
       worker = SolidObjects::Worker.new
       measure("process #{count} cold actors") { drain(worker) }
     ensure
@@ -147,7 +147,7 @@ def cold_actors
     # @rbs () -> void
     def hot_actor
       reference = CounterActor.ref("hot")
-      count.times { reference.async(:increment) }
+      count.times { reference.async.increment }
       worker = SolidObjects::Worker.new
       measure("process #{count} messages for one hot actor") { drain(worker) }
     ensure
@@ -172,7 +172,7 @@ def sync_latency
 
       count.times do |index|
         started_at = monotonic_now
-        CounterActor.ref("sync-#{index}").sync(:count, timeout: 5)
+        CounterActor.ref("sync-#{index}").sync(timeout: 5).count
         samples << monotonic_now - started_at
       end
 
@@ -211,7 +211,7 @@ def adoption_latency
     # @rbs () -> void
     def activation_cache
       reference = CounterActor.ref("cache")
-      count.times { reference.async(:increment) }
+      count.times { reference.async.increment }
       activations = 0
       subscriber = ActiveSupport::Notifications.subscribe("solid_objects.activation.started") do
         activations += 1
@@ -303,7 +303,7 @@ def query_count
     # Runs the turn on the measuring thread, so nothing else can contribute.
     # @rbs () -> Integer
     def message_turn_query_count
-      CounterActor.ref("queries").async(:increment)
+      CounterActor.ref("queries").async.increment
       worker = SolidObjects::Worker.new
       count_queries { worker.run_once }
     ensure
@@ -322,8 +322,8 @@ def synchronous_caller_query_count
       reference = CounterActor.ref("sync-queries")
       worker = SolidObjects::Worker.new
       runner = Thread.new { worker.run }
-      reference.sync(:increment)
-      count_queries { reference.sync(:increment) }
+      reference.sync.increment
+      count_queries { reference.sync.increment }
     ensure
       worker&.request_shutdown
       runner&.join(5)
@@ -428,7 +428,7 @@ def load_models
     def enqueue_round_robin
       actor_count = [ concurrency * 10, count ].min
       references = Array.new(actor_count) { |index| CounterActor.ref("actor-#{index}") }
-      count.times { |index| references[index % actor_count].async(:increment) }
+      count.times { |index| references[index % actor_count].async.increment }
     end
 
     # @rbs (SolidObjects::Worker) -> Integer

docs/architecture.md

@@ -63,8 +63,9 @@ The registry maps a stable persisted actor type string to a Ruby actor class. Re
 A reference contains actor type and normalized actor ID. It is cheap,
 serializable as data, and does not imply an active Ruby object. Declared
 message, query, and attribute methods use synchronous caller-assisted
-invocation. `sync` provides the same behavior for a dynamic operation name,
-while `async` only durably enqueues a message and returns its reference.
+invocation. `sync(...)` configures a caller-assisted invocation before its
+operation is selected, while `async(...)` configures durable enqueue and
+returns its message reference after the operation is selected.
 `destroy` is a reserved synchronous reference operation. Every path authorizes
 through the client.
 
@@ -155,8 +156,9 @@ Public instance methods declared on the actor are messages. Declare helpers as
 private or protected. Messages and queries are exposed as methods on a
 reference through the same synchronous caller-assisted path. Returned state
 snapshots are deeply frozen. Use `async` for durable fire-and-forget delivery
-and `sync` for dynamic operation names. The explicit `message` DSL remains
-available for dynamic definitions.
+and configured `sync` when a committed call needs a non-default timeout,
+idempotency key, or authorization context. The explicit `message` DSL remains
+available for declarations defined dynamically.
 
 `reference.snapshot` is the explicit unordered read path. It invokes query
 authorization, reads the most recently committed instance state without
@@ -426,7 +428,7 @@ An effect handler receives JSON arguments plus an effect context. Outcome messag
 Actor code sends to another actor through a staged outbox:
 
 ```ruby
-send_to InventoryActor.ref("sku-123"), :reserve, order_id: id, quantity: 2
+send_to(InventoryActor.ref("sku-123")).reserve(order_id: id, quantity: 2)
 ```
 
 The source actor commit never waits for the target. The outbox worker allocates the target's sequence after source commit. There is no global order across actors.
@@ -436,7 +438,7 @@ The source actor commit never waits for the target. The outbox worker allocates
 A reminder record contains actor identity, a reminder name, target message, JSON arguments, next run time, optional interval, status, and occurrence counter.
 
 ```ruby
-schedule :expire, at: 30.minutes.from_now, arguments: {}
+schedule(at: 30.minutes.from_now).expire
 ```
 
 Reminders are keyed by `(actor, reminder name)`, enforced by a unique index on `(instance_id, name)`. `schedule` is therefore an upsert: scheduling a name that is already armed moves that alarm instead of adding another, which is what makes re-arming safe from a handler that may run more than once. An actor needing several pending items should arm one alarm for the earliest and drain everything due when it fires, rather than one alarm per item; the [reminders guide](../README.md#a-reminder-is-one-named-alarm-per-actor) shows that pattern. A move that changes `next_run_at` emits `solid_objects.reminder.replaced`, because the replacement is otherwise indistinguishable from a first schedule.
@@ -467,7 +469,7 @@ The expected drift categories are actors with a lost alarm, missing actors for l
 
 Bulk repair updates to `solid_objects_instances` are forbidden. They bypass activation ownership and fencing and can overwrite a concurrently committed actor state. Direct reads are observational; writes go through actor messages.
 
-Large repairs use `async(..., available_at:)` to spread work over an application-defined dispatch window. The durable message records the requested availability and ready membership drives the hot polling query. This prevents reconciliation from flooding mailboxes and starving normal traffic.
+Large repairs use `async(available_at: ...).repair(...)` to spread work over an application-defined dispatch window. The durable message records the requested availability and ready membership drives the hot polling query. This prevents reconciliation from flooding mailboxes and starving normal traffic.
 
 ## Realtime integration
 

docs/development.md

@@ -56,7 +56,7 @@ Use `drain_solid_objects` to process actor, reminder, effect, callback, and
 broadcast work to a deterministic fixed point without arbitrary sleeps:
 
 ```ruby
-message = Counter.ref("test").async(:increment)
+message = Counter.ref("test").async.increment
 
 assert_equal 1, drain_solid_objects
 assert_equal "completed", message.status

docs/migrating-existing-state.md

@@ -71,11 +71,9 @@ Give every bootstrap call an idempotency key derived from the legacy record:
 
 ```ruby
 session.actor.async(
-  :bootstrap,
-  answers: legacy.answers,
   idempotency_key: "legacy-assessment:#{legacy.id}",
   available_at: jittered_time
-)
+).bootstrap(answers: legacy.answers)
 ```
 
 Spread large backfills over a dispatch window and monitor mailbox age,

docs/roadmap.md

@@ -4,7 +4,7 @@
 
 - Rails engine, install generator, migration, and CLI
 - Explicit actor registry, references, JSON state, and state migrations
-- Direct synchronous RPC, explicit `sync`, and durable `async`
+- Fluent direct synchronous RPC, configured `sync`, and durable `async`
 - Durable message history plus ready/claimed membership tables
 - Concurrent sequence allocation and actor creation
 - Activation leases, renewal, unique activation tokens, generations, and

lib/solid_objects.rb

@@ -22,6 +22,7 @@
 require "solid_objects/actor_registry"
 require "solid_objects/state"
 require "solid_objects/actor_definition"
+require "solid_objects/operation_dispatcher"
 require "solid_objects/application_write_guard"
 require "solid_objects/actor"
 require "solid_objects/reference"

lib/solid_objects/actor.rb

@@ -196,8 +196,8 @@ def commit_action(name, **arguments)
       nil
     end
 
-    # @rbs (Symbol | String, at: Time, ?every: Numeric?, ?missed: Symbol | String, arguments: Hash[Symbol | String, untyped]) -> nil
-    def schedule(name, at:, every: nil, missed: :latest, arguments: {})
+    # @rbs (at: Time, ?every: Numeric?, ?missed: Symbol | String) -> OperationDispatcher
+    def schedule(at:, every: nil, missed: :latest)
       interval_seconds = every&.to_f
       if interval_seconds && !interval_seconds.positive?
         raise ArgumentError, "reminder interval must be positive"
@@ -207,22 +207,33 @@ def schedule(name, at:, every: nil, missed: :latest, arguments: {})
         raise ArgumentError, "missed reminder policy must be all or latest"
       end
 
-      ReminderIntent.new(
-        name: name.to_s,
-        at:,
-        arguments: Serialization.dump(arguments),
-        interval_seconds:,
-        missed_policy:
-      ).tap do |intent|
-        reminder_intents << intent
+      OperationDispatcher.new(
+        actor_type: self.class.actor_type,
+        handlers: self.class.definition.messages
+      ) do |message_name, arguments|
+        ReminderIntent.new(
+          name: message_name.to_s,
+          at:,
+          arguments: Serialization.dump(arguments),
+          interval_seconds:,
+          missed_policy:
+        ).tap do |intent|
+          reminder_intents << intent
+        end
+        nil
       end
-      nil
     end
 
-    # @rbs (Reference, Symbol | String, ?available_at: Time?, ?idempotency_key: String?, **untyped) -> nil
-    def send_to(reference, message_name, available_at: nil, idempotency_key: nil, **arguments)
-      stage_outbound_message(reference, message_name, arguments, available_at:, idempotency_key:)
-      nil
+    # @rbs (Reference, ?available_at: Time?, ?idempotency_key: String?) -> OperationDispatcher
+    def send_to(reference, available_at: nil, idempotency_key: nil)
+      actor_class = SolidObjects.registry.fetch(reference.actor_type)
+      OperationDispatcher.new(
+        actor_type: reference.actor_type,
+        handlers: actor_class.definition.messages
+      ) do |message_name, arguments|
+        stage_outbound_message(reference, message_name, arguments, available_at:, idempotency_key:)
+        nil
+      end
     end
 
     # @rbs (Symbol | String, Hash[String, untyped]) -> untyped