Agent Signals and WMQ — Fit Analysis

[mp-orkes]

Agent Signals and WMQ — Fit Analysis

Date: 2026-03-27

This document analyses whether the Workflow Message Queue (WMQ) can serve as the foundation for the Agent Signals feature, what it already covers, and what gaps remain.

WMQ

• feat: Workflow Message Queue (WMQ) — push messages into running workflows conductor-oss/conductor#917

Signals

• Agent Signals — Requirements Specification
• Agent Signals — Design Specification

---

tl;dr

WMQ is a good fit as the transport layer for normal signals. It handles per-workflow message storage, REST push, FIFO ordering, and workflow-state validation out of the box. It is not sufficient on its own: urgent signals, signal disposition (accept/reject), status tracking, agent-name resolution, broadcast, sub-workflow propagation, and durability requirements all need additional work.

---

What WMQ already covers

Signal requirement	WMQ capability
Per-workflow message queue	✅ One queue per workflow ID
REST push from any caller	✅ POST /api/workflow/{workflowId}/messages
Arbitrary JSON payload	✅ payload field accepts any JSON object
FIFO ordering	✅ Guaranteed by RPUSH/LRANGE
Reject push to non-RUNNING workflow	✅ 409 Conflict
batchSize / batch dequeue	✅ PULL_WORKFLOW_MESSAGES.batchSize
Fast delivery (~1s latency)	✅ getEvaluationOffset() returns 1s

Signal message, data, priority, and sender can all be placed inside WMQ's payload field — WMQ is agnostic to payload structure. The signal service pushes a JSON object like:

{
  "signalId": "uuid-...",
  "message": "Focus on error correction",
  "data": {"topic": "QEC"},
  "priority": "normal",
  "sender": "supervisor_agent",
  "timestamp": 1711234567890
}

and the PULL_WORKFLOW_MESSAGES task delivers it as-is to the workflow.

---

What WMQ does not cover — gaps and required additions

1. Urgent signals (pause/resume)

WMQ has no concept of priority. Urgent signals must pause the workflow after the current task completes, then auto-resume after signal injection. This requires calling Conductor's pauseWorkflow / resumeWorkflow APIs from the signal service — WMQ is not involved.

Required addition: The Agentspan signal service checks priority == "urgent" and calls pauseWorkflow on the Conductor API. The workflow is then resumed immediately after the signal is queued.

2. Signal disposition (accept/reject) and status tracking

WMQ is a fire-and-forget queue: once a message is popped by PULL_WORKFLOW_MESSAGES, it is gone. There is no concept of the agent accepting or rejecting a message, and no way to query the outcome.

Signals require:

• disposition: pending → accepted | rejected
• rejectionReason per signal
• GET /agent/signal/{signalId}/status queryable at any time

This state cannot live in WMQ. It belongs in workflow variables (_processing_signals, _processed_signals) as the design describes — the INLINE + SET_VARIABLE pair in the DO_WHILE loop manages these transitions.

Required addition: The existing design's _pending_signals / _processing_signals / _processed_signals variable model is still needed. WMQ can replace _pending_signals as the storage medium (popping from WMQ instead of reading from a workflow variable), but the processed/disposition state still needs to live in workflow variables.

3. Signal count limits (100 lifetime, 10 pending)

WMQ has maxQueueSize (queue depth cap) but no per-signal lifetime counter. The 100-signal lifetime limit (FR-17.1) and 10-pending limit (FR-17.2) need to be enforced at the signal service layer by reading _signal_counts from workflow variables before pushing.

Required addition: The Agentspan signal service reads _signal_counts from the workflow's variables, validates limits, and only then pushes to WMQ.

4. Durability

FR-4.1 requires signals to survive server restarts. The current default WMQ DAO is InMemoryWorkflowMessageQueueDAO, which does not survive restarts. For signals to be durable, either:

• Redis must be configured (Redis WMQ DAO is durable), or
• A SQLite-backed WMQ DAO must be implemented (see the SQLite gap analysis in the WMQ docs)

Required addition: For production signal durability, the Redis DAO must be used, or a persistent DAO must be built for SQLite deployments.

5. Agent name resolution and broadcast

POST /agent/{workflowId}/signal targets one workflow. But the signal API also supports:

• runtime.signal(agent_name="researcher", ...) — resolves to one or more running workflows by name
• runtime.broadcast(workflow_ids=[...], ...) — sends to multiple workflows

WMQ operates per workflowId only. Resolution and broadcast are service-level concerns.

Required addition: The Agentspan signal service implements agent-name resolution (search for RUNNING/PAUSED workflows by workflow type) and calls WMQ push for each resolved workflow ID.

