fix(service): surface duplicate dispatch conflicts as Task errors

Author: NickSeagull

core/service/Service/Application.hs

@@ -1026,21 +1026,13 @@ runWithResolved eventStore maybeFileUploadSetup fileUploadCleanup maybeWebAuthSe
         wiredQueries
           |> Array.reduce (\(_reg, (name, _handler, schema)) acc -> acc |> Map.set name schema) Map.empty
 
-  -- 4. Create query subscriber with combined registry
-  subscriber <- Subscriber.new eventStore combinedRegistry
-
-  -- 5. Rebuild all queries from historical events
-  Subscriber.rebuildAll subscriber
-
-  -- 6. Start live subscription
-  Subscriber.start subscriber
-
-  -- 7. Collect command endpoints and schemas from all services, grouped by transport
+  -- 4. Collect command endpoints and schemas from all services, grouped by transport
   endpointsAndSchemasByTransport <-
     app.serviceRunners
       |> Task.mapArray (\runner -> runner.getEndpointsByTransport eventStore app.transports)
 
-  -- 8. Merge all public endpoints and schemas by transport name, plus dispatch handlers
+  -- 5. Merge all public endpoints and schemas by transport name, plus dispatch handlers
+  -- Duplicate dispatch handler names are surfaced as Task errors (not panics).
   let mergeEndpointsAndSchemas (serviceEndpoints, serviceSchemas, serviceDispatch) (handlersAcc, schemasAcc, dispatchAcc) = do
         let mergedHandlers = serviceEndpoints
               |> Map.entries
@@ -1060,20 +1052,45 @@ runWithResolved eventStore maybeFileUploadSetup fileUploadCleanup maybeWebAuthSe
                         Just existing -> innerAcc |> Map.set transportName (Map.merge cmdSchemas existing)
                    )
                    schemasAcc
-        let mergedDispatch = serviceDispatch
+        let mergedDispatchResult = serviceDispatch
               |> Map.entries
               |> Array.reduce
