Apply M31 commands through the shared ledger

The assembly consumes its router queue and validates physical targets
before single or grouped output writes. Contiguous steady coils share
one write and a complete readback. Transport failures fail the batch,
while feedback mismatches fail only the affected members. Pulses avoid
transient feedback, and analog commands enforce exact voltage bounds.

The shared ledger stores outcomes before any feedback or result send.
Duplicate deliveries replay those outcomes without another physical
action, including after a publication failure or device reconnect.

Issue: 8fdc39a

diff --git a/doc/m31.rst b/doc/m31.rst
index c1a7af67bf0d78e249c75a78682fef6f07ac219e..69c1703f637f9d5f8be85027710d8e29b2400ef2 100644 (file)
--- a/doc/m31.rst
+++ b/doc/m31.rst
@@ -74,3 +74,51 @@ expose input counters. A configured poll interval is a target, not a
 guaranteed edge latency. Tests use fake clients and simulated time for
 baseline, transitions, pulse verification, reconnect, refresh, shutdown,
 and error propagation. Physical timing remains unverified.
 guaranteed edge latency. Tests use fake clients and simulated time for
 baseline, transitions, pulse verification, reconnect, refresh, shutdown,
 and error propagation. Physical timing remains unverified.

+
+*****************
+ Output commands
+*****************
+
+The shared router supplies one device FIFO for singles and batches. The
+driver checks ownership, output access, value type, and expiry before
+physical work. Steady outputs accept Boolean values. Pulse outputs
+accept only true and complete after the function 05 acknowledgement,
+without coil readback or transient reports. Analog outputs accept JSON
+numbers from zero through ten volts and round them to millivolts. Exact
+range checks precede floating-point conversion.
+
+A batch accepts steady digital outputs only. It sorts eligible targets
+by coil address, writes each contiguous range with function 15, and uses
+function 05 for an isolated target. Gaps remain unchanged. It reads the
+complete coil range once after all writes. A transport failure fails all
+pending members, including those in an earlier successful range. A
+mismatch fails only that member. Commands that expire before their write
+are rejected. Physical writes also have the earliest member deadline. If
+an expiry rejection cannot be sent after an earlier range is written,
+the driver reads back that range and records its outcomes before it
+returns the publication failure. A readback failure records failed
+outcomes, so replay cannot repeat the earlier physical writes.
+
+Changed feedback is sent before results. Each changed commanded point
+has its command ID as causation. Unrelated output changes have no
+command cause. Equal feedback needs no second report. The framework
+ledger records all outcomes before feedback or result publication, so a
+failed send cannot cause another physical action on replay. A repeated
+command reuses its first result ID, time, status, and reason. The ledger
+is memory-only and bounded to 256 records for the assembly. An eviction
+or process restart removes that duplicate protection.
+
+Commands received during device backoff are rejected if invalid or
+expired, and fail as unavailable otherwise. No command executes before
+the connection setup and baseline complete. Shutdown does not retry a
+physical action. Results use fixed failure reasons without transport
+addresses or untrusted response data. Drivers return errors and do not
+log them. Publication and independent cleanup failures stop the driver
+through the runner.
+
+Fake tests cover single writes, cross-bank batches, gaps, partial
+transport failure, mismatches, pulses, analog boundaries, expired work,
+cancellation, duplicate replay, and failed feedback or result sends.
+They do not verify electrical simultaneity, contact timing, hardware
+interlocks, or physical pulse width. Local motor protection remains
+necessary when network commands control moving equipment.

