diff --git a/src/exec.cpp b/src/exec.cpp
index 0d2291c42..34c402fae 100644
--- a/src/exec.cpp
+++ b/src/exec.cpp
@@ -1096,13 +1096,15 @@ static void populate_subshell_output(wcstring_list_t *lst, const io_buffer_t &bu
 
 /// Execute \p cmd in a subshell in \p parser. If \p lst is not null, populate it with the output.
 /// Return $status in \p out_status.
+/// If \p parent_pgid is set, any spawned commands should join that pgroup.
 /// If \p apply_exit_status is false, then reset $status back to its original value.
 /// \p is_subcmd controls whether we apply a read limit.
 /// \p break_expand is used to propagate whether the result should be "expansion breaking" in the
 /// sense that subshells used during string expansion should halt that expansion. \return the value
 /// of $status.
-static int exec_subshell_internal(const wcstring &cmd, parser_t &parser, wcstring_list_t *lst,
-                                  bool *break_expand, bool apply_exit_status, bool is_subcmd) {
+static int exec_subshell_internal(const wcstring &cmd, parser_t &parser, maybe_t<pid_t> parent_pgid,
+                                  wcstring_list_t *lst, bool *break_expand, bool apply_exit_status,
+                                  bool is_subcmd) {
     ASSERT_IS_MAIN_THREAD();
     auto &ld = parser.libdata();
 
@@ -1125,7 +1127,8 @@ static int exec_subshell_internal(const wcstring &cmd, parser_t &parser, wcstrin
         *break_expand = true;
         return STATUS_CMD_ERROR;
     }
-    eval_res_t eval_res = parser.eval(cmd, io_chain_t{bufferfill}, block_type_t::subst);
+    eval_res_t eval_res =
+        parser.eval(cmd, io_chain_t{bufferfill}, parent_pgid, block_type_t::subst);
     std::shared_ptr<io_buffer_t> buffer = io_bufferfill_t::finish(std::move(bufferfill));
     if (buffer->buffer().discarded()) {
         *break_expand = true;
@@ -1144,21 +1147,24 @@ static int exec_subshell_internal(const wcstring &cmd, parser_t &parser, wcstrin
     return eval_res.status.status_value();
 }
 
-int exec_subshell_for_expand(const wcstring &cmd, parser_t &parser, wcstring_list_t &outputs) {
+int exec_subshell_for_expand(const wcstring &cmd, parser_t &parser, maybe_t<pid_t> parent_pgid,
+                             wcstring_list_t &outputs) {
     ASSERT_IS_MAIN_THREAD();
     bool break_expand = false;
-    int ret = exec_subshell_internal(cmd, parser, &outputs, &break_expand, true, true);
+    int ret = exec_subshell_internal(cmd, parser, parent_pgid, &outputs, &break_expand, true, true);
     // Only return an error code if we should break expansion.
     return break_expand ? ret : STATUS_CMD_OK;
 }
 
 int exec_subshell(const wcstring &cmd, parser_t &parser, bool apply_exit_status) {
     bool break_expand = false;
-    return exec_subshell_internal(cmd, parser, nullptr, &break_expand, apply_exit_status, false);
+    return exec_subshell_internal(cmd, parser, none(), nullptr, &break_expand, apply_exit_status,
+                                  false);
 }
 
 int exec_subshell(const wcstring &cmd, parser_t &parser, wcstring_list_t &outputs,
                   bool apply_exit_status) {
     bool break_expand = false;
-    return exec_subshell_internal(cmd, parser, &outputs, &break_expand, apply_exit_status, false);
+    return exec_subshell_internal(cmd, parser, none(), &outputs, &break_expand, apply_exit_status,
+                                  false);
 }
diff --git a/src/exec.h b/src/exec.h
index 6c52be8db..7eb657209 100644
--- a/src/exec.h
+++ b/src/exec.h
@@ -29,8 +29,10 @@ int exec_subshell(const wcstring &cmd, parser_t &parser, wcstring_list_t &output
 
 /// Like exec_subshell, but only returns expansion-breaking errors. That is, a zero return means
 /// "success" (even though the command may have failed), a non-zero return means that we should
-/// halt expansion.
-int exec_subshell_for_expand(const wcstring &cmd, parser_t &parser, wcstring_list_t &outputs);
+/// halt expansion. If the \p pgid is supplied, then any spawned external commands should join that
+/// pgroup.
+int exec_subshell_for_expand(const wcstring &cmd, parser_t &parser, maybe_t<pid_t> parent_pgid,
+                             wcstring_list_t &outputs);
 
 /// Loops over close until the syscall was run without being interrupted.
 void exec_close(int fd);
diff --git a/src/expand.cpp b/src/expand.cpp
index 942461dff..8165645d1 100644
--- a/src/expand.cpp
+++ b/src/expand.cpp
@@ -591,10 +591,12 @@ static expand_result_t expand_braces(wcstring &&instr, expand_flags_t flags, com
     return expand_result_t::ok;
 }
 
-/// Expand a command substitution \p input, executing on \p parser, and inserting the results into
+/// Expand a command substitution \p input, executing on \p ctx, and inserting the results into
 /// \p out_list, or any errors into \p errors. \return an expand result.
-static expand_result_t expand_cmdsubst(wcstring input, parser_t &parser,
+static expand_result_t expand_cmdsubst(wcstring input, const operation_context_t &ctx,
                                        completion_list_t *out_list, parse_error_list_t *errors) {
+    assert(ctx.parser && "Cannot expand without a parser");
+
     wchar_t *paren_begin = nullptr, *paren_end = nullptr;
     wchar_t *tail_begin = nullptr;
     size_t i, j;
@@ -620,7 +622,7 @@ static expand_result_t expand_cmdsubst(wcstring input, parser_t &parser,
 
     wcstring_list_t sub_res;
     const wcstring subcmd(paren_begin + 1, paren_end - paren_begin - 1);
-    int subshell_status = exec_subshell_for_expand(subcmd, parser, sub_res);
+    int subshell_status = exec_subshell_for_expand(subcmd, *ctx.parser, ctx.parent_pgid, sub_res);
     if (subshell_status != 0) {
         // TODO: Ad-hoc switch, how can we enumerate the possible errors more safely?
         const wchar_t *err;
@@ -674,7 +676,7 @@ static expand_result_t expand_cmdsubst(wcstring input, parser_t &parser,
     // Recursively call ourselves to expand any remaining command substitutions. The result of this
     // recursive call using the tail of the string is inserted into the tail_expand array list
     completion_list_t tail_expand;
-    expand_cmdsubst(tail_begin, parser, &tail_expand, errors);  // TODO: offset error locations
+    expand_cmdsubst(tail_begin, ctx, &tail_expand, errors);  // TODO: offset error locations
 
     // Combine the result of the current command substitution with the result of the recursive tail
     // expansion.
@@ -919,7 +921,7 @@ expand_result_t expander_t::stage_cmdsubst(wcstring input, completion_list_t *ou
         }
     } else {
         assert(ctx.parser && "Must have a parser to expand command substitutions");
-        return expand_cmdsubst(std::move(input), *ctx.parser, out, errors);
+        return expand_cmdsubst(std::move(input), ctx, out, errors);
     }
 }
 
diff --git a/src/operation_context.h b/src/operation_context.h
index a230596a8..6eb79831e 100644
--- a/src/operation_context.h
+++ b/src/operation_context.h
@@ -29,6 +29,11 @@ class operation_context_t {
     // context itself.
     const environment_t &vars;
 
+    /// The pgid of the parental job.
+    /// This is used only when expanding command substitutions. If this is set, any jobs created by
+    /// the command substitions should use this pgid.
+    maybe_t<pid_t> parent_pgid{};
+
     // A function which may be used to poll for cancellation.
     cancel_checker_t cancel_checker;
 
diff --git a/src/parser.cpp b/src/parser.cpp
index 45250008b..87a000d0e 100644
--- a/src/parser.cpp
+++ b/src/parser.cpp
@@ -640,11 +640,12 @@ profile_item_t *parser_t::create_profile_item() {
     return result;
 }
 
-eval_res_t parser_t::eval(const wcstring &cmd, const io_chain_t &io, enum block_type_t block_type) {
+eval_res_t parser_t::eval(const wcstring &cmd, const io_chain_t &io, maybe_t<pid_t> parent_pgid,
+                          enum block_type_t block_type) {
     // Parse the source into a tree, if we can.
     parse_error_list_t error_list;
     if (parsed_source_ref_t ps = parse_source(cmd, parse_flag_none, &error_list)) {
-        return this->eval(ps, io, block_type);
+        return this->eval(ps, io, parent_pgid, block_type);
     } else {
         // Get a backtrace. This includes the message.
         wcstring backtrace_and_desc;
@@ -661,11 +662,12 @@ eval_res_t parser_t::eval(const wcstring &cmd, const io_chain_t &io, enum block_
 }
 
 eval_res_t parser_t::eval(const parsed_source_ref_t &ps, const io_chain_t &io,
-                          enum block_type_t block_type) {
+                          maybe_t<pid_t> parent_pgid, enum block_type_t block_type) {
     assert(block_type == block_type_t::top || block_type == block_type_t::subst);
     if (!ps->tree.empty()) {
         job_lineage_t lineage;
         lineage.block_io = io;
+        lineage.parent_pgid = parent_pgid;
         // Execute the first node.
         tnode_t<grammar::job_list> start{&ps->tree, &ps->tree.front()};
         return this->eval_node(ps, start, std::move(lineage), block_type);
@@ -703,6 +705,9 @@ eval_res_t parser_t::eval_node(const parsed_source_ref_t &ps, tnode_t<T> node,
     operation_context_t op_ctx = this->context();
     block_t *scope_block = this->push_block(block_t::scope_block(block_type));
 
+    // Propogate any parent pgid.
+    op_ctx.parent_pgid = lineage.parent_pgid;
+
     // Create and set a new execution context.
     using exc_ctx_ref_t = std::unique_ptr<parse_execution_context_t>;
     scoped_push<exc_ctx_ref_t> exc(&execution_context, make_unique<parse_execution_context_t>(
diff --git a/src/parser.h b/src/parser.h
index b93b031b4..6839158b3 100644
--- a/src/parser.h
+++ b/src/parser.h
@@ -282,19 +282,20 @@ class parser_t : public std::enable_shared_from_this<parser_t> {
     ///
     /// \param cmd the string to evaluate
     /// \param io io redirections to perform on all started jobs
+    /// \param parent_pgid if set, the pgid to give to spawned jobs
     /// \param block_type The type of block to push on the block stack, which must be either 'top'
     /// or 'subst'.
     /// \param break_expand If not null, return by reference whether the error ought to be an expand
     /// error. This includes nested expand errors, and command-not-found.
     ///
     /// \return the result of evaluation.
-    eval_res_t eval(const wcstring &cmd, const io_chain_t &io,
+    eval_res_t eval(const wcstring &cmd, const io_chain_t &io, maybe_t<pid_t> parent_pgid = {},
                     block_type_t block_type = block_type_t::top);
 
     /// Evaluate the parsed source ps.
     /// Because the source has been parsed, a syntax error is impossible.
     eval_res_t eval(const parsed_source_ref_t &ps, const io_chain_t &io,
-                    block_type_t block_type = block_type_t::top);
+                    maybe_t<pid_t> parent_pgid = {}, block_type_t block_type = block_type_t::top);
 
     /// Evaluates a node.
     /// The node type must be grammar::statement or grammar::job_list.
diff --git a/tests/checks/pipeline-pgroup.fish b/tests/checks/pipeline-pgroup.fish
index 7973d45c8..4b7c86e8b 100644
--- a/tests/checks/pipeline-pgroup.fish
+++ b/tests/checks/pipeline-pgroup.fish
@@ -11,11 +11,13 @@ end
 # Here everything should live in the pgroup of the first fish_test_helper.
 $fth print_pgrp | read -g global_group | save_pgroup g1 | begin
     save_pgroup g2
+end | begin
+    echo (save_pgroup g3) >/dev/null
 end
 
-[ "$global_group" -eq "$g1" ] && [ "$g1" -eq "$g2" ]
+[ "$global_group" -eq "$g1" ] && [ "$g1" -eq "$g2" ] && [ "$g2" -eq "$g3" ]
 and echo "All pgroups agreed"
-or echo "Pgroups disagreed. Should be in $global_group but found $g1 and $g2"
+or echo "Pgroups disagreed. Should be in $global_group but found $g1, $g2, $g3"
 # CHECK: All pgroups agreed
 
 # Here everything should live in fish's pgroup.