feat: Workflow Message Queue (WMQ) — push messages into running workflows

[mp-orkes]

What is this?

The Workflow Message Queue (WMQ) is a new opt-in feature that lets external systems push arbitrary JSON messages into a running workflow instance. The workflow consumes those messages at defined checkpoints using a new system task: PULL_WORKFLOW_MESSAGES.

This fills a gap that existing Conductor mechanisms don't cover:

• Event handlers operate at the workflow-definition level and can't target a specific running instance by ID.
• WAIT task has no structured payload — unblocking it carries no data.
• HTTP task requires the workflow to reach out. WMQ inverts that: the workflow receives data pushed by an external party.

Usage

1. Enable the feature

conductor.workflow-message-queue.enabled=true
conductor.workflow-message-queue.maxQueueSize=1000
conductor.workflow-message-queue.ttlSeconds=86400
conductor.workflow-message-queue.maxBatchSize=100

2. Add a PULL_WORKFLOW_MESSAGES task to your workflow

{
  "name": "wait_for_approval",
  "taskReferenceName": "wait_for_approval_ref",
  "type": "PULL_WORKFLOW_MESSAGES",
  "inputParameters": {
    "batchSize": 1
  }
}

The task stays IN_PROGRESS until at least one message arrives, then completes with the messages in its output:

{
  "messages": [
    {
      "id": "3f2504e0-...",
      "workflowId": "8e2c14e1-...",
      "payload": { "decision": "approved", "approvedBy": "user@example.com" },
      "receivedAt": "2025-06-15T10:30:00Z"
    }
  ],
  "count": 1
}

3. Push a message from an external system

POST /api/workflow/{workflowId}/messages
Content-Type: application/json

{ "decision": "approved", "approvedBy": "user@example.com" }

Returns the generated message ID. Rejects with 409 Conflict if the workflow is not RUNNING.

When the feature flag is off, the endpoint does not exist at all (absent from Swagger, returns 404).

Use cases

Use case	How WMQ helps
Agentic / agent loops	An AI agent workflow waits for tool results or human confirmations. The caller pushes a message; the loop unblocks.
Webhook-driven workflows	An async HTTP callback injects data into a paused workflow. The callback target is the WMQ push endpoint.
Human-in-the-loop	Structured approval decisions are injected into a running workflow by an operator tool or UI.
Notification pipelines	A workflow loops and reads messages in configurable batches, then fans out.

Implementation overview

Full details are in the docs added with this PR:

• docs/workflow-message-queue-architecture.md — design rationale, data flows, failure modes, security considerations
• docs/workflow-message-queue-implementation.md — class-level reference for every file added or modified

Components added

Component	Module	Purpose
WorkflowMessage	common	POJO for a single message (id, workflowId, payload, receivedAt)
WorkflowMessageQueueDAO	core	Interface: push, pop, size, delete
WorkflowMessageQueueProperties	core	@ConfigurationProperties for all WMQ settings
InMemoryWorkflowMessageQueueDAO	core	Default fallback DAO using ConcurrentHashMap
PullWorkflowMessages	core	Async system task; polls every 1s via getEvaluationOffset()
PullWorkflowMessagesTaskMapper	core	Task mapper so DeciderService can schedule the task
RedisWorkflowMessageQueueDAO	redis-persistence	Redis List-backed DAO (active when AnyRedisCondition is true)
WorkflowMessageQueueResource	rest	REST controller for the push endpoint

Files modified

File	Change
TaskType.java	Added PULL_WORKFLOW_MESSAGES enum value and string constant
JedisProxy.java	Added rpush, llen, lrange, ltrim list operations
application.properties	WMQ feature enabled by default in the OSS server

Test plan

• Start server, create a workflow with a PULL_WORKFLOW_MESSAGES task, verify task enters IN_PROGRESS
• Push a message via POST /api/workflow/{id}/messages, verify task completes within ~1s with message in output
• Push multiple messages with batchSize > 1, verify all are returned in a single completion
• Push to a completed workflow, verify 409 Conflict response
• Set conductor.workflow-message-queue.enabled=false, verify endpoint returns 404 and task type is unknown
• Verify maxQueueSize is enforced (push beyond limit returns error)

🤖 Generated with Claude Code

[mp-orkes]

Closing in favor of a new PR from branch feat/workflow-message-queue.