diff --git a/driver/m31/command_handler.go b/driver/m31/command_handler.go
new file mode 100644 (file)
index 0000000..0cba121
--- /dev/null
+++ b/driver/m31/command_handler.go
@@ -0,0 +1,347 @@
+package m31
+
+import (
+       "cmp"
+       "context"
+       "errors"
+       "fmt"
+       "slices"
+       "time"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+type commandHandler struct {
+       connection *connection
+}
+
+func (h *commandHandler) handle(ctx context.Context, commands []driver.Command) error {
+       pending, err := h.prepare(ctx, commands)
+       if err != nil {
+               return fmt.Errorf("prepare output commands: %w", err)
+       }
+
+       if len(pending) == 0 {
+               return nil
+       }
+
+       if len(commands) == 1 && (pending[0].analog || pending[0].point.pulse != 0) {
+               err = h.special(ctx, pending[0])
+               if err != nil {
+                       return fmt.Errorf("apply special output: %w", err)
+               }
+
+               return nil
+       }
+
+       slices.SortFunc(pending, func(first outputRequest, second outputRequest) int {
+               return cmp.Compare(first.point.address, second.point.address)
+       })
+
+       active := make([]outputRequest, 0, len(pending))
+
+       for start := 0; start < len(pending); {
+               if pending[start].command.Expired(time.Now()) {
+                       err = h.result(ctx, pending[start].command, protocol.ResultPayload{
+                               Status: "rejected",
+                               Reason: "Command expired",
+                       })
+                       if err != nil {
+                               finishErr := h.finish(ctx, active)
+
+                               var failure *deviceError
+
+                               if errors.As(finishErr, &failure) {
+                                       finishErr = fmt.Errorf("finish written commands: %w", failure.err)
+                               }
+
+                               return fmt.Errorf("reject expired write: %w", errors.Join(err, finishErr))
+                       }
+
+                       start++
+
+                       continue
+               }
+
+               end := start + 1
+
+               for end < len(pending) && pending[end].point.address == pending[end-1].point.address+1 && !pending[end].command.Expired(time.Now()) {
+                       end++
+               }
+
+               err = h.writeRange(ctx, pending[start:end])
+               if err != nil {
+                       return fmt.Errorf("fail coil writes: %w", h.fail(ctx, pending, err))
+               }
+
+               active = append(active, pending[start:end]...)
+               start = end
+       }
+
+       err = h.finish(ctx, active)
+       if err != nil {
+               return fmt.Errorf("finish coil commands: %w", err)
+       }
+
+       return nil
+}
+
+func (h *commandHandler) finish(ctx context.Context, active []outputRequest) error {
+       if len(active) == 0 {
+               return nil
+       }
+
+       next := h.connection.state
+
+       var err error
+
+       next.outputs, err = h.connection.owner.client.ReadCoils(ctx, uint16(len(h.connection.owner.topology.outputs)))
+       if err != nil {
+               return fmt.Errorf("fail coil readback: %w", h.fail(ctx, active, fmt.Errorf("read command coils: %w", err)))
+       }
+
+       if len(next.outputs) != len(h.connection.owner.topology.outputs) {
+               return fmt.Errorf("fail incomplete readback: %w", h.fail(ctx, active, fmt.Errorf("incomplete coil readback")))
+       }
+
+       next.observedAt = time.Now().UTC()
+
+       err = h.complete(ctx, active, next)
+       if err != nil {
+               return fmt.Errorf("complete coil commands: %w", err)
+       }
+
+       return nil
+}
+
+func (h *commandHandler) prepare(ctx context.Context, commands []driver.Command) ([]outputRequest, error) {
+       var pending []outputRequest
+
+       for _, command := range commands {
+               record, exists := h.connection.owner.ledger.Lookup(command.Metadata.MessageID)
+               if exists {
+                       err := h.send(ctx, command.Metadata.MessageID, record)
+                       if err != nil {
+                               return nil, fmt.Errorf("replay result: %w", err)
+                       }
+
+                       continue
+               }
+
+               request, reason := newOutputRequest(command, h.connection.owner.topology, h.connection.owner.deviceID, len(commands) > 1)
+               if reason != "" {
+                       err := h.result(ctx, command, protocol.ResultPayload{
+                               Status: "rejected",
+                               Reason: reason,
+                       })
+                       if err != nil {
+                               return nil, fmt.Errorf("reject command: %w", err)
+                       }
+
+                       continue
+               }
+
+               pending = append(pending, request)
+       }
+
+       return pending, nil
+}
+
+func (h *commandHandler) send(ctx context.Context, commandID string, record driver.ResultRecord) error {
+       err := h.connection.owner.publisher.Result(ctx, h.connection.owner.deviceID, commandID, record.Metadata, record.Payload)
+       if err != nil {
+               return fmt.Errorf("publish result: %w", err)
+       }
+
+       return nil
+}
+
+func (h *commandHandler) result(ctx context.Context, command driver.Command, payload protocol.ResultPayload) error {
+       record, err := h.connection.owner.ledger.Remember(command.Metadata.MessageID, payload)
+       if err != nil {
+               return fmt.Errorf("remember command result: %w", err)
+       }
+
+       err = h.send(ctx, command.Metadata.MessageID, record)
+       if err != nil {
+               return fmt.Errorf("send command result: %w", err)
+       }
+
+       return nil
+}
+
+func (h *commandHandler) writeRange(ctx context.Context, requests []outputRequest) error {
+       expires := *requests[0].command.Metadata.ExpiresAt
+       values := make([]bool, len(requests))
+
+       for index, request := range requests {
+               values[index] = request.value == 1
+               if request.command.Metadata.ExpiresAt.Before(expires) {
+                       expires = *request.command.Metadata.ExpiresAt
+               }
+       }
+
+       ctx, cancel := context.WithDeadline(ctx, expires)
+       defer cancel()
+
+       var err error
+
+       if len(requests) == 1 {
+               err = h.connection.owner.client.WriteCoil(ctx, requests[0].point.address, values[0])
+       } else {
+               err = h.connection.owner.client.WriteCoils(ctx, requests[0].point.address, values)
+       }
+
+       if err != nil {
+               return fmt.Errorf("write output coils: %w", err)
+       }
+
+       return nil
+}
+
+func (h *commandHandler) fail(ctx context.Context, requests []outputRequest, cause error) error {
+       records := make([]driver.ResultRecord, len(requests))
+
+       for index, request := range requests {
+               record, err := h.connection.owner.ledger.Remember(request.command.Metadata.MessageID, protocol.ResultPayload{
+                       Status: "failed",
+                       Reason: "Modbus operation failed",
+               })
+               if err != nil {
+                       return fmt.Errorf("remember failed commands: %w", errors.Join(cause, err))
+               }
+
+               records[index] = record
+       }
+
+       for index, request := range requests {
+               err := h.send(ctx, request.command.Metadata.MessageID, records[index])
+               if err != nil {
+                       return fmt.Errorf("send failed commands: %w", errors.Join(cause, err))
+               }
+       }
+
+       return &deviceError{err: cause}
+}
+
+func (h *commandHandler) complete(ctx context.Context, requests []outputRequest, next snapshot) error {
+       records := make([]driver.ResultRecord, len(requests))
+       causes := make(map[string]string, len(requests))
+
+       for index, request := range requests {
+               actual := uint16(0)
+               if request.analog {
+                       actual = next.analog[request.point.address]
+               } else if next.outputs[request.point.address] {
+                       actual = 1
+               }
+
+               payload := protocol.ResultPayload{Status: "completed"}
+               if actual != request.value {
+                       payload.Status, payload.Reason = "failed", "Output readback differs from command"
+               }
+
+               record, err := h.connection.owner.ledger.Remember(request.command.Metadata.MessageID, payload)
+               if err != nil {
+                       return fmt.Errorf("remember output results: %w", err)
+               }
+
+               records[index] = record
+               causes[request.point.id] = request.command.Metadata.MessageID
+       }
+
+       err := h.connection.publish(ctx, next, false, causes)
+       if err != nil {
+               return fmt.Errorf("publish output feedback: %w", err)
+       }
+
+       for index, request := range requests {
+               err = h.send(ctx, request.command.Metadata.MessageID, records[index])
+               if err != nil {
+                       return fmt.Errorf("publish completed results: %w", err)
+               }
+       }
+
+       return nil
+}
+
+func (h *commandHandler) special(ctx context.Context, request outputRequest) error {
+       next, err := h.applySpecial(ctx, request)
+       if err != nil {
+               return fmt.Errorf("fail special command: %w", h.fail(ctx, []outputRequest{request}, err))
+       }
+
+       if !request.analog {
+               err = h.result(ctx, request.command, protocol.ResultPayload{Status: "completed"})
+               if err != nil {
+                       return fmt.Errorf("complete pulse: %w", err)
+               }
+
+               return nil
+       }
+
+       err = h.complete(ctx, []outputRequest{request}, next)
+       if err != nil {
+               return fmt.Errorf("complete voltage command: %w", err)
+       }
+
+       return nil
+}
+
+func (h *commandHandler) applySpecial(ctx context.Context, request outputRequest) (snapshot, error) {
+       ctx, cancel := context.WithDeadline(ctx, *request.command.Metadata.ExpiresAt)
+       defer cancel()
+
+       if !request.analog {
+               err := h.connection.owner.client.WriteCoil(ctx, request.point.address, true)
+               if err != nil {
+                       return snapshot{}, fmt.Errorf("write pulse: %w", err)
+               }
+
+               return h.connection.state, nil
+       }
+
+       err := h.connection.owner.client.WriteRegister(ctx, request.point.address, request.value)
+       if err != nil {
+               return snapshot{}, fmt.Errorf("write voltage: %w", err)
+       }
+
+       next := h.connection.state
+       next.analog, err = h.connection.owner.client.ReadRegisters(ctx, 0, uint16(len(h.connection.owner.topology.analog)))
+       if err != nil {
+               return snapshot{}, fmt.Errorf("read voltage feedback: %w", err)
+       }
+
+       if len(next.analog) != len(h.connection.owner.topology.analog) {
+               return snapshot{}, fmt.Errorf("incomplete voltage feedback")
+       }
+
+       err = validateVoltage(next.analog)
+       if err != nil {
+               return snapshot{}, fmt.Errorf("validate voltage feedback: %w", err)
+       }
+
+       next.observedAt = time.Now().UTC()
+
+       return next, nil
+}
+
+func (h *commandHandler) unavailable(ctx context.Context, commands []driver.Command) error {
+       pending, err := h.prepare(ctx, commands)
+       if err != nil {
+               return fmt.Errorf("prepare unavailable commands: %w", err)
+       }
+
+       for _, request := range pending {
+               err = h.result(ctx, request.command, protocol.ResultPayload{
+                       Status: "failed",
+                       Reason: "Modbus device is unavailable",
+               })
+               if err != nil {
+                       return fmt.Errorf("publish unavailable result: %w", err)
+               }
+       }
+
+       return nil
+}

diff --git a/driver/m31/command_handler_test.go b/driver/m31/command_handler_test.go
new file mode 100644 (file)
index 0000000..0d796f5
--- /dev/null
+++ b/driver/m31/command_handler_test.go
@@ -0,0 +1,420 @@
+package m31
+
+import (
+       "context"
+       "errors"
+       "fmt"
+       "slices"
+       "testing"
+       "testing/synctest"
+       "time"
+
+       "github.com/stretchr/testify/require"
+
+       "onehouse/driver"
+       "onehouse/protocol"
+)
+
+func TestCommandsWriteContiguousRangesAndReplayResults(t *testing.T) {
+       c, client, transport := testConnection(t)
+       h := &commandHandler{connection: c}
+
+       var commands []driver.Command
+
+       for address := 5; address < 18; address++ {
+               commands = append(commands, testCommand(t, c.owner.topology.outputs[address].id, "true"))
+       }
+
+       slices.Reverse(commands)
+       require.NoError(t, h.handle(t.Context(), commands))
+       require.Len(t, client.writes, 1)
+       require.EqualValues(t, 5, client.writes[0].address)
+       require.True(t, client.writes[0].multiple)
+       require.Len(t, client.writes[0].values, 13)
+       require.Equal(t, 1, client.coilReads)
+
+       reports := transport.ofKind(protocol.Reported)
+       results := transport.ofKind(protocol.Result)
+
+       require.Len(t, reports, 13)
+       require.Len(t, results, 13)
+
+       for _, command := range commands {
+               for _, report := range reports {
+                       if report.Address.Target == command.Capability {
+                               require.Equal(t, command.Metadata.MessageID, report.Metadata.CausationID)
+                       }
+               }
+       }
+
+       require.NoError(t, h.handle(t.Context(), commands))
+       require.Len(t, client.writes, 1)
+       require.Equal(t, 1, client.coilReads)
+
+       replayed := transport.ofKind(protocol.Result)[13:]
+       byCommand := make(map[string]protocol.Metadata)
+
+       for _, result := range results {
+               byCommand[result.Address.Target] = result.Metadata
+       }
+
+       for _, result := range replayed {
+               require.Equal(t, byCommand[result.Address.Target], result.Metadata)
+       }
+}
+
+func TestCommandsKeepGapsAndRejectNonsteadyMembers(t *testing.T) {
+       c, client, transport := testConnection(t)
+       h := &commandHandler{connection: c}
+       commands := []driver.Command{
+               testCommand(t, "m0.do0", "true"),
+               testCommand(t, "m0.do1", "true"),
+               testCommand(t, "m0.do2", "true"),
+               testCommand(t, "m0.do4", "true"),
+               testCommand(t, "m2.ao0", "2"),
+               testCommand(t, "m0.di0", "true"),
+       }
+       require.NoError(t, h.handle(t.Context(), commands))
+       require.Len(t, client.writes, 2)
+       require.EqualValues(t, 0, client.writes[0].address)
+       require.Len(t, client.writes[0].values, 2)
+       require.EqualValues(t, 4, client.writes[1].address)
+       require.False(t, client.writes[1].multiple)
+       require.False(t, client.coils[2])
+       require.False(t, client.coils[3])
+       require.Len(t, transport.ofKind(protocol.Result), 6)
+}
+
+func TestCommandsFailAllOnTransportErrorAndOnlyMismatchOtherwise(t *testing.T) {
+       testCases := []struct {
+               name             string
+               change           func(*fakeClient)
+               transportFailure bool
+       }{
+               {name: "fails all on write failure", transportFailure: true, change: func(c *fakeClient) {
+                       c.writeErr = errors.New("write failed")
+               }},
+               {name: "fails all after a partial write", transportFailure: true, change: func(c *fakeClient) {
+                       c.writeErr = errors.New("later write failed")
+                       c.failAfterWrites = 1
+               }},
+               {name: "fails all on read failure", transportFailure: true, change: func(c *fakeClient) {
+                       c.readErr = errors.New("read failed")
+               }},
+               {name: "fails all on short read", transportFailure: true, change: func(c *fakeClient) {
+                       c.shortRead = true
+               }},
+               {name: "fails only mismatching member", change: func(c *fakeClient) {
+                       c.mismatch = 1
+               }},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       c, client, transport := testConnection(t)
+                       tc.change(client)
+
+                       h := &commandHandler{connection: c}
+                       commands := []driver.Command{testCommand(t, "m0.do0", "true"), testCommand(t, "m0.do1", "true"), testCommand(t, "m1.do7", "true")}
+                       err := h.handle(t.Context(), commands)
+                       if tc.transportFailure {
+                               var failure *deviceError
+
+                               require.ErrorAs(t, err, &failure)
+                       } else {
+                               require.NoError(t, err)
+                       }
+
+                       require.Len(t, transport.ofKind(protocol.Result), 3)
+
+                       for index, command := range commands {
+                               record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
+                               require.True(t, exists)
+
+                               status := "completed"
+                               if tc.transportFailure || index == 1 {
+                                       status = "failed"
+                               }
+
+                               require.Equal(t, status, record.Payload.Status)
+                       }
+               })
+       }
+}
+
+func TestCommandsCompletePulseWithoutReadbackOrFeedback(t *testing.T) {
+       c, client, transport := testConnection(t)
+       h := &commandHandler{connection: c}
+       command := testCommand(t, "m0.do2", "true")
+       require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+       require.Len(t, client.writes, 1)
+       require.Zero(t, client.coilReads)
+       require.Empty(t, transport.ofKind(protocol.Reported))
+       require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+       require.Len(t, client.writes, 1)
+
+       command = testCommand(t, "m0.do2", "false")
+       require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+       require.Len(t, client.writes, 1)
+
+       record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
+       require.True(t, exists)
+       require.Equal(t, "rejected", record.Payload.Status)
+}
+
+func TestCommandsWriteVoltageAndCheckFeedback(t *testing.T) {
+       testCases := []struct {
+               name       string
+               change     func(*fakeClient)
+               wantStatus string
+               wantError  bool
+       }{
+               {name: "completes verified voltage", wantStatus: "completed", change: func(*fakeClient) {
+               }},
+               {name: "fails mismatching voltage", wantStatus: "failed", change: func(c *fakeClient) {
+                       c.mismatch = 3
+               }},
+               {name: "fails register transport", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
+                       c.registerErr = errors.New("register failed")
+               }},
+               {name: "fails analog read transport", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
+                       c.registerReadErr = errors.New("read failed")
+               }},
+               {name: "fails unsafe feedback", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
+                       c.analog[0] = 10001
+               }},
+               {name: "fails incomplete feedback", wantStatus: "failed", wantError: true, change: func(c *fakeClient) {
+                       c.analog = c.analog[:4]
+               }},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       c, client, transport := testConnection(t)
+                       tc.change(client)
+
+                       h := &commandHandler{connection: c}
+                       command := testCommand(t, "m2.ao3", "5.5")
+                       err := h.handle(t.Context(), []driver.Command{command})
+                       if tc.wantError {
+                               require.Error(t, err)
+                       } else {
+                               require.NoError(t, err)
+                       }
+
+                       record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
+                       require.True(t, exists)
+                       require.Equal(t, tc.wantStatus, record.Payload.Status)
+
+                       if tc.wantStatus == "completed" {
+                               require.EqualValues(t, 5500, client.analog[3])
+                               require.Len(t, transport.messages, 2)
+                               require.Equal(t, protocol.Reported, transport.messages[0].Address.Kind)
+                               require.Equal(t, command.Metadata.MessageID, transport.messages[0].Metadata.CausationID)
+                       }
+               })
+       }
+}
+
+func TestCommandsRememberBeforeFailedPublication(t *testing.T) {
+       for _, kind := range []protocol.Kind{protocol.Reported, protocol.Result} {
+               t.Run(fmt.Sprint("preserves first outcome after ", kind, " failure"), func(t *testing.T) {
+                       c, client, transport := testConnection(t)
+                       h := &commandHandler{connection: c}
+                       failure := errors.New("publication failed")
+                       transport.failKind, transport.err = kind, failure
+                       commands := []driver.Command{testCommand(t, "m0.do0", "true"), testCommand(t, "m0.do1", "true")}
+                       require.ErrorIs(t, h.handle(t.Context(), commands), failure)
+
+                       record, exists := c.owner.ledger.Lookup(commands[1].Metadata.MessageID)
+                       require.True(t, exists)
+                       require.Equal(t, "completed", record.Payload.Status)
+
+                       transport.err = nil
+
+                       require.NoError(t, h.handle(t.Context(), commands))
+                       require.Len(t, client.writes, 1)
+               })
+       }
+}
+
+func TestCommandsRecheckExpiryBeforeEachWriteAndHonorCancellation(t *testing.T) {
+       synctest.Test(t, func(t *testing.T) {
+               c, client, _ := testConnection(t)
+               h := &commandHandler{connection: c}
+               first := testCommand(t, "m0.do0", "true")
+               second := testCommand(t, "m0.do4", "true")
+               expires := time.Now().Add(time.Millisecond)
+               second.Metadata.ExpiresAt = &expires
+               client.writeDelay = 2 * time.Millisecond
+
+               require.NoError(t, h.handle(t.Context(), []driver.Command{first, second}))
+               require.Len(t, client.writes, 1)
+
+               record, exists := c.owner.ledger.Lookup(second.Metadata.MessageID)
+               require.True(t, exists)
+               require.Equal(t, "rejected", record.Payload.Status)
+
+               ctx, cancel := context.WithCancel(t.Context())
+               cancel()
+
+               third := testCommand(t, "m0.do1", "true")
+               require.ErrorIs(t, h.handle(ctx, []driver.Command{third}), context.Canceled)
+               require.Len(t, client.writes, 1)
+       })
+}
+
+func TestCommandsKeepWrittenResultsAfterExpiryPublicationFails(t *testing.T) {
+       testCases := []struct {
+               name        string
+               mismatch    bool
+               readFailure bool
+               failOnce    bool
+               status      string
+       }{
+               {
+                       name:   "keeps completed write",
+                       status: "completed",
+               },
+               {
+                       name:     "keeps mismatching write",
+                       mismatch: true,
+                       status:   "failed",
+               },
+               {
+                       name:        "keeps unverified write",
+                       readFailure: true,
+                       status:      "failed",
+               },
+               {
+                       name:        "preserves rejection failure after readback failure",
+                       readFailure: true,
+                       failOnce:    true,
+                       status:      "failed",
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       synctest.Test(t, func(t *testing.T) {
+                               c, client, transport := testConnection(t)
+                               h := &commandHandler{connection: c}
+                               first := testCommand(t, "m0.do0", "true")
+                               second := testCommand(t, "m0.do4", "true")
+                               expires := time.Now().Add(time.Millisecond)
+                               second.Metadata.ExpiresAt = &expires
+                               client.writeDelay = 2 * time.Millisecond
+                               sendErr := errors.New("rejection publication failed")
+                               transport.failOnce = tc.failOnce
+                               transport.failKind, transport.err = protocol.Result, sendErr
+                               if tc.mismatch {
+                                       client.mismatch = 0
+                               }
+
+                               if tc.readFailure {
+                                       client.readErr = errors.New("readback failed")
+                               }
+
+                               commands := []driver.Command{first, second}
+                               err := h.handle(t.Context(), commands)
+                               require.ErrorIs(t, err, sendErr)
+
+                               var failure *deviceError
+
+                               require.NotErrorAs(t, err, &failure)
+                               require.Len(t, client.writes, 1)
+
+                               record, exists := c.owner.ledger.Lookup(first.Metadata.MessageID)
+                               require.True(t, exists)
+                               require.Equal(t, tc.status, record.Payload.Status)
+
+                               rejected, exists := c.owner.ledger.Lookup(second.Metadata.MessageID)
+                               require.True(t, exists)
+                               require.Equal(t, "rejected", rejected.Payload.Status)
+
+                               transport.err, client.readErr = nil, nil
+
+                               require.NoError(t, h.handle(t.Context(), commands))
+                               require.Len(t, client.writes, 1)
+
+                               results := transport.ofKind(protocol.Result)
+                               require.Equal(t, record.Metadata, results[len(results)-2].Metadata)
+                               require.Equal(t, record.Payload, results[len(results)-2].Payload)
+                               require.Equal(t, rejected.Metadata, results[len(results)-1].Metadata)
+                       })
+               })
+       }
+}
+
+func TestCommandsCompleteUnchangedSingleWithoutState(t *testing.T) {
+       c, client, transport := testConnection(t)
+       h := &commandHandler{connection: c}
+       command := testCommand(t, "m0.do0", "false")
+       require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+       require.Len(t, client.writes, 1)
+       require.False(t, client.writes[0].multiple)
+       require.Len(t, transport.messages, 1)
+       require.Equal(t, protocol.Result, transport.messages[0].Address.Kind)
+}
+
+func TestCommandsReturnPublicationErrorsWithoutLosingOutcomes(t *testing.T) {
+       testCases := []struct {
+               name            string
+               capability      string
+               value           string
+               offline         bool
+               physicalFailure bool
+       }{
+               {name: "retains completed pulse after result failure", capability: "m0.do2", value: "true"},
+               {name: "retains voltage after result failure", capability: "m2.ao0", value: "5"},
+               {name: "returns rejection failure", capability: "m0.di0", value: "true"},
+               {name: "returns offline result failure", capability: "m0.do0", value: "true", offline: true},
+               {name: "returns offline rejection failure", capability: "m0.di0", value: "true", offline: true},
+               {name: "joins pulse and result failure", capability: "m0.do2", value: "true", physicalFailure: true},
+               {name: "joins batch and result failure", capability: "m0.do0", value: "true", physicalFailure: true},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       c, client, transport := testConnection(t)
+                       h := &commandHandler{connection: c}
+                       command := testCommand(t, tc.capability, tc.value)
+                       sendErr := errors.New("send failed")
+                       physicalErr := errors.New("physical failure")
+                       transport.failKind, transport.err = protocol.Result, sendErr
+                       if tc.physicalFailure {
+                               client.writeErr = physicalErr
+                       }
+
+                       var err error
+
+                       if tc.offline {
+                               err = h.unavailable(t.Context(), []driver.Command{command})
+                       } else {
+                               err = h.handle(t.Context(), []driver.Command{command})
+                       }
+
+                       require.ErrorIs(t, err, sendErr)
+
+                       if tc.physicalFailure {
+                               require.ErrorIs(t, err, physicalErr)
+
+                               var physical *deviceError
+
+                               require.NotErrorAs(t, err, &physical)
+                       }
+
+                       record, exists := c.owner.ledger.Lookup(command.Metadata.MessageID)
+                       require.True(t, exists)
+
+                       transport.err, client.writeErr = nil, nil
+                       count := len(client.writes)
+
+                       require.NoError(t, h.handle(t.Context(), []driver.Command{command}))
+                       require.Len(t, client.writes, count)
+
+                       replay := transport.ofKind(protocol.Result)
+                       require.Equal(t, record.Metadata, replay[len(replay)-1].Metadata)
+               })
+       }
+}

