feat(dispatch): turn-boundary batching dispatcher v2

charts/openab/templates/configmap.yaml

@@ -58,6 +58,19 @@ data:
     {{- if hasKey ($cfg.discord | default dict) "maxBotTurns" }}
     max_bot_turns = {{ ($cfg.discord).maxBotTurns | int }}
     {{- end }}
+    {{- /* messageProcessingMode: per-message (default) | per-thread | per-lane (turn-boundary batching) */ -}}
+    {{- if hasKey ($cfg.discord | default dict) "messageProcessingMode" }}
+    {{- if not (has ($cfg.discord).messageProcessingMode (list "per-message" "per-thread" "per-lane")) }}
+    {{- fail (printf "agents.%s.discord.messageProcessingMode must be one of: per-message, per-thread, per-lane — got: %s" $name ($cfg.discord).messageProcessingMode) }}
+    {{- end }}
+    message_processing_mode = {{ ($cfg.discord).messageProcessingMode | toJson }}
+    {{- end }}
+    {{- if hasKey ($cfg.discord | default dict) "maxBufferedMessages" }}
+    max_buffered_messages = {{ ($cfg.discord).maxBufferedMessages | int }}
+    {{- end }}
+    {{- if hasKey ($cfg.discord | default dict) "maxBatchTokens" }}
+    max_batch_tokens = {{ ($cfg.discord).maxBatchTokens | int }}
+    {{- end }}
     {{- end }}
 
     {{- if and ($cfg.slack).enabled }}
@@ -97,6 +110,19 @@ data:
     {{- if hasKey ($cfg.slack | default dict) "maxBotTurns" }}
     max_bot_turns = {{ ($cfg.slack).maxBotTurns | int }}
     {{- end }}
+    {{- /* messageProcessingMode: per-message (default) | per-thread | per-lane (turn-boundary batching) */ -}}
+    {{- if hasKey ($cfg.slack | default dict) "messageProcessingMode" }}
+    {{- if not (has ($cfg.slack).messageProcessingMode (list "per-message" "per-thread" "per-lane")) }}
+    {{- fail (printf "agents.%s.slack.messageProcessingMode must be one of: per-message, per-thread, per-lane — got: %s" $name ($cfg.slack).messageProcessingMode) }}
+    {{- end }}
+    message_processing_mode = {{ ($cfg.slack).messageProcessingMode | toJson }}
+    {{- end }}
+    {{- if hasKey ($cfg.slack | default dict) "maxBufferedMessages" }}
+    max_buffered_messages = {{ ($cfg.slack).maxBufferedMessages | int }}
+    {{- end }}
+    {{- if hasKey ($cfg.slack | default dict) "maxBatchTokens" }}
+    max_batch_tokens = {{ ($cfg.slack).maxBatchTokens | int }}
+    {{- end }}
     {{- end }}
 
     [agent]
@@ -170,6 +196,19 @@ data:
     {{- end }}
     {{- end }}
     allowed_users = {{ ($cfg.gateway).allowedUsers | default list | toJson }}
+    {{- /* messageProcessingMode: per-message (default) | per-thread | per-lane (turn-boundary batching) */ -}}
+    {{- if hasKey ($cfg.gateway | default dict) "messageProcessingMode" }}
+    {{- if not (has ($cfg.gateway).messageProcessingMode (list "per-message" "per-thread" "per-lane")) }}
+    {{- fail (printf "agents.%s.gateway.messageProcessingMode must be one of: per-message, per-thread, per-lane — got: %s" $name ($cfg.gateway).messageProcessingMode) }}
+    {{- end }}
+    message_processing_mode = {{ ($cfg.gateway).messageProcessingMode | toJson }}
+    {{- end }}
+    {{- if hasKey ($cfg.gateway | default dict) "maxBufferedMessages" }}
+    max_buffered_messages = {{ ($cfg.gateway).maxBufferedMessages | int }}
+    {{- end }}
+    {{- if hasKey ($cfg.gateway | default dict) "maxBatchTokens" }}
+    max_batch_tokens = {{ ($cfg.gateway).maxBatchTokens | int }}
+    {{- end }}
     {{- end }}
     {{- if or ($cfg.cronjobs) (($cfg.cron).usercronEnabled) (($cfg.cron).usercronPath) }}
 

charts/openab/tests/message-processing-mode_test.yaml