-                  ( \(commandName, handler) innerAcc ->
-                      case innerAcc |> Map.get commandName of
-                        Nothing -> innerAcc |> Map.set commandName handler
-                        Just _ -> panic [fmt|Duplicate command handler registered for integration dispatch: #{commandName}|]
+                  ( \(commandName, handler) innerResult ->
+                      case innerResult of
+                        Err err -> Err err
+                        Ok innerAcc ->
+                          case innerAcc |> Map.get commandName of
+                            Nothing -> Ok (innerAcc |> Map.set commandName handler)
+                            Just _ -> Err [fmt|Duplicate command handler registered for integration dispatch: #{commandName}|]
                   )
-                  dispatchAcc
-        (mergedHandlers, mergedSchemas, mergedDispatch)
+                  (Ok dispatchAcc)
+        case mergedDispatchResult of
+          Err err -> Err err
+          Ok mergedDispatch -> Ok (mergedHandlers, mergedSchemas, mergedDispatch)
 
-  let (combinedEndpointsByTransport, combinedSchemasByTransport, combinedCommandEndpoints) =
+  let mergedEndpointsResult =
         endpointsAndSchemasByTransport
-          |> Array.reduce mergeEndpointsAndSchemas (Map.empty, Map.empty, Map.empty)
+          |> Array.reduce
+              (\serviceResult accResult ->
+                case accResult of
+                  Err err -> Err err
+                  Ok acc -> mergeEndpointsAndSchemas serviceResult acc
+              )
+              (Ok (Map.empty, Map.empty, Map.empty))
+
+  (combinedEndpointsByTransport, combinedSchemasByTransport, combinedCommandEndpoints) <-
+    case mergedEndpointsResult of
+      Err err -> Task.throw err
+      Ok merged -> Task.yield merged
+
+  -- 6. Create query subscriber with combined registry
+  subscriber <- Subscriber.new eventStore combinedRegistry
+
+  -- 7. Rebuild all queries from historical events
+  Subscriber.rebuildAll subscriber
+
+  -- 8. Start live subscription
+  Subscriber.start subscriber
 
   -- 9. Initialize auth if configured (using pre-resolved WebAuthSetup)
   maybeAuthEnabled <- case maybeWebAuthSetup of

core/service/Service/ServiceDefinition/Core.hs

@@ -210,6 +210,52 @@ buildDispatchHandler _ handler requestContext body respond = do
       respond (errorResponse, responseJson)
 
 
+buildCommandResponseHandler ::
+  forall cmd entity event entityName name.
+  ( Command cmd,
+    name ~ NameOf cmd,
+    Record.KnownSymbol name,
+    Record.KnownHash name,
+    Entity entity,
+    Event event,
+    event ~ EventOf entity,
+    entity ~ EntityOf cmd,
+    entity ~ EntityOf event,
+    IsMultiTenant cmd ~ False,
+    StreamId.ToStreamId (EntityIdType entity),
+    Eq (EntityIdType entity),
+    Ord (EntityIdType entity),
+    Show (EntityIdType entity),
+    Show event,
+    Record.KnownSymbol entityName,
+    Json.FromJSON event,
+    Json.ToJSON event
+  ) =>
+  EventStore event ->
+  Maybe (SnapshotCache entity) ->
+  RequestContext ->
+  cmd ->
+  Task Text Response.CommandResponse
+buildCommandResponseHandler eventStore maybeCache requestContext cmdInstance = do
+  fetcher <- case maybeCache of
+    Just cache ->
+      EntityFetcher.newWithCache
+        eventStore
+        cache
+        (initialStateImpl @entity)
+        (updateImpl @entity)
+        |> Task.mapError toText
+    Nothing ->
+      EntityFetcher.new
+        eventStore
+        (initialStateImpl @entity)
+        (updateImpl @entity)
+        |> Task.mapError toText
+  let entityName = EntityName (getSymbolText (Record.Proxy @entityName))
+  result <- CommandExecutor.execute eventStore fetcher entityName requestContext cmdInstance
+  Task.yield (Response.fromExecutionResult result)
+
+
 instance
   ( Command cmd,
     Entity entity,
@@ -260,48 +306,16 @@ instance
   createHandlers _ eventStore maybeCache transportsMap cmd = do
     -- Build the handler that will receive RequestContext at call time
     let handler :: RequestContext -> cmd -> Task Text Response.CommandResponse
-        handler requestContext cmdInstance = do
-          fetcher <- case maybeCache of
-            Just cache ->
-              EntityFetcher.newWithCache
-                eventStore
-                cache
-                (initialStateImpl @entity)
-                (updateImpl @entity)
-                |> Task.mapError toText
-            Nothing ->
-              EntityFetcher.new
-                eventStore
-                (initialStateImpl @entity)
-                (updateImpl @entity)
-                |> Task.mapError toText
-          let entityName = EntityName (getSymbolText (Record.Proxy @(entityName)))
-          result <- CommandExecutor.execute eventStore fetcher entityName requestContext cmdInstance
-          Task.yield (Response.fromExecutionResult result)
+        handler requestContext cmdInstance =
+          buildCommandResponseHandler @cmd @entity @event @entityName eventStore maybeCache requestContext cmdInstance
 
     buildHandlersForAll @(PublicTransports transports) transportsMap cmd handler
 
 
   createDispatchHandler _ eventStore maybeCache cmd = do
     let handler :: RequestContext -> cmd -> Task Text Response.CommandResponse
-        handler requestContext cmdInstance = do
-          fetcher <- case maybeCache of
-            Just cache ->
-              EntityFetcher.newWithCache
-                eventStore
-                cache
-                (initialStateImpl @entity)
-                (updateImpl @entity)
-                |> Task.mapError toText
-            Nothing ->
-              EntityFetcher.new
-                eventStore
-                (initialStateImpl @entity)
-                (updateImpl @entity)
-                |> Task.mapError toText
-          let entityName = EntityName (getSymbolText (Record.Proxy @(entityName)))
-          result <- CommandExecutor.execute eventStore fetcher entityName requestContext cmdInstance
-          Task.yield (Response.fromExecutionResult result)
+        handler requestContext cmdInstance =
+          buildCommandResponseHandler @cmd @entity @event @entityName eventStore maybeCache requestContext cmdInstance
 
     buildDispatchHandler @cmd cmd handler
 
@@ -565,15 +579,22 @@ buildEndpointsByTransport rawEventStore commandDefinitions transportsMap = do
                 let newSchemas = schemasAcc |> Map.set transportNameText (existingSchemas |> Map.set cmdName endpointSchema)
                 (newHandlers, newSchemas)
 
-  let groupDispatch (cmdName, handler) dispatchAcc =
-        case dispatchAcc |> Map.get cmdName of
-          Just _ -> panic [fmt|Duplicate command handler registered for integration dispatch: #{cmdName}|]
-          Nothing -> dispatchAcc |> Map.set cmdName handler
+  let groupDispatch (cmdName, handler) dispatchResult =
+        case dispatchResult of
+          Err err -> Err err
+          Ok dispatchAcc ->
+            case dispatchAcc |> Map.get cmdName of
+              Just _ -> Err [fmt|Duplicate command handler registered for integration dispatch: #{cmdName}|]
+              Nothing -> Ok (dispatchAcc |> Map.set cmdName handler)
 
   let (handlersByTransport, schemasByTransport) =
         endpoints |> Array.reduce groupByTransport (Map.empty, Map.empty)
 
-  let dispatchMap = dispatchHandlers |> Array.reduce groupDispatch Map.empty
+  let dispatchMapResult = dispatchHandlers |> Array.reduce groupDispatch (Ok Map.empty)
+
+  dispatchMap <- case dispatchMapResult of
+    Err err -> Task.throw err
+    Ok mapValue -> Task.yield mapValue
 
   Task.yield (handlersByTransport, schemasByTransport, dispatchMap)
 

docs/decisions/0050-internal-command-transport.md

@@ -122,6 +122,10 @@ module Service.Transport.Internal (
 data InternalTransport = InternalTransport
 
 type instance NameOf InternalTransport = "InternalTransport"
+```
+
+```haskell
+-- Runtime hooks are owned by service runtime modules, not Service.Transport.Internal.
 
 dispatchCommand ::
   Map Text EndpointHandler ->
@@ -135,6 +139,8 @@ buildEndpointsByTransport ::
   Task Text (Map Text (Map Text EndpointHandler), Map Text (Map Text EndpointSchema), Map Text EndpointHandler)
 ```
 
+`dispatchCommand` is implemented in `Service.Integration.Dispatcher` and `buildEndpointsByTransport` is implemented in `Service.ServiceDefinition.Core`.
+
 Implementation consequences of this API:
 
 - `Service.ServiceDefinition.Core` must filter public transports away from `InternalTransport` while still building dispatch handlers for commands that use it.