fix(engine): coerce Python literal "True"/"False"/"None" in workflow output

src/conductor/engine/workflow.py

@@ -3018,13 +3018,27 @@ def _build_final_output(
     def _maybe_parse_json(value: str) -> Any:
         """Attempt to parse a string as JSON.
 
+        Also coerces Python literal string forms ("True", "False", "None") that
+        commonly arise from Jinja expressions like ``{{ a == b }}`` rendering a
+        Python ``bool`` via ``str()``. Without this, those values survive as
+        truthy non-empty strings downstream and silently misbehave in route
+        ``when:`` clauses.
+
         Args:
             value: The string to parse.
 
         Returns:
             Parsed JSON value if successful, original string otherwise.
         """
         stripped = value.strip()
+        # Python literal forms produced by str(bool) / str(None) — common from
+        # Jinja expressions in workflow output templates.
+        if stripped == "True":
+            return True
+        if stripped == "False":
+            return False
+        if stripped == "None":
+            return None
         if stripped.startswith(("{", "[", '"')) or stripped in ("true", "false", "null"):
             try:
                 return json.loads(stripped)

tests/test_engine/test_workflow.py

@@ -554,6 +554,118 @@ def mock_handler(agent, prompt, context):
 
         assert result["total"] == 42
 
+    @pytest.mark.asyncio
+    async def test_output_template_python_bool_literals(self) -> None:
+        """Python str(bool) outputs ('True'/'False') from Jinja expressions
+        coerce to native bool, not truthy non-empty strings.
+
+        Without this, ``{{ a == b }}`` in a workflow ``output:`` block renders
+        ``"True"`` / ``"False"`` and downstream ``when:`` clauses comparing to
+        ``true`` / ``false`` silently misbehave (the strings are truthy).
+        """
+        config = WorkflowConfig(
+            workflow=WorkflowDef(
+                name="bool-output",
+                entry_point="agent1",
+            ),
+            agents=[
+                AgentDef(
+                    name="agent1",
+                    model="gpt-4",
+                    prompt="x",
+                    output={
+                        "left": OutputField(type="string"),
+                        "right": OutputField(type="string"),
+                    },
+                    routes=[RouteDef(to="$end")],
+                ),
+            ],
+            output={
+                "matched": "{{ agent1.output.left == agent1.output.right }}",
+                "differs": "{{ agent1.output.left != agent1.output.right }}",
+            },
+        )
+
+        def mock_handler(agent, prompt, context):
+            return {"left": "abc", "right": "abc"}
+
+        provider = CopilotProvider(mock_handler=mock_handler)
+        engine = WorkflowEngine(config, provider)
+
+        result = await engine.run({})
+
+        assert result["matched"] is True
+        assert result["differs"] is False
+
+    @pytest.mark.asyncio
+    async def test_output_template_python_none_literal(self) -> None:
+        """Python str(None) ('None') from Jinja coerces to native None."""
+        config = WorkflowConfig(
+            workflow=WorkflowDef(
+                name="none-output",
+                entry_point="agent1",
+            ),
+            agents=[
+                AgentDef(
+                    name="agent1",
+                    model="gpt-4",
+                    prompt="x",
+                    output={"thing": OutputField(type="string")},
+                    routes=[RouteDef(to="$end")],
+                ),
+            ],
+            output={
+                # Jinja's `none` value renders as the string "None" via str(None).
+                "missing": "{{ none }}",
+            },
+        )
+
+        def mock_handler(agent, prompt, context):
+            return {"thing": "value"}
+
+        provider = CopilotProvider(mock_handler=mock_handler)
+        engine = WorkflowEngine(config, provider)
+
+        result = await engine.run({})
+
+        assert result["missing"] is None
+
+    @pytest.mark.asyncio
+    async def test_output_template_lowercase_json_literals_still_work(self) -> None:
+        """Regression: lowercase JSON literals 'true'/'false'/'null' remain coerced."""
+        config = WorkflowConfig(
+            workflow=WorkflowDef(
+                name="json-literals-output",
+                entry_point="agent1",
+            ),
+            agents=[
+                AgentDef(
+                    name="agent1",
+                    model="gpt-4",
+                    prompt="x",
+                    output={"v": OutputField(type="string")},
+                    routes=[RouteDef(to="$end")],
+                ),
+            ],
+            output={
+                "t": "true",
+                "f": "false",
+                "n": "null",
+            },
+        )
+
+        def mock_handler(agent, prompt, context):
+            return {"v": "x"}
+
+        provider = CopilotProvider(mock_handler=mock_handler)
+        engine = WorkflowEngine(config, provider)
+
+        result = await engine.run({})
+
+        assert result["t"] is True
+        assert result["f"] is False
+        assert result["n"] is None
+
 
 class TestWorkflowEngineLoopBack:
     """Tests for loop-back routing patterns."""

tests/test_integration/test_examples.py

@@ -211,7 +211,7 @@ def mock_handler(agent, prompt, context):
         assert "approval_decision" in result
         assert result["approval_decision"] == "approved"
         assert "syntax_passed" in result
-        assert result["syntax_passed"] == "True"  # Templates return strings
+        assert result["syntax_passed"] is True
 
 
 class TestExampleWorkflowsValidity:

tests/test_integration/test_parallel_workflows.py

@@ -407,7 +407,7 @@ def mock_handler(agent, prompt, context):
 
         # Verify output
         assert result["summary"] == "All tasks completed successfully"
-        assert result["success"] == "True"  # Boolean rendered as string
+        assert result["success"] is True
 
     def test_routing_from_parallel_group_based_on_results(self) -> None:
         """Test routing decisions based on parallel group outputs."""