@@ -0,0 +1,124 @@
+suite: messageProcessingMode & batching params
+templates:
+  - templates/configmap.yaml
+tests:
+  - it: discord renders message_processing_mode = "per-lane"
+    set:
+      agents.kiro.discord.messageProcessingMode: per-lane
+    asserts:
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'message_processing_mode = "per-lane"'
+
+  - it: discord renders message_processing_mode = "per-thread"
+    set:
+      agents.kiro.discord.messageProcessingMode: per-thread
+    asserts:
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'message_processing_mode = "per-thread"'
+
+  - it: discord renders message_processing_mode = "per-message"
+    set:
+      agents.kiro.discord.messageProcessingMode: per-message
+    asserts:
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'message_processing_mode = "per-message"'
+
+  - it: discord rejects invalid messageProcessingMode value
+    set:
+      agents.kiro.discord.messageProcessingMode: batched
+    asserts:
+      - failedTemplate:
+          errorPattern: "must be one of: per-message, per-thread, per-lane"
+
+  - it: discord omits message_processing_mode when not set
+    asserts:
+      - notMatchRegex:
+          path: data["config.toml"]
+          pattern: 'message_processing_mode'
+
+  - it: discord renders maxBufferedMessages and maxBatchTokens
+    set:
+      agents.kiro.discord.messageProcessingMode: per-lane
+      agents.kiro.discord.maxBufferedMessages: 25
+      agents.kiro.discord.maxBatchTokens: 32000
+    asserts:
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'max_buffered_messages = 25'
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'max_batch_tokens = 32000'
+
+  - it: slack renders message_processing_mode = "per-thread"
+    set:
+      agents.kiro.slack.enabled: true
+      agents.kiro.slack.botToken: xoxb-x
+      agents.kiro.slack.appToken: xapp-x
+      agents.kiro.slack.messageProcessingMode: per-thread
+    asserts:
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'message_processing_mode = "per-thread"'
+
+  - it: slack rejects invalid messageProcessingMode value
+    set:
+      agents.kiro.slack.enabled: true
+      agents.kiro.slack.botToken: xoxb-x
+      agents.kiro.slack.appToken: xapp-x
+      agents.kiro.slack.messageProcessingMode: batched
+    asserts:
+      - failedTemplate:
+          errorPattern: "must be one of: per-message, per-thread, per-lane"
+
+  - it: slack renders maxBufferedMessages and maxBatchTokens
+    set:
+      agents.kiro.slack.enabled: true
+      agents.kiro.slack.botToken: xoxb-x
+      agents.kiro.slack.appToken: xapp-x
+      agents.kiro.slack.messageProcessingMode: per-lane
+      agents.kiro.slack.maxBufferedMessages: 15
+      agents.kiro.slack.maxBatchTokens: 18000
+    asserts:
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'max_buffered_messages = 15'
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'max_batch_tokens = 18000'
+
+  - it: gateway renders message_processing_mode = "per-lane"
+    set:
+      agents.kiro.gateway.enabled: true
+      agents.kiro.gateway.url: ws://openab-gateway:8080/ws
+      agents.kiro.gateway.messageProcessingMode: per-lane
+    asserts:
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'message_processing_mode = "per-lane"'
+
+  - it: gateway rejects invalid messageProcessingMode value
+    set:
+      agents.kiro.gateway.enabled: true
+      agents.kiro.gateway.url: ws://openab-gateway:8080/ws
+      agents.kiro.gateway.messageProcessingMode: batched
+    asserts:
+      - failedTemplate:
+          errorPattern: "must be one of: per-message, per-thread, per-lane"
+
+  - it: gateway renders maxBufferedMessages and maxBatchTokens
+    set:
+      agents.kiro.gateway.enabled: true
+      agents.kiro.gateway.url: ws://openab-gateway:8080/ws
+      agents.kiro.gateway.messageProcessingMode: per-thread
+      agents.kiro.gateway.maxBufferedMessages: 50
+      agents.kiro.gateway.maxBatchTokens: 12000
+    asserts:
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'max_buffered_messages = 50'
+      - matchRegex:
+          path: data["config.toml"]
+          pattern: 'max_batch_tokens = 12000'

charts/openab/values.yaml

@@ -161,6 +161,14 @@ agents:
       # multi-agent collaborations; lower to throttle runaway loops more
       # aggressively. Hard cap remains 100 regardless (compiled-in).
       # maxBotTurns: 20
+      # messageProcessingMode: "per-message" (default) | "per-thread" | "per-lane"
+      # per-thread: all senders in a thread share one batch → one ACP turn per turn boundary
+      # per-lane:   each (thread, sender) batches independently → no silent-drop risk
+      # messageProcessingMode: "per-lane"
+      # maxBufferedMessages: per-thread mpsc capacity for batching modes (default 10)
+      # maxBufferedMessages: 10
+      # maxBatchTokens: soft token cap per ACP turn for batching modes (default 24000)
+      # maxBatchTokens: 24000
     slack:
       enabled: false
       botToken: ""       # Bot User OAuth Token (xoxb-...)
@@ -185,6 +193,14 @@ agents:
       # multi-agent collaborations; lower to throttle runaway loops more
       # aggressively. Hard cap remains 100 regardless (compiled-in).
       # maxBotTurns: 20
+      # messageProcessingMode: "per-message" (default) | "per-thread" | "per-lane"
+      # per-thread: all senders in a thread share one batch → one ACP turn per turn boundary
+      # per-lane:   each (thread, sender) batches independently → no silent-drop risk
+      # messageProcessingMode: "per-lane"
+      # maxBufferedMessages: per-thread mpsc capacity for batching modes (default 10)
+      # maxBufferedMessages: 10
+      # maxBatchTokens: soft token cap per ACP turn for batching modes (default 24000)
+      # maxBatchTokens: 24000
     workingDir: /home/agent
     env: {}
     envFrom: []
@@ -211,6 +227,14 @@ agents:
       platform: "telegram"  # default platform when gateway is enabled
       token: ""         # optional shared secret (injected via GATEWAY_WS_TOKEN env var)
       botUsername: ""   # optional, for @mention gating
+      # messageProcessingMode: "per-message" (default) | "per-thread" | "per-lane"
+      # per-thread: all senders in a thread share one batch → one ACP turn per turn boundary
+      # per-lane:   each (thread, sender) batches independently → no silent-drop risk
+      # messageProcessingMode: "per-lane"
+      # maxBufferedMessages: per-thread mpsc capacity for batching modes (default 10)
+      # maxBufferedMessages: 10
+      # maxBatchTokens: soft token cap per ACP turn for batching modes (default 24000)
+      # maxBatchTokens: 24000
       image: "ghcr.io/openabdev/openab-gateway"  # gateway container image
       tag: ""           # defaults to Chart.AppVersion
       strategy: "Recreate"  # Recreate (default, prevents concurrent WS conflicts) or RollingUpdate