6. Sub-workflow propagation

When a signal is sent to a parent workflow, it should propagate to all active SUB_WORKFLOW tasks recursively. WMQ has no awareness of workflow hierarchy.

Required addition: The Agentspan signal service traverses active SUB_WORKFLOW tasks in the parent workflow and pushes the signal to each child workflow's WMQ.

7. SSE event emission

The spec requires signal_received, signal_accepted, and signal_rejected SSE events. WMQ pushes a message but does not emit any SSE events.

Required addition: The Agentspan signal service emits the signal_received SSE event after successfully pushing to WMQ. signal_accepted / signal_rejected events are emitted from the post-JOIN enrichment INLINE tasks (already in the design).

---

How WMQ fits into the overall signal flow

External caller (human, agent, SDK)
  └─ POST /agent/{workflowId}/signal   ← Agentspan server
       └─ AgentService.signal()
            ├─ Validate workflow is RUNNING (via Conductor API)
            ├─ Validate limits (_signal_counts from workflow variables)
            ├─ If urgent: pauseWorkflow()
            ├─ Build signal envelope (signalId, message, data, priority, sender)
            ├─ Push to WMQ: POST /api/workflow/{workflowId}/messages  ← Conductor WMQ
            ├─ If urgent: resumeWorkflow()
            ├─ Emit SSE: signal_received
            ├─ If propagate: repeat for each active sub-workflow
            └─ Return SignalReceipt

[Inside the Conductor workflow DO_WHILE loop]
  └─ PULL_WORKFLOW_MESSAGES (batchSize=10) ← replaces INLINE reading _pending_signals
       └─ If messages: pop signal envelopes
            └─ INLINE signal intake script
                 ├─ If auto_accept: move to _processed_signals, inject as context
                 ├─ If evaluate: move to _processing_signals, inject + ephemeral tools
                 └─ SET_VARIABLE: persist state, set _signal_injection
            └─ AgentChatCompleteTaskMapper reads _signal_injection (read-only)
            └─ LLM sees signal messages + accept/reject tools
            └─ accept_signal / reject_signal → INLINE → SET_VARIABLE → _processed_signals

WMQ replaces the updateVariables call to append to _pending_signals. Instead, the signal service pushes to WMQ, and the workflow reads it via PULL_WORKFLOW_MESSAGES. This is a cleaner separation: WMQ owns the inbox, workflow variables own the disposition state.

---

Where should the /agent signal endpoints live?

They belong in the Agentspan server (agentspan/server), not in Conductor.

Conductor is a general-purpose workflow engine. The signal endpoints contain Agentspan-specific logic that Conductor should not know about:

Concern	Lives in
Per-workflow message queue (durable inbox)	Conductor (WMQ)
Push a raw JSON payload to a workflow	Conductor (POST /api/workflow/{id}/messages)
Signal schema validation (message, priority, sender)	Agentspan server
Signal count limit enforcement	Agentspan server
Urgent pause/resume orchestration	Agentspan server
Agent-name resolution	Agentspan server
Broadcast	Agentspan server
Sub-workflow propagation	Agentspan server
SSE event emission	Agentspan server
Signal status queryability (disposition, rejectionReason)	Agentspan server + workflow variables
SignalReceipt / SignalStatus response types	Agentspan server

The Agentspan server calls Conductor's WMQ push endpoint as one step in a larger orchestration. Conductor remains unaware that "signals" exist as a concept.

---

Summary of required changes

Area	Change needed	Where
Normal signal delivery	Use WMQ push instead of updateVariables for _pending_signals	Agentspan server
Normal signal intake	Replace INLINE reading _pending_signals with PULL_WORKFLOW_MESSAGES	Agentspan task mapper
Urgent signal	Call pauseWorkflow + push to WMQ + resumeWorkflow	Agentspan server
Signal limits	Read _signal_counts from workflow variables before WMQ push	Agentspan server
Disposition tracking	Keep _processing_signals / _processed_signals in workflow variables	No change needed
Status query	GET /agent/signal/{signalId}/status reads _processed_signals variable	Agentspan server
Agent name resolution	Resolve name → workflow IDs, push to each WMQ	Agentspan server
Broadcast	Loop over workflow IDs, push to each WMQ	Agentspan server
Sub-workflow propagation	Traverse active SUB_WORKFLOW tasks, push to each WMQ	Agentspan server
SSE events	Emit signal_received after WMQ push	Agentspan server
Durability	Use Redis WMQ DAO or implement SQLite WMQ DAO	Conductor