fix(engine): inherit background mode in subworkflows

src/conductor/engine/workflow.py

@@ -484,6 +484,7 @@ def __init__(
         instructions_preamble: str | None = None,
         plugin_marketplaces: Mapping[str, Marketplace] | None = None,
         _guidance_channel: GuidanceChannel | None = None,
+        _inherited_bg_mode: bool = False,
     ) -> None:
         """Initialize the WorkflowEngine.
 
@@ -531,6 +532,10 @@ def __init__(
                 mid-run would stall an agent on a network round trip, and
                 a failure there would surface as an agent error rather
                 than a configuration one. Inherited by sub-workflows.
+            _inherited_bg_mode: Parent engine's background-mode flag. Only the
+                CLI supplies a ``run_context``, so child engines must be told
+                explicitly that stdin cannot be prompted; otherwise a gate
+                inside a sub-workflow crashes on ``EOFError``.
             _guidance_channel: Shared mid-run guidance channel (issue #400).
                 When None, a fresh :class:`GuidanceChannel` is created. Child
                 engines inherit the parent's channel so a paused sub-workflow
@@ -724,7 +729,11 @@ def __init__(
 
         # System metadata fields (set by CLI, used in workflow_started event)
         self._dashboard_port = self._run_context.dashboard_port
-        self._bg_mode = self._run_context.bg_mode
+        # Only the CLI builds a RunContext, so a child engine would otherwise
+        # read bg_mode as False and let a gate prompt a stdin that is not
+        # there. Interaction environment is a property of the process, not of
+        # the nesting level.
+        self._bg_mode = self._run_context.bg_mode or _inherited_bg_mode
         self._system_metadata: dict[str, Any] = {}
 
         # When True, ``_execute_loop`` skips its ``workflow_started`` emit.
@@ -2635,6 +2644,7 @@ async def _execute_subworkflow(
             instructions_preamble=child_preamble,
             plugin_marketplaces=child_marketplaces,
             _guidance_channel=self._guidance,
+            _inherited_bg_mode=self._bg_mode,
         )
 
         output = await self._run_child_engine(child_engine, sub_inputs, agent)
@@ -2722,6 +2732,7 @@ async def _execute_subworkflow_with_inputs(
             "web_dashboard": self._web_dashboard,
             "_subworkflow_depth": self._subworkflow_depth + 1,
             "_guidance_channel": self._guidance,
+            "_inherited_bg_mode": self._bg_mode,
         }
         # Thread the dashboard context path into the child engine when the
         # field exists on this engine (added by the breadcrumb-navigation PR).

tests/test_engine/test_subworkflow.py

@@ -3056,3 +3056,94 @@ def mock_handler(agent, prompt, context):
         # one fetch for the whole run, not two.
         assert len(fetch_calls) == 1
         assert result["result"] == "ok"
+
+
+class TestSubWorkflowGateEnvironment:
+    """A gate inside a sub-workflow must see the parent's interaction environment."""
+
+    @pytest.mark.asyncio
+    async def test_nested_gate_inherits_bg_mode_and_skips_cli_prompt(
+        self, tmp_workflow_dir: Path
+    ) -> None:
+        """In ``--web-bg``, a gate nested in a sub-workflow must wait web-only.
+
+        Only the CLI builds a ``RunContext``, so a child engine used to read
+        ``bg_mode`` as False. With a dashboard attached and stdin looking like
+        a TTY, the gate then raced the CLI arm, ``Prompt.ask`` raised
+        ``EOFError`` instantly, won ``FIRST_COMPLETED``, and failed the run --
+        the issue #286 crash surviving one level of nesting.
+        """
+        from unittest.mock import AsyncMock, patch
+
+        from conductor.engine.workflow import RunContext
+
+        _write_yaml(
+            tmp_workflow_dir / "sub.yaml",
+            """\
+            workflow:
+              name: sub-gate
+              entry_point: approval_gate
+              runtime:
+                provider: copilot
+              limits:
+                max_iterations: 5
+            agents:
+              - name: approval_gate
+                type: human_gate
+                prompt: "Approve?"
+                options:
+                  - label: Approve
+                    value: approve
+                    route: "$end"
+            output:
+              decision: "approve"
+            """,
+        )
+
+        parent_path = tmp_workflow_dir / "parent.yaml"
+        parent_path.write_text("dummy", encoding="utf-8")
+
+        config = WorkflowConfig(
+            workflow=WorkflowDef(
+                name="parent-gate",
+                entry_point="nested",
+                runtime=RuntimeConfig(provider="copilot"),
+                limits=LimitsConfig(max_iterations=10),
+            ),
+            agents=[
+                AgentDef(
+                    name="nested",
+                    type="workflow",
+                    workflow="sub.yaml",
+                    routes=[RouteDef(to="$end")],
+                ),
+            ],
+            output={"decision": "{{ nested.output.decision }}"},
+        )
+
+        mock_dashboard = MagicMock()
+        mock_dashboard.wait_for_gate_response = AsyncMock(
+            return_value={"selected_value": "approve", "additional_input": {}}
+        )
+        mock_dashboard.has_connections = MagicMock(return_value=True)
+
+        provider = CopilotProvider(mock_handler=lambda agent, prompt, context: {})
+        engine = WorkflowEngine(
+            config,
+            provider,
+            workflow_path=parent_path,
+            skip_gates=False,
+            web_dashboard=mock_dashboard,
+            run_context=RunContext(bg_mode=True),
+        )
+
+        # A TTY-looking stdin is what made the child take the racing path.
+        with (
+            patch("conductor.gates.human.sys.stdin.isatty", return_value=True),
+            patch("conductor.gates.human.Prompt.ask", side_effect=EOFError) as cli_prompt,
+        ):
+            result = await engine.run({})
+
+        assert result["decision"] == "approve"
+        cli_prompt.assert_not_called()
+        mock_dashboard.wait_for_gate_response.assert_awaited_once()