fix(orchestrator): auto-reply for terminal agent nodes without reply_chat

When an agent node (agent/deep_agent) is terminal (no downstream
executable nodes) and the workflow has no explicit reply_chat node,
promote its output to state["output"] so _extract_output() → deliver()
sends the response back via the gateway.

This fixes the E2E chat round-trip for the default-agent workflow,
which has no reply_chat node but should still reply to the user.

Constraint: Only agent-type nodes trigger auto-reply — other terminal
  nodes (switch, code, etc.) do not produce user-facing responses
Rejected: Add reply_chat to every workflow template | couples template
  design to delivery mechanism
Confidence: high
Scope-risk: narrow

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: aka

platform/services/orchestrator.py

@@ -1024,6 +1024,46 @@ def _resume_from_child(
 
 # ── Advance / finalize ─────────────────────────────────────────────────────────
 
+_AGENT_COMPONENT_TYPES = {"agent", "deep_agent"}
+
+
+def _auto_promote_agent_output(
+    completed_node_id: str,
+    state: dict,
+    topo_data: dict,
+    execution_id: str,
+) -> None:
+    """Promote a terminal agent node's output to state["output"].
+
+    When an agent node is terminal (no downstream executable nodes) and the
+    workflow has no explicit ``reply_chat`` node, the agent's output is the
+    intended chat reply.  Setting ``state["output"]`` ensures
+    ``_extract_output()`` finds it and ``deliver()`` sends it back.
+    """
+    if state.get("output") is not None:
+        return  # already set by another node
+
+    node_info = topo_data["nodes"].get(completed_node_id)
+    if not node_info or node_info.get("component_type") not in _AGENT_COMPONENT_TYPES:
+        return
+
+    has_reply_chat = any(
+        n.get("component_type") == "reply_chat"
+        for n in topo_data["nodes"].values()
+    )
+    if has_reply_chat:
+        return
+
+    node_output = state.get("node_outputs", {}).get(completed_node_id, {})
+    output_text = node_output.get("output")
+    if output_text:
+        state["output"] = output_text
+        save_state(execution_id, state)
+        logger.info(
+            "Auto-promoted terminal agent %s output for execution %s",
+            completed_node_id, execution_id,
+        )
+
 
 def _advance(
     execution_id: str,
@@ -1043,6 +1083,12 @@ def _advance(
         if _check_loop_body_done(execution_id, completed_node_id, topo_data, db, delay_seconds=delay_seconds):
             r.decr(_inflight_key(execution_id))
             return
+
+        # Auto-reply: if a terminal agent node completes and the workflow
+        # has no explicit reply_chat node, promote its output to
+        # state["output"] so _finalize → deliver() sends the reply.
+        _auto_promote_agent_output(completed_node_id, state, topo_data, execution_id)
+
         remaining = r.decr(_inflight_key(execution_id))
         if remaining <= 0:
             _finalize(execution_id, db)

platform/tests/test_orchestrator.py

@@ -373,6 +373,123 @@ def test_advance_no_successors_tries_finalize(self):
             _advance("exec-1", "a", state, topo_data, MagicMock())
             mock_fin.assert_called_once()
 
+    def test_advance_terminal_agent_auto_promotes_output(self):
+        """Terminal agent node with no reply_chat promotes output to state."""
+        from services.orchestrator import _advance
+
+        topo_data = {
+            "nodes": {"deep_agent_1": {"component_type": "deep_agent"}},
+            "edges_by_source": {},
+            "incoming_count": {"deep_agent_1": 0},
+        }
+        state = {
+            "node_outputs": {"deep_agent_1": {"output": "Hello from agent"}},
+        }
+
+        with patch("services.orchestrator._finalize") as mock_fin, \
+             patch("services.orchestrator._redis") as mock_r, \
+             patch("services.orchestrator.save_state") as mock_save, \
+             patch("services.orchestrator._publish_event"):
+            mock_redis = MagicMock()
+            mock_r.return_value = mock_redis
+            mock_redis.decr.return_value = 0
+
+            _advance("exec-1", "deep_agent_1", state, topo_data, MagicMock())
+
+            assert state["output"] == "Hello from agent"
+            mock_save.assert_called_once_with("exec-1", state)
+            mock_fin.assert_called_once()
+
+    def test_advance_terminal_agent_skips_when_reply_chat_exists(self):
+        """Terminal agent does NOT auto-promote if reply_chat node exists."""
+        from services.orchestrator import _advance
+
+        topo_data = {
+            "nodes": {
+                "agent_1": {"component_type": "agent"},
+                "reply_chat_1": {"component_type": "reply_chat"},
+            },
+            "edges_by_source": {},
+            "incoming_count": {"agent_1": 0, "reply_chat_1": 1},
+        }
+        state = {
+            "node_outputs": {"agent_1": {"output": "Hello"}},
+        }
+
+        with patch("services.orchestrator._finalize") as mock_fin, \
+             patch("services.orchestrator._redis") as mock_r, \
+             patch("services.orchestrator.save_state") as mock_save, \
+             patch("services.orchestrator._publish_event"):
+            mock_redis = MagicMock()
+            mock_r.return_value = mock_redis
+            mock_redis.decr.return_value = 0
+
+            _advance("exec-1", "agent_1", state, topo_data, MagicMock())
+
+            assert "output" not in state
+            mock_save.assert_not_called()
+
+    def test_advance_terminal_non_agent_does_not_promote(self):
+        """Terminal non-agent node (e.g. switch) does NOT auto-promote."""
+        from services.orchestrator import _advance
+
+        topo_data = {
+            "nodes": {"switch_1": {"component_type": "switch"}},
+            "edges_by_source": {},
+            "incoming_count": {"switch_1": 0},
+        }
+        state = {
+            "node_outputs": {"switch_1": {"route": "category_a"}},
+        }
+
+        with patch("services.orchestrator._finalize") as mock_fin, \
+             patch("services.orchestrator._redis") as mock_r, \
+             patch("services.orchestrator.save_state") as mock_save, \
+             patch("services.orchestrator._publish_event"):
+            mock_redis = MagicMock()
+            mock_r.return_value = mock_redis
+            mock_redis.decr.return_value = 0
+
+            _advance("exec-1", "switch_1", state, topo_data, MagicMock())
+
+            assert "output" not in state
+            mock_save.assert_not_called()
+
+    def test_advance_non_terminal_agent_does_not_promote(self):
+        """Agent with downstream edges does NOT auto-promote."""
+        from services.orchestrator import _advance
+
+        topo_data = {
+            "nodes": {
+                "agent_1": {"component_type": "agent"},
+                "switch_1": {"component_type": "switch"},
+            },
+            "edges_by_source": {
+                "agent_1": [{"source_node_id": "agent_1", "target_node_id": "switch_1", "edge_type": "direct"}],
+            },
+            "incoming_count": {"agent_1": 0, "switch_1": 1},
+        }
+        state = {
+            "node_outputs": {"agent_1": {"output": "Hello"}},
+        }
+
+        with patch("services.orchestrator._queue") as mock_q, \
+             patch("services.orchestrator._redis") as mock_r, \
+             patch("services.orchestrator.save_state") as mock_save, \
+             patch("services.orchestrator._publish_event"):
+            mock_queue = MagicMock()
+            mock_q.return_value = mock_queue
+            mock_redis = MagicMock()
+            mock_r.return_value = mock_redis
+            mock_redis.decr.return_value = 0
+
+            _advance("exec-1", "agent_1", state, topo_data, MagicMock())
+
+            assert "output" not in state
+            mock_save.assert_not_called()
+            # Should enqueue the switch instead
+            mock_queue.enqueue.assert_called_once()
+
 
 # ── Switch component tests ────────────────────────────────────────────────────