diff --git a/driver/m31/connection_test.go b/driver/m31/connection_test.go
index f2e67a6c22e10e333a4ae8b841ac27593adf460b..ede502b97f1baa0e6a35bc801fb1168c1bd43c36 100644 (file)
--- a/driver/m31/connection_test.go
+++ b/driver/m31/connection_test.go
@@ -58,6 +58,12 @@ func TestConnectionRejectsConfigurationAndReadFailuresBeforeOnline(t *testing.T)
                {name: "rejects pulse register write failure", change: func(c *fakeClient) {
                        c.registerErr = errors.New("configuration failed")
                }},
                {name: "rejects pulse register write failure", change: func(c *fakeClient) {
                        c.registerErr = errors.New("configuration failed")
                }},

+               {name: "rejects configuration read failure", change: func(c *fakeClient) {
+                       c.registerReadErr = errors.New("configuration read failed")
+               }},
+               {name: "rejects pulse width mismatch", change: func(c *fakeClient) {
+                       c.wrongRegister = pulseWidthBase + 2
+               }},

                {name: "rejects configuration mismatch", change: func(c *fakeClient) {
                        c.wrongConfig = true
                }},
                {name: "rejects configuration mismatch", change: func(c *fakeClient) {
                        c.wrongConfig = true
                }},


@@ -77,7 +83,7 @@ func TestConnectionRejectsConfigurationAndReadFailuresBeforeOnline(t *testing.T)
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {

-                       d, client, transport := testAssembly(t)
+                       d, client, transport, _ := testAssembly(t)

                        tc.change(client)
 
                        c := &connection{owner: d}
                        tc.change(client)
 
                        c := &connection{owner: d}

diff --git a/driver/m31/driver.go b/driver/m31/driver.go
index 9201a59d6e9d7b5e9db5d17bd14f5e96092b3bbd..8849c824837613e50b0c27a4ffac20f2f3b7f873 100644 (file)
--- a/driver/m31/driver.go
+++ b/driver/m31/driver.go
@@ -12,11 +12,13 @@ import (
        "onehouse/protocol"
 )
 
        "onehouse/protocol"
 )
 

-// Driver polls one assembly. Run has one owner and closes the supplied Client
-// on each attempt. Recoverable device failures publish offline and retry with
-// backoff. Publication or cleanup failures return to the framework runner. The
-// shared publisher must run alongside this driver. Refresh is safe during Run.
-// No operation starts a physical connection before Run.
+// Driver serializes one assembly's polling and commands. Run has one owner and
+// closes the supplied Client on each attempt. Recoverable device failures
+// publish offline and retry with backoff. Publication or cleanup failures
+// return to the framework runner. The shared publisher must run alongside this
+// driver. Refresh is safe during Run. Duplicate protection ends at ledger
+// eviction or process restart. No operation starts a physical connection before
+// Run.

 type Driver struct {
        deviceID     string
        address      string
 type Driver struct {
        deviceID     string
        address      string


@@ -25,11 +27,14 @@ type Driver struct {
        topology     topology
        publisher    *driver.Publisher
        client       Client
        topology     topology
        publisher    *driver.Publisher
        client       Client

+       queue        <-chan []driver.Command
+       ledger       *driver.Ledger

 }
 
 }
 

-// New validates settings. The caller supplies valid publisher and client
-// dependencies. The driver owns the client until Run returns.
-func New(settings Settings, publisher *driver.Publisher, client Client) (*Driver, error) {
+// New validates settings and requires the device's queue in router. The caller
+// supplies valid publisher, router, and client dependencies and starts the
+// router's intake separately. The driver owns the client until Run returns.
+func New(settings Settings, publisher *driver.Publisher, router *driver.Router, client Client) (*Driver, error) {

        if !protocol.ValidID(settings.DeviceID) {
                return nil, fmt.Errorf("device ID must be a canonical ULID")
        }
        if !protocol.ValidID(settings.DeviceID) {
                return nil, fmt.Errorf("device ID must be a canonical ULID")
        }


@@ -57,6 +62,16 @@ func New(settings Settings, publisher *driver.Publisher, client Client) (*Driver
                return nil, fmt.Errorf("validate topology: %w", err)
        }
 
                return nil, fmt.Errorf("validate topology: %w", err)
        }
 

+       queue, exists := router.Queue(settings.DeviceID)
+       if !exists {
+               return nil, fmt.Errorf("device has no command queue")
+       }
+
+       ledger, err := driver.NewLedger(settings.DeviceID)
+       if err != nil {
+               return nil, fmt.Errorf("create result ledger: %w", err)
+       }
+

        return &Driver{
                deviceID:     settings.DeviceID,
                address:      settings.Address,
        return &Driver{
                deviceID:     settings.DeviceID,
                address:      settings.Address,


@@ -65,13 +80,15 @@ func New(settings Settings, publisher *driver.Publisher, client Client) (*Driver
                topology:     points,
                publisher:    publisher,
                client:       client,
                topology:     points,
                publisher:    publisher,
                client:       client,

+               queue:        queue,
+               ledger:       ledger,

        }, nil
 }
 
        }, nil
 }
 

-// Run configures and verifies output modes before online or polling. A
+// Run configures and verifies output modes before online or command work. A

 // successful first read establishes a baseline without input events. Device
 // failures retry after one second, doubling to thirty seconds until a complete
 // successful first read establishes a baseline without input events. Device
 // failures retry after one second, doubling to thirty seconds until a complete

-// baseline succeeds.
+// baseline succeeds. Commands received during backoff fail without I/O.

 func (d *Driver) Run(ctx context.Context) error {
        err := d.publisher.Descriptor(ctx, d.deviceID, protocol.Metadata{}, d.topology.descriptor())
        if err != nil {
 func (d *Driver) Run(ctx context.Context) error {
        err := d.publisher.Descriptor(ctx, d.deviceID, protocol.Metadata{}, d.topology.descriptor())
        if err != nil {


@@ -135,6 +152,7 @@ func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err err
        }
 
        healthy = true
        }
 
        healthy = true

+       handler := &commandHandler{connection: connection}

        ticker := time.NewTicker(d.pollInterval)
        defer ticker.Stop()
 
        ticker := time.NewTicker(d.pollInterval)
        defer ticker.Stop()
 


@@ -142,6 +160,11 @@ func (d *Driver) attempt(ctx context.Context) (healthy bool, retry bool, err err
                select {
                case <-ctx.Done():
                        return healthy, false, ctx.Err()
                select {
                case <-ctx.Done():
                        return healthy, false, ctx.Err()

+               case commands := <-d.queue:
+                       err = handler.handle(ctx, commands)
+                       if err != nil {
+                               return healthy, false, fmt.Errorf("process commands: %w", err)
+                       }

                case <-ticker.C:
                        err = connection.poll(ctx)
                        if err != nil {
                case <-ticker.C:
                        err = connection.poll(ctx)
                        if err != nil {


@@ -155,10 +178,19 @@ func (d *Driver) wait(ctx context.Context, delay time.Duration) error {
        timer := time.NewTimer(delay)
        defer timer.Stop()
 
        timer := time.NewTimer(delay)
        defer timer.Stop()
 

-       select {
-       case <-ctx.Done():
-               return ctx.Err()
-       case <-timer.C:
-               return nil
+       handler := &commandHandler{connection: &connection{owner: d}}
+
+       for {
+               select {
+               case <-ctx.Done():
+                       return ctx.Err()
+               case <-timer.C:
+                       return nil
+               case commands := <-d.queue:
+                       err := handler.unavailable(ctx, commands)
+                       if err != nil {
+                               return fmt.Errorf("finish unavailable commands: %w", err)
+                       }
+               }

        }
 }
        }
 }

diff --git a/driver/m31/driver_test.go b/driver/m31/driver_test.go
index 1ba61909a48c5cf8b5b8f7b54ea4750565f04158..7081f929e9077bfcf1e1c93fa171fca8be9f1223 100644 (file)
--- a/driver/m31/driver_test.go
+++ b/driver/m31/driver_test.go
@@ -9,6 +9,7 @@ import (
 
        "github.com/stretchr/testify/require"
 
 
        "github.com/stretchr/testify/require"
 

+       "onehouse/driver"

        "onehouse/protocol"
 )
 
        "onehouse/protocol"
 )
 


@@ -54,7 +55,7 @@ func TestNewRejectsInvalidAssemblySettings(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {

-                       d, client, _ := testAssembly(t)
+                       d, client, _, router := testAssembly(t)

                        settings := Settings{
                                DeviceID:     testDeviceID,
                                Address:      "example.com:502",
                        settings := Settings{
                                DeviceID:     testDeviceID,
                                Address:      "example.com:502",


@@ -64,15 +65,28 @@ func TestNewRejectsInvalidAssemblySettings(t *testing.T) {
                        }
                        tc.change(&settings)
 
                        }
                        tc.change(&settings)
 

-                       _, err := New(settings, d.publisher, client)
+                       _, err := New(settings, d.publisher, router, client)

                        require.Error(t, err)
                })
        }
                        require.Error(t, err)
                })
        }

+
+       d, client, _, _ := testAssembly(t)
+       router, err := driver.NewRouter(nil)
+       require.NoError(t, err)
+
+       _, err = New(Settings{
+               DeviceID:     testDeviceID,
+               Address:      "example.com:502",
+               Unit:         1,
+               PollInterval: 50 * time.Millisecond,
+               Modules:      []Module{{Model: HostModel}},
+       }, d.publisher, router, client)
+       require.ErrorContains(t, err, "command queue")

 }
 
 func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {
 }
 
 func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {

-               d, client, transport := testAssembly(t)
+               d, client, transport, router := testAssembly(t)

                client.connectErrors = []error{errors.New("unavailable")}
                client.inputs[1] = true
                ctx, cancel := context.WithCancel(t.Context())
                client.connectErrors = []error{errors.New("unavailable")}
                client.inputs[1] = true
                ctx, cancel := context.WithCancel(t.Context())


@@ -91,6 +105,16 @@ func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
                })
                require.Empty(t, transport.ofKind(protocol.Event))
 
                })
                require.Empty(t, transport.ofKind(protocol.Event))
 

+               command := testCommand(t, "m0.do0", "true")
+               queueCommands(t, router, []driver.Command{command})
+               synctest.Wait()
+
+               record, exists := d.ledger.Lookup(command.Metadata.MessageID)
+               require.True(t, exists)
+               require.Equal(t, "failed", record.Payload.Status)
+               client.withLock(func(client *fakeClient) {
+                       require.Empty(t, client.writes)
+               })

                time.Sleep(time.Second)
                synctest.Wait()
                client.withLock(func(client *fakeClient) {
                time.Sleep(time.Second)
                synctest.Wait()
                client.withLock(func(client *fakeClient) {


@@ -132,6 +156,18 @@ func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
                        require.Len(t, client.registerWrites, 66)
                })
                require.Len(t, transport.ofKind(protocol.Event), 1)
                        require.Len(t, client.registerWrites, 66)
                })
                require.Len(t, transport.ofKind(protocol.Event), 1)

+               queueCommands(t, router, []driver.Command{command})
+               synctest.Wait()
+               client.withLock(func(client *fakeClient) {
+                       require.Empty(t, client.writes)
+               })
+
+               command = testCommand(t, "m1.do15", "true")
+               queueCommands(t, router, []driver.Command{command})
+               synctest.Wait()
+               client.withLock(func(client *fakeClient) {
+                       require.Len(t, client.writes, 1)
+               })

                cancel()
                require.ErrorIs(t, <-done, context.Canceled)
                client.withLock(func(client *fakeClient) {
                cancel()
                require.ErrorIs(t, <-done, context.Canceled)
                client.withLock(func(client *fakeClient) {


@@ -142,7 +178,7 @@ func TestDriverReconnectsWithConfigurationAndFreshBaseline(t *testing.T) {
 
 func TestDriverBoundsReconnectBackoff(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {
 
 func TestDriverBoundsReconnectBackoff(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {

-               d, client, _ := testAssembly(t)
+               d, client, _, _ := testAssembly(t)

                for range 8 {
                        client.withLock(func(client *fakeClient) {
                                client.connectErrors = append(client.connectErrors, errors.New("offline"))
                for range 8 {
                        client.withLock(func(client *fakeClient) {
                                client.connectErrors = append(client.connectErrors, errors.New("offline"))


@@ -188,7 +224,7 @@ func TestDriverReturnsPublicationAndCleanupFailures(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
 
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {

-                       d, client, transport := testAssembly(t)
+                       d, client, transport, _ := testAssembly(t)

                        failure := errors.New("independent failure")
                        if tc.closeFailure {
                                client.closeErr = failure
                        failure := errors.New("independent failure")
                        if tc.closeFailure {
                                client.closeErr = failure


@@ -205,7 +241,7 @@ func TestDriverReturnsPublicationAndCleanupFailures(t *testing.T) {
 
 func TestDriverRefreshDoesNotRepeatCommandsOrEvents(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {
 
 func TestDriverRefreshDoesNotRepeatCommandsOrEvents(t *testing.T) {
        synctest.Test(t, func(t *testing.T) {

-               d, client, transport := testAssembly(t)
+               d, client, transport, _ := testAssembly(t)

                ctx, cancel := context.WithCancel(t.Context())
                defer cancel()
 
                ctx, cancel := context.WithCancel(t.Context())
                defer cancel()
 

diff --git a/driver/m31/fake_helper_test.go b/driver/m31/fake_helper_test.go
index c53dbac7b0a624b3c1017a7eac3a9a0b72fec95b..6b5a1d4d0129cf988d08d1b31730fd2645c1f97f 100644 (file)
--- a/driver/m31/fake_helper_test.go
+++ b/driver/m31/fake_helper_test.go
@@ -18,27 +18,30 @@ import (
 const testDeviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
 
 type fakeClient struct {
 const testDeviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
 
 type fakeClient struct {

-       mu             sync.Mutex
-       inputs         []bool
-       coils          []bool
-       analog         []uint16
-       registers      map[uint16]uint16
-       writes         []coilWrite
-       registerWrites []uint16
-       attempts       []time.Time
-       closes         int
-       coilReads      int
-       connectErrors  []error
-       inputErr       error
-       readErr        error
-       writeErr       error
-       registerErr    error
-       closeErr       error
-       wrongConfig    bool
-       shortRead      bool
-       mismatch       int
-       writeDelay     time.Duration
-       inputDelay     time.Duration
+       registerReadErr error
+       wrongRegister   uint16
+       failAfterWrites int
+       mu              sync.Mutex
+       inputs          []bool
+       coils           []bool
+       analog          []uint16
+       registers       map[uint16]uint16
+       writes          []coilWrite
+       registerWrites  []uint16
+       attempts        []time.Time
+       closes          int
+       coilReads       int
+       connectErrors   []error
+       inputErr        error
+       readErr         error
+       writeErr        error
+       registerErr     error
+       closeErr        error
+       wrongConfig     bool
+       shortRead       bool
+       mismatch        int
+       writeDelay      time.Duration
+       inputDelay      time.Duration

 }
 
 func (c *fakeClient) Connect(ctx context.Context, _ string, _ byte) error {
 }
 
 func (c *fakeClient) Connect(ctx context.Context, _ string, _ byte) error {


@@ -98,13 +101,13 @@ func (c *fakeClient) ReadRegisters(ctx context.Context, address uint16, count ui
        c.mu.Lock()
        defer c.mu.Unlock()
 
        c.mu.Lock()
        defer c.mu.Unlock()
 

-       if c.registerErr != nil || ctx.Err() != nil {
-               return nil, errors.Join(c.registerErr, ctx.Err())
+       if c.registerErr != nil || c.registerReadErr != nil || ctx.Err() != nil {
+               return nil, errors.Join(c.registerErr, c.registerReadErr, ctx.Err())

        }
 
        if address >= pulseWidthBase {
                value := c.registers[address]
        }
 
        if address >= pulseWidthBase {
                value := c.registers[address]

-               if c.wrongConfig {
+               if c.wrongConfig || address == c.wrongRegister {

                        value++
                }
 
                        value++
                }
 


@@ -128,7 +131,7 @@ func (c *fakeClient) write(ctx context.Context, address uint16, values []bool, m
 
        time.Sleep(c.writeDelay)
 
 
        time.Sleep(c.writeDelay)
 

-       if c.writeErr != nil || ctx.Err() != nil {
+       if c.writeErr != nil && len(c.writes) >= c.failAfterWrites || ctx.Err() != nil {

                return errors.Join(c.writeErr, ctx.Err())
        }
 
                return errors.Join(c.writeErr, ctx.Err())
        }
 


@@ -182,6 +185,7 @@ type fakeTransport struct {
        mu       sync.Mutex
        messages []protocol.Message[any]
        failKind protocol.Kind
        mu       sync.Mutex
        messages []protocol.Message[any]
        failKind protocol.Kind

+       failOnce bool

        err      error
 }
 
        err      error
 }
 


@@ -195,7 +199,12 @@ func (f *fakeTransport) Publish(ctx context.Context, publication protocol.Public
        }
 
        if message.Address.Kind == f.failKind && f.err != nil {
        }
 
        if message.Address.Kind == f.failKind && f.err != nil {

-               return f.err
+               err = f.err
+               if f.failOnce {
+                       f.err = nil
+               }
+
+               return err

        }
 
        if ctx.Err() != nil {
        }
 
        if ctx.Err() != nil {


@@ -229,11 +238,14 @@ func (f *fakeTransport) all() []protocol.Message[any] {
        return slices.Clone(f.messages)
 }
 
        return slices.Clone(f.messages)
 }
 

-func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport) {
+func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport, *driver.Router) {

        t.Helper()
 
        transport := &fakeTransport{}
        publisher := driver.NewPublisher(transport)
        t.Helper()
 
        transport := &fakeTransport{}
        publisher := driver.NewPublisher(transport)

+       router, err := driver.NewRouter([]string{testDeviceID})
+       require.NoError(t, err)
+

        client := &fakeClient{
                inputs:    make([]bool, 16),
                coils:     make([]bool, 32),
        client := &fakeClient{
                inputs:    make([]bool, 16),
                coils:     make([]bool, 32),


@@ -253,16 +265,35 @@ func testAssembly(t *testing.T) (*Driver, *fakeClient, *fakeTransport) {
                        {Model: MixedModel},
                },
        }
                        {Model: MixedModel},
                },
        }

-       d, err := New(settings, publisher, client)
+       d, err := New(settings, publisher, router, client)
+       require.NoError(t, err)
+
+       return d, client, transport, router
+}
+
+func testCommand(t *testing.T, capability string, value string) driver.Command {
+       t.Helper()
+
+       id, err := protocol.NewID(time.Now())

        require.NoError(t, err)
 
        require.NoError(t, err)
 

-       return d, client, transport
+       expires := time.Now().UTC().Add(time.Minute)
+
+       return driver.Command{
+               DeviceID:   testDeviceID,
+               Capability: capability,
+               Metadata: protocol.Metadata{
+                       MessageID: id,
+                       ExpiresAt: &expires,
+               },
+               Payload: protocol.ValuePayload{Value: []byte(value)},
+       }

 }
 
 func testConnection(t *testing.T) (*connection, *fakeClient, *fakeTransport) {
        t.Helper()
 
 }
 
 func testConnection(t *testing.T) (*connection, *fakeClient, *fakeTransport) {
        t.Helper()
 

-       d, client, transport := testAssembly(t)
+       d, client, transport, _ := testAssembly(t)

        connection := &connection{owner: d}
        require.NoError(t, connection.start(t.Context()))
 
        connection := &connection{owner: d}
        require.NoError(t, connection.start(t.Context()))
 


@@ -272,3 +303,32 @@ func testConnection(t *testing.T) (*connection, *fakeClient, *fakeTransport) {
 
        return connection, client, transport
 }
 
        return connection, client, transport
 }

+
+func queueCommands(t *testing.T, router *driver.Router, commands []driver.Command) {
+       t.Helper()
+
+       var (
+               publication protocol.Publication
+               err         error
+       )
+
+       if len(commands) == 1 {
+               command := commands[0]
+               publication, err = protocol.BuildCommand(command.DeviceID, command.Capability, command.Metadata, command.Payload)
+       } else {
+               members := make([]protocol.BatchCommand, len(commands))
+
+               for index, command := range commands {
+                       members[index] = protocol.BatchCommand{
+                               Capability: command.Capability,
+                               Metadata:   command.Metadata,
+                               Payload:    command.Payload,
+                       }
+               }
+
+               publication, err = protocol.BuildCommandBatch(testDeviceID, commands[0].Metadata, protocol.BatchPayload{Commands: members})
+       }
+
+       require.NoError(t, err)
+       require.NoError(t, router.Handle(t.Context(), publication.Topic, publication.Data, false))
+}

diff --git a/driver/m31/output_request.go b/driver/m31/output_request.go
new file mode 100644 (file)
index 0000000..bdba479
--- /dev/null
+++ b/driver/m31/output_request.go
@@ -0,0 +1,84 @@
+package m31
+
+import (
+       "bytes"
+       "encoding/json"
+       "math"
+       "math/big"
+       "time"
+
+       "onehouse/driver"
+)
+
+type outputRequest struct {
+       command driver.Command
+       point   point
+       analog  bool
+       value   uint16
+}
+
+func newOutputRequest(command driver.Command, topology topology, deviceID string, batch bool) (outputRequest, string) {
+       request := outputRequest{command: command}
+
+       if command.DeviceID != deviceID {
+               return request, "Command belongs to another device"
+       }
+
+       if command.Expired(time.Now()) {
+               return request, "Command expired"
+       }
+
+       value := bytes.TrimSpace(command.Payload.Value)
+
+       for _, output := range topology.outputs {
+               if output.id != command.Capability {
+                       continue
+               }
+
+               request.point = output
+               if !bytes.Equal(value, []byte("true")) && !bytes.Equal(value, []byte("false")) {
+                       return request, "Digital output requires a Boolean value"
+               }
+
+               if bytes.Equal(value, []byte("true")) {
+                       request.value = 1
+               }
+
+               if output.pulse != 0 && request.value != 1 {
+                       return request, "Pulse output requires true"
+               }
+
+               if batch && output.pulse != 0 {
+                       return request, "Batches require steady digital outputs"
+               }
+
+               return request, ""
+       }
+
+       for _, output := range topology.analog {
+               if output.id != command.Capability {
+                       continue
+               }
+
+               request.point, request.analog = output, true
+               if batch {
+                       return request, "Batches require steady digital outputs"
+               }
+
+               if len(value) == 0 || !json.Valid(value) || value[0] != '-' && (value[0] < '0' || value[0] > '9') {
+                       return request, "Analog output requires zero through ten volts"
+               }
+
+               number, valid := new(big.Rat).SetString(string(value))
+               if !valid || number.Sign() < 0 || number.Cmp(big.NewRat(10, 1)) > 0 {
+                       return request, "Analog output requires zero through ten volts"
+               }
+
+               voltage, _ := number.Float64()
+               request.value = uint16(math.Round(voltage * 1000))
+
+               return request, ""
+       }
+
+       return request, "Capability is not a configured output"
+}

diff --git a/driver/m31/output_request_test.go b/driver/m31/output_request_test.go
new file mode 100644 (file)
index 0000000..e96a9d3
--- /dev/null
+++ b/driver/m31/output_request_test.go
@@ -0,0 +1,76 @@
+package m31
+
+import (
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestOutputRequestRejectsUnsafeValuesAndAccess(t *testing.T) {
+       d, _, _, _ := testAssembly(t)
+       testCases := []struct {
+               name  string
+               point string
+               value string
+               batch bool
+       }{
+               {name: "rejects null Boolean", point: "m0.do0", value: "null"},
+               {name: "rejects numeric Boolean", point: "m0.do0", value: "1"},
+               {name: "rejects string Boolean", point: "m0.do0", value: `"true"`},
+               {name: "rejects input writes", point: "m0.di0", value: "true"},
+               {name: "rejects unknown points", point: "m8.do0", value: "true"},
+               {name: "rejects false pulses", point: "m0.do2", value: "false"},
+               {name: "rejects pulses in batches", point: "m0.do2", value: "true", batch: true},
+               {name: "rejects analog in batches", point: "m2.ao0", value: "5", batch: true},
+               {name: "rejects negative voltage", point: "m2.ao0", value: "-0.00000000000000001"},
+               {name: "rejects excessive voltage without float rounding", point: "m2.ao0", value: "10.00000000000000001"},
+               {name: "rejects string voltage", point: "m2.ao0", value: `"5"`},
+               {name: "rejects null voltage", point: "m2.ao0", value: "null"},
+               {name: "rejects malformed voltage", point: "m2.ao0", value: "5x"},
+               {name: "rejects missing voltage", point: "m2.ao0"},
+               {name: "rejects object voltage", point: "m2.ao0", value: "{}"},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       command := testCommand(t, tc.point, tc.value)
+                       _, reason := newOutputRequest(command, d.topology, testDeviceID, tc.batch)
+                       require.NotEmpty(t, reason)
+               })
+       }
+
+       command := testCommand(t, "m0.do0", "true")
+       command.DeviceID = "01ARZ3NDEKTSV4RRFFQ69G5FAW"
+       _, reason := newOutputRequest(command, d.topology, testDeviceID, false)
+       require.Contains(t, reason, "another device")
+
+       command.DeviceID = testDeviceID
+       expires := time.Now()
+       command.Metadata.ExpiresAt = &expires
+       _, reason = newOutputRequest(command, d.topology, testDeviceID, false)
+       require.Contains(t, reason, "expired")
+}
+
+func TestOutputRequestRoundsValidVoltageToMillivolts(t *testing.T) {
+       d, _, _, _ := testAssembly(t)
+       testCases := []struct {
+               name  string
+               value string
+               want  uint16
+       }{
+               {name: "accepts zero", value: "0"},
+               {name: "accepts full range", value: "10", want: 10000},
+               {name: "rounds half millivolt up", value: "5.5005", want: 5501},
+               {name: "accepts exponent notation", value: "55e-1", want: 5500},
+               {name: "accepts surrounding whitespace", value: " 5.5 ", want: 5500},
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       request, reason := newOutputRequest(testCommand(t, "m2.ao0", tc.value), d.topology, testDeviceID, false)
+                       require.Empty(t, reason)
+                       require.Equal(t, tc.want, request.value)
+               })
+       }
+}