diff --git a/include/swift/ABI/MetadataValues.h b/include/swift/ABI/MetadataValues.h
index ace93d8b4f9fe..f84b691ec272b 100644
--- a/include/swift/ABI/MetadataValues.h
+++ b/include/swift/ABI/MetadataValues.h
@@ -2334,6 +2334,7 @@ class TaskCreateFlags : public FlagSet<size_t> {
     Task_InheritContext                           = 11,
     Task_EnqueueJob                               = 12,
     Task_AddPendingGroupTaskUnconditionally       = 13,
+    Task_IsDiscardingTask                         = 14,
   };
 
   explicit constexpr TaskCreateFlags(size_t bits) : FlagSet(bits) {}
@@ -2360,6 +2361,9 @@ class TaskCreateFlags : public FlagSet<size_t> {
   FLAGSET_DEFINE_FLAG_ACCESSORS(Task_AddPendingGroupTaskUnconditionally,
                                 addPendingGroupTaskUnconditionally,
                                 setAddPendingGroupTaskUnconditionally)
+  FLAGSET_DEFINE_FLAG_ACCESSORS(Task_IsDiscardingTask,
+                                isDiscardingTask,
+                                setIsDiscardingTask)
 };
 
 /// Flags for schedulable jobs.
diff --git a/stdlib/public/Concurrency/DiscardingTaskGroup.swift b/stdlib/public/Concurrency/DiscardingTaskGroup.swift
index f2b531aafc35b..049a5692afa36 100644
--- a/stdlib/public/Concurrency/DiscardingTaskGroup.swift
+++ b/stdlib/public/Concurrency/DiscardingTaskGroup.swift
@@ -171,21 +171,21 @@ public struct DiscardingTaskGroup {
   #if SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
   @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model", renamed: "addTask(operation:)")
   #endif
-  public mutating func addTask<DiscardedResult>(
+  public mutating func addTask(
     priority: TaskPriority? = nil,
-    operation: __owned @Sendable @escaping () async -> DiscardedResult
+    operation: __owned @Sendable @escaping () async -> Void
   ) {
 #if SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: false,
-      addPendingGroupTaskUnconditionally: true
+      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true
     )
 #else
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true
+      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true
     )
 #endif
 
@@ -206,9 +206,9 @@ public struct DiscardingTaskGroup {
   #if SWIFT_STDLIB_TASK_TO_THREAD_MODEL_CONCURRENCY
   @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model", renamed: "addTask(operation:)")
   #endif
-  public mutating func addTaskUnlessCancelled<DiscardedResult>(
+  public mutating func addTaskUnlessCancelled(
     priority: TaskPriority? = nil,
-    operation: __owned @Sendable @escaping () async -> DiscardedResult
+    operation: __owned @Sendable @escaping () async -> Void
   ) -> Bool {
     let canAdd = _taskGroupAddPendingTask(group: _group, unconditionally: false)
 
@@ -220,13 +220,13 @@ public struct DiscardingTaskGroup {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: false,
-      addPendingGroupTaskUnconditionally: false
+      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true
     )
 #else
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
+      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true
     )
 #endif
 
@@ -237,13 +237,13 @@ public struct DiscardingTaskGroup {
   }
 
   @_alwaysEmitIntoClient
-  public mutating func addTask<DiscardedResult>(
-    operation: __owned @Sendable @escaping () async -> DiscardedResult
+  public mutating func addTask(
+    operation: __owned @Sendable @escaping () async -> Void
   ) {
     let flags = taskCreateFlags(
       priority: nil, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true
+      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true
     )
 
     // Create the task in this group.
@@ -260,8 +260,8 @@ public struct DiscardingTaskGroup {
   @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model", renamed: "addTaskUnlessCancelled(operation:)")
 #endif
   @_alwaysEmitIntoClient
-  public mutating func addTaskUnlessCancelled<DiscardedResult>(
-    operation: __owned @Sendable @escaping () async -> DiscardedResult
+  public mutating func addTaskUnlessCancelled(
+    operation: __owned @Sendable @escaping () async -> Void
   ) -> Bool {
 #if compiler(>=5.5) && $BuiltinCreateAsyncTaskInGroup
     let canAdd = _taskGroupAddPendingTask(group: _group, unconditionally: false)
@@ -274,7 +274,7 @@ public struct DiscardingTaskGroup {
     let flags = taskCreateFlags(
       priority: nil, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
+      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true
     )
 
     // Create the task in this group.
@@ -547,15 +547,15 @@ public struct ThrowingDiscardingTaskGroup<Failure: Error> {
   @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model", renamed: "addTask(operation:)")
 #endif
   @_alwaysEmitIntoClient
-  public mutating func addTask<DiscardedResult>(
+  public mutating func addTask(
     priority: TaskPriority? = nil,
-    operation: __owned @Sendable @escaping () async throws -> DiscardedResult
+    operation: __owned @Sendable @escaping () async throws -> Void
   ) {
 #if compiler(>=5.5) && $BuiltinCreateAsyncTaskInGroup
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true
+      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true
     )
 
     // Create the task in this group.
@@ -569,9 +569,9 @@ public struct ThrowingDiscardingTaskGroup<Failure: Error> {
   @available(*, unavailable, message: "Unavailable in task-to-thread concurrency model", renamed: "addTask(operation:)")
 #endif
   @_alwaysEmitIntoClient
-  public mutating func addTaskUnlessCancelled<DiscardedResult>(
+  public mutating func addTaskUnlessCancelled(
     priority: TaskPriority? = nil,
-    operation: __owned @Sendable @escaping () async throws -> DiscardedResult
+    operation: __owned @Sendable @escaping () async throws -> Void
   ) -> Bool {
 #if compiler(>=5.5) && $BuiltinCreateAsyncTaskInGroup
     let canAdd = _taskGroupAddPendingTask(group: _group, unconditionally: false)
@@ -584,7 +584,7 @@ public struct ThrowingDiscardingTaskGroup<Failure: Error> {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
+      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true
     )
 
     // Create the task in this group.
diff --git a/stdlib/public/Concurrency/Task.cpp b/stdlib/public/Concurrency/Task.cpp
index 6e4e515fd61c4..1d4d57d40b022 100644
--- a/stdlib/public/Concurrency/Task.cpp
+++ b/stdlib/public/Concurrency/Task.cpp
@@ -940,6 +940,7 @@ static AsyncTaskAndContext swift_task_create_commonImpl(
   // be is the final hop.  Store a signed null instead.
   initialContext->Parent = nullptr;
 
+  // FIXME: add discarding flag
   concurrency::trace::task_create(
       task, parent, group, asyncLet,
       static_cast<uint8_t>(task->Flags.getPriority()),
@@ -1060,21 +1061,40 @@ void swift::swift_task_run_inline(OpaqueValue *result, void *closureAFP,
 
 SWIFT_CC(swift)
 AsyncTaskAndContext swift::swift_task_create(
-    size_t taskCreateFlags,
+    size_t rawTaskCreateFlags,
     TaskOptionRecord *options,
     const Metadata *futureResultType,
     void *closureEntry, HeapObject *closureContext) {
-  FutureAsyncSignature::FunctionType *taskEntry;
-  size_t initialContextSize;
-  std::tie(taskEntry, initialContextSize) =
+  TaskCreateFlags taskCreateFlags(rawTaskCreateFlags);
+
+  if (taskCreateFlags.isDiscardingTask()) {
+    ThinNullaryAsyncSignature::FunctionType *taskEntry;
+    size_t initialContextSize;
+
+    std::tie(taskEntry, initialContextSize) =
       getAsyncClosureEntryPointAndContextSize<
-          FutureAsyncSignature,
-          SpecialPointerAuthDiscriminators::AsyncFutureFunction>(closureEntry);
+        ThinNullaryAsyncSignature,
+        SpecialPointerAuthDiscriminators::AsyncThinNullaryFunction>(closureEntry);
+
+    return swift_task_create_common(
+        rawTaskCreateFlags, options, futureResultType,
+        reinterpret_cast<TaskContinuationFunction *>(taskEntry), closureContext,
+        initialContextSize);
 
-  return swift_task_create_common(
-      taskCreateFlags, options, futureResultType,
-      reinterpret_cast<TaskContinuationFunction *>(taskEntry), closureContext,
-      initialContextSize);
+  } else {
+    FutureAsyncSignature::FunctionType *taskEntry;
+    size_t initialContextSize;
+
+    std::tie(taskEntry, initialContextSize) =
+        getAsyncClosureEntryPointAndContextSize<
+            FutureAsyncSignature,
+            SpecialPointerAuthDiscriminators::AsyncFutureFunction>(closureEntry);
+
+    return swift_task_create_common(
+        rawTaskCreateFlags, options, futureResultType,
+        reinterpret_cast<TaskContinuationFunction *>(taskEntry), closureContext,
+        initialContextSize);
+  }
 }
 
 #ifdef __ARM_ARCH_7K__
diff --git a/stdlib/public/Concurrency/Task.swift b/stdlib/public/Concurrency/Task.swift
index ef852e7ca79bc..6988cbde7363a 100644
--- a/stdlib/public/Concurrency/Task.swift
+++ b/stdlib/public/Concurrency/Task.swift
@@ -200,7 +200,8 @@ extension Task where Failure == Error {
         let flags = taskCreateFlags(priority: priority, isChildTask: false,
                                     copyTaskLocals: true, inheritContext: true,
                                     enqueueJob: false,
-                                    addPendingGroupTaskUnconditionally: false)
+                                    addPendingGroupTaskUnconditionally: false,
+                                    isDiscardingTask: false)
         let (task, _) = Builtin.createAsyncTask(flags, work)
         _startTaskOnMainActor(task)
         return Task<Success, Error>(task)
@@ -222,7 +223,8 @@ extension Task where Failure == Never {
         let flags = taskCreateFlags(priority: priority, isChildTask: false,
                                     copyTaskLocals: true, inheritContext: true,
                                     enqueueJob: false,
-                                    addPendingGroupTaskUnconditionally: false)
+                                    addPendingGroupTaskUnconditionally: false,
+                                    isDiscardingTask: false)
         let (task, _) = Builtin.createAsyncTask(flags, work)
         _startTaskOnMainActor(task)
         return Task(task)
@@ -503,7 +505,8 @@ struct JobFlags {
 func taskCreateFlags(
   priority: TaskPriority?, isChildTask: Bool, copyTaskLocals: Bool,
   inheritContext: Bool, enqueueJob: Bool,
-  addPendingGroupTaskUnconditionally: Bool
+  addPendingGroupTaskUnconditionally: Bool,
+  isDiscardingTask: Bool
 ) -> Int {
   var bits = 0
   bits |= (bits & ~0xFF) | Int(priority?.rawValue ?? 0)
@@ -522,6 +525,9 @@ func taskCreateFlags(
   if addPendingGroupTaskUnconditionally {
     bits |= 1 << 13
   }
+  if isDiscardingTask {
+    bits |= 1 << 14
+  }
   return bits
 }
 
@@ -573,7 +579,8 @@ extension Task where Failure == Never {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: false, copyTaskLocals: true,
       inheritContext: true, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false)
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 
     // Create the asynchronous task.
     let (task, _) = Builtin.createAsyncTask(flags, operation)
@@ -633,8 +640,8 @@ extension Task where Failure == Error {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: false, copyTaskLocals: true,
       inheritContext: true, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
-    )
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 
     // Create the asynchronous task future.
     let (task, _) = Builtin.createAsyncTask(flags, operation)
@@ -692,7 +699,8 @@ extension Task where Failure == Never {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: false, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false)
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 
     // Create the asynchronous task future.
     let (task, _) = Builtin.createAsyncTask(flags, operation)
@@ -751,8 +759,8 @@ extension Task where Failure == Error {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: false, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
-    )
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 
     // Create the asynchronous task future.
     let (task, _) = Builtin.createAsyncTask(flags, operation)
diff --git a/stdlib/public/Concurrency/TaskGroup.cpp b/stdlib/public/Concurrency/TaskGroup.cpp
index 67830e832e7b0..1395dc10716ea 100644
--- a/stdlib/public/Concurrency/TaskGroup.cpp
+++ b/stdlib/public/Concurrency/TaskGroup.cpp
@@ -228,7 +228,7 @@ class TaskGroupBase : public TaskGroupTaskStatusRecord {
                       reinterpret_cast<OpaqueValue *>(fragment->getError()) :
                       fragment->getStoragePtr(),
           /*successType=*/fragment->getResultType(),
-          /*task=*/asyncTask
+          /*retainedTask==*/asyncTask
       };
     }
 
@@ -390,7 +390,11 @@ class TaskGroupBase : public TaskGroupTaskStatusRecord {
   virtual void enqueueCompletedTask(AsyncTask *completedTask, bool hadErrorResult) = 0;
 
   /// Resume waiting task with result from `completedTask`
-  void resumeWaitingTask(AsyncTask *completedTask, TaskGroupStatus &assumed, bool hadErrorResult, bool alreadyDecremented = false);
+  void resumeWaitingTask(AsyncTask *completedTask,
+                         TaskGroupStatus &assumed,
+                         bool hadErrorResult,
+                         bool alreadyDecremented = false,
+                         bool taskWasRetained = false);
 
   // ==== Status manipulation -------------------------------------------------
 
@@ -827,7 +831,9 @@ class DiscardingTaskGroup: public TaskGroupBase {
 
 private:
   /// Resume waiting task with specified error
-  void resumeWaitingTaskWithError(SwiftError *error, TaskGroupStatus &assumed, bool alreadyDecremented);
+  void resumeWaitingTaskWithError(SwiftError *error,
+                                  TaskGroupStatus &assumed,
+                                  bool alreadyDecremented);
 };
 
 } // end anonymous namespace
@@ -1146,7 +1152,9 @@ void AccumulatingTaskGroup::offer(AsyncTask *completedTask, AsyncContext *contex
   // will need to release in the other path.
   lock(); // TODO: remove fragment lock, and use status for synchronization
 
-  SWIFT_TASK_GROUP_DEBUG_LOG(this, "offer, completedTask:%p , status:%s", completedTask, statusString().c_str());
+  SWIFT_TASK_GROUP_DEBUG_LOG(this, "offer, completedTask:%p, status:%s",
+                             completedTask,
+                             statusString().c_str());
 
   // Immediately increment ready count and acquire the status
   //
@@ -1218,7 +1226,6 @@ void DiscardingTaskGroup::offer(AsyncTask *completedTask, AsyncContext *context)
   // We can do this, since in this mode there is no ready count to keep track of,
   // and we immediately discard the result.
   auto afterComplete = statusCompletePendingAssumeRelease();
-  (void) afterComplete;
   const bool alreadyDecrementedStatus = true;
   SWIFT_TASK_GROUP_DEBUG_LOG(this, "offer, complete, status afterComplete:%s", afterComplete.to_string(this).c_str());
 
@@ -1238,11 +1245,15 @@ void DiscardingTaskGroup::offer(AsyncTask *completedTask, AsyncContext *context)
         switch (readyErrorItem.getStatus()) {
           case ReadyStatus::RawError:
             SWIFT_TASK_GROUP_DEBUG_LOG(this, "offer, complete, resume with raw error:%p", readyErrorItem.getRawError(this));
-            resumeWaitingTaskWithError(readyErrorItem.getRawError(this), assumed, alreadyDecrementedStatus);
+            resumeWaitingTaskWithError(readyErrorItem.getRawError(this), assumed,
+                                       alreadyDecrementedStatus);
             break;
           case ReadyStatus::Error:
             SWIFT_TASK_GROUP_DEBUG_LOG(this, "offer, complete, resume with errorItem.task:%p", readyErrorItem.getTask());
-            resumeWaitingTask(readyErrorItem.getTask(), assumed, /*hadErrorResult=*/true, alreadyDecrementedStatus);
+            resumeWaitingTask(readyErrorItem.getTask(), assumed,
+                              /*hadErrorResult=*/true,
+                              alreadyDecrementedStatus,
+                              /*taskWasRetained=*/true);
             break;
           default:
             swift_Concurrency_fatalError(0,
@@ -1281,7 +1292,10 @@ void DiscardingTaskGroup::offer(AsyncTask *completedTask, AsyncContext *context)
           resumeWaitingTaskWithError(readyErrorItem.getRawError(this), assumed, alreadyDecrementedStatus);
           break;
         case ReadyStatus::Error:
-          resumeWaitingTask(readyErrorItem.getTask(), assumed, /*hadErrorResult=*/true, alreadyDecrementedStatus);
+          resumeWaitingTask(readyErrorItem.getTask(), assumed,
+                            /*hadErrorResult=*/true,
+                            alreadyDecrementedStatus,
+                            /*taskWasRetained=*/true);
           break;
         default:
           swift_Concurrency_fatalError(0,
@@ -1307,7 +1321,8 @@ void TaskGroupBase::resumeWaitingTask(
     AsyncTask *completedTask,
     TaskGroupStatus &assumed,
     bool hadErrorResult,
-    bool alreadyDecremented) {
+    bool alreadyDecremented,
+    bool taskWasRetained) {
   auto waitingTask = waitQueue.load(std::memory_order_acquire);
   assert(waitingTask && "waitingTask must not be null when attempting to resume it");
   assert(assumed.hasWaitingTask());
@@ -1351,13 +1366,12 @@ void TaskGroupBase::resumeWaitingTask(
         // Run the task.
         auto result = PollResult::get(completedTask, hadErrorResult);
         SWIFT_TASK_GROUP_DEBUG_LOG(this,
-                                   "resume waiting DONE, task = %p, backup = %p, error:%d, complete with = %p, status = %s",
-                                   waitingTask, backup, hadErrorResult, completedTask, statusString().c_str());
+                                   "resume waiting DONE, task = %p, error:%d, complete with = %p, status = %s",
+                                   waitingTask, hadErrorResult, completedTask, statusString().c_str());
 
         auto waitingContext =
             static_cast<TaskFutureWaitAsyncContext *>(
                 waitingTask->ResumeContext);
-
         fillGroupNextResult(waitingContext, result);
 
         // Remove the child from the task group's running tasks list.
@@ -1369,6 +1383,11 @@ void TaskGroupBase::resumeWaitingTask(
         // locks) because we know that the child task is completed and
         // we can't be holding its locks ourselves.
         _swift_taskGroup_detachChild(asAbstract(this), completedTask);
+        if (isDiscardingResults() && hadErrorResult && taskWasRetained) {
+          // We only used the task to keep the error in the future fragment around
+          // so now that we emitted the error and detached the task, we are free to release the task immediately.
+          swift_release(completedTask);
+        }
 
         _swift_tsan_acquire(static_cast<Job *>(waitingTask));
         // TODO: allow the caller to suggest an executor
@@ -1377,7 +1396,7 @@ void TaskGroupBase::resumeWaitingTask(
 #endif /* SWIFT_CONCURRENCY_TASK_TO_THREAD_MODEL */
     } else {
       SWIFT_TASK_GROUP_DEBUG_LOG(this, "CAS failed, task = %p, backup = %p, complete with = %p, status = %s",
-                                 waitingTask, backup, completedTask, statusString().c_str());
+                                 waitingTask, completedTask, statusString().c_str());
     }
   }
 }
diff --git a/stdlib/public/Concurrency/TaskGroup.swift b/stdlib/public/Concurrency/TaskGroup.swift
index c8a7930d27f17..35414ec9d2072 100644
--- a/stdlib/public/Concurrency/TaskGroup.swift
+++ b/stdlib/public/Concurrency/TaskGroup.swift
@@ -272,14 +272,15 @@ public struct TaskGroup<ChildTaskResult: Sendable> {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: false,
-      addPendingGroupTaskUnconditionally: true
+      addPendingGroupTaskUnconditionally: true,
+      isDiscardingTask: false
     )
 #else
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true
-    )
+      addPendingGroupTaskUnconditionally: true,
+      isDiscardingTask: false)
 #endif
 
     // Create the task in this group.
@@ -314,14 +315,14 @@ public struct TaskGroup<ChildTaskResult: Sendable> {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: false,
-      addPendingGroupTaskUnconditionally: false
-    )
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 #else
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
-    )
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 #endif
 
     // Create the task in this group.
@@ -354,8 +355,8 @@ public struct TaskGroup<ChildTaskResult: Sendable> {
     let flags = taskCreateFlags(
       priority: nil, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true
-    )
+      addPendingGroupTaskUnconditionally: true,
+      isDiscardingTask: false)
 
     // Create the task in this group.
     _ = Builtin.createAsyncTaskInGroup(flags, _group, operation)
@@ -394,8 +395,8 @@ public struct TaskGroup<ChildTaskResult: Sendable> {
     let flags = taskCreateFlags(
       priority: nil, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
-    )
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 
     // Create the task in this group.
     _ = Builtin.createAsyncTaskInGroup(flags, _group, operation)
@@ -682,7 +683,8 @@ public struct ThrowingTaskGroup<ChildTaskResult: Sendable, Failure: Error> {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true
+      addPendingGroupTaskUnconditionally: true,
+      isDiscardingTask: false
     )
 
     // Create the task in this group.
@@ -720,8 +722,8 @@ public struct ThrowingTaskGroup<ChildTaskResult: Sendable, Failure: Error> {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
-    )
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 
     // Create the task in this group.
     _ = Builtin.createAsyncTaskInGroup(flags, _group, operation)
@@ -756,8 +758,8 @@ public struct ThrowingTaskGroup<ChildTaskResult: Sendable, Failure: Error> {
     let flags = taskCreateFlags(
       priority: nil, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true
-    )
+      addPendingGroupTaskUnconditionally: true,
+      isDiscardingTask: false)
 
     // Create the task in this group.
     _ = Builtin.createAsyncTaskInGroup(flags, _group, operation)
@@ -799,8 +801,8 @@ public struct ThrowingTaskGroup<ChildTaskResult: Sendable, Failure: Error> {
     let flags = taskCreateFlags(
       priority: nil, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false
-    )
+      addPendingGroupTaskUnconditionally: false,
+      isDiscardingTask: false)
 
     // Create the task in this group.
     _ = Builtin.createAsyncTaskInGroup(flags, _group, operation)
diff --git a/stdlib/public/Concurrency/TaskSleep.swift b/stdlib/public/Concurrency/TaskSleep.swift
index 043d4a3d31b9d..c84f5ba2f1b0d 100644
--- a/stdlib/public/Concurrency/TaskSleep.swift
+++ b/stdlib/public/Concurrency/TaskSleep.swift
@@ -240,7 +240,8 @@ extension Task where Success == Never, Failure == Never {
               let sleepTaskFlags = taskCreateFlags(
                 priority: nil, isChildTask: false, copyTaskLocals: false,
                 inheritContext: false, enqueueJob: false,
-                addPendingGroupTaskUnconditionally: false)
+                addPendingGroupTaskUnconditionally: false,
+                isDiscardingTask: false)
               let (sleepTask, _) = Builtin.createAsyncTask(sleepTaskFlags) {
                 onSleepWake(wordPtr)
               }
diff --git a/stdlib/public/Concurrency/TaskSleepDuration.swift b/stdlib/public/Concurrency/TaskSleepDuration.swift
index 8981d91fa7819..5431926561937 100644
--- a/stdlib/public/Concurrency/TaskSleepDuration.swift
+++ b/stdlib/public/Concurrency/TaskSleepDuration.swift
@@ -58,7 +58,8 @@ extension Task where Success == Never, Failure == Never {
               let sleepTaskFlags = taskCreateFlags(
                 priority: nil, isChildTask: false, copyTaskLocals: false,
                 inheritContext: false, enqueueJob: false,
-                addPendingGroupTaskUnconditionally: false)
+                addPendingGroupTaskUnconditionally: false,
+                isDiscardingTask: false)
               let (sleepTask, _) = Builtin.createAsyncTask(sleepTaskFlags) {
                 onSleepWake(wordPtr)
               }
diff --git a/stdlib/toolchain/Compatibility56/include/Runtime/Concurrency.h b/stdlib/toolchain/Compatibility56/include/Runtime/Concurrency.h
index 5ba9fe3fd9669..eff5b075219f7 100644
--- a/stdlib/toolchain/Compatibility56/include/Runtime/Concurrency.h
+++ b/stdlib/toolchain/Compatibility56/include/Runtime/Concurrency.h
@@ -51,7 +51,6 @@ struct AsyncTaskAndContext {
   AsyncContext *InitialContext;
 };
 
-
 /// Caution: not all future-initializing functions actually throw, so
 /// this signature may be incorrect.
 using FutureAsyncSignature =
diff --git a/test/Concurrency/Runtime/async_taskgroup_discarding.swift b/test/Concurrency/Runtime/async_taskgroup_discarding.swift
index e1bdf17575c9d..2557a875d8113 100644
--- a/test/Concurrency/Runtime/async_taskgroup_discarding.swift
+++ b/test/Concurrency/Runtime/async_taskgroup_discarding.swift
@@ -1,4 +1,5 @@
-// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking -parse-as-library)
+// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking -parse-as-library) | %FileCheck %s
+// TODO: move to target-run-simple-leaks-swift once CI is using at least Xcode 14.3
 
 // REQUIRES: concurrency
 // REQUIRES: executable_test
@@ -12,28 +13,32 @@
 
 import _Concurrency
 
-struct Boom: Error {
-  let id: String
+final class PayloadFirst {}
+final class PayloadSecond {}
+
+final class ErrorFirst: Error {
+  let first: PayloadFirst
 
   init(file: String = #fileID, line: UInt = #line) {
-    self.id = "\(file):\(line)"
+    first = .init()
   }
-
-  init(id: String) {
-    self.id = id
+  deinit {
+    print("deinit \(self)")
   }
 }
 
-struct IgnoredBoom: Error {}
+final class ErrorSecond: Error {
+  let second: PayloadSecond
 
-@discardableResult
-func echo(_ i: Int) -> Int { i }
-@discardableResult
-func boom(file: String = #fileID, line: UInt = #line) throws -> Int { throw Boom(file: file, line: line) }
+  init(file: String = #fileID, line: UInt = #line) {
+    second = .init()
+  }
 
-func shouldEqual<T: Equatable>(_ lhs: T, _ rhs: T) {
-  precondition(lhs == rhs, "'\(lhs)' did not equal '\(rhs)'")
+  deinit {
+    print("deinit \(self)")
+  }
 }
+
 func shouldStartWith(_ lhs: Any, _ rhs: Any) {
   let l = "\(lhs)"
   let r = "\(rhs)"
@@ -44,22 +49,26 @@ func shouldStartWith(_ lhs: Any, _ rhs: Any) {
 
 @main struct Main {
   static func main() async {
-    for i in 0...1_000 {
-      do {
-        let got = try await withThrowingDiscardingTaskGroup() { group in
-          group.addTask {
-            echo(1)
-          }
-          group.addTask {
-            try boom()
-          }
+    do {
+      let got = try await withThrowingDiscardingTaskGroup() { group in
+        group.addTask {
+          1
+        }
+        group.addTask {
+          throw ErrorFirst()
+        }
 
-          return 12
+        group.addTask {
+          throw ErrorSecond()
         }
-        fatalError("(iteration:\(i)) expected error to be re-thrown, got: \(got)")
-      } catch {
-        shouldStartWith(error, "Boom(")
+
+        return 12
       }
+      fatalError("expected error to be re-thrown, got: \(got)")
+    } catch {
+       shouldStartWith(error, "main.Error")
     }
+    // CHECK: deinit main.Error
+    // CHECK: deinit main.Error
   }
 }
diff --git a/test/Concurrency/Runtime/async_taskgroup_discarding_dontLeak.swift b/test/Concurrency/Runtime/async_taskgroup_discarding_dontLeak.swift
new file mode 100644
index 0000000000000..f41d29adf58b5
--- /dev/null
+++ b/test/Concurrency/Runtime/async_taskgroup_discarding_dontLeak.swift
@@ -0,0 +1,152 @@
+// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking -parse-as-library) | %FileCheck %s --dump-input=always
+// TODO: move to target-run-simple-leaks-swift once CI is using at least Xcode 14.3
+
+// REQUIRES: concurrency
+// REQUIRES: executable_test
+// REQUIRES: concurrency_runtime
+
+// UNSUPPORTED: back_deployment_runtime
+// UNSUPPORTED: freestanding
+
+// FIXME: enable discarding taskgroup on windows; rdar://104762037
+// UNSUPPORTED: OS=windows-msvc
+
+import _Concurrency
+
+final class PrintDeinit {
+  let id: String
+  init(id: String) {
+    self.id = id
+  }
+
+  deinit {
+    print("deinit, id: \(id)")
+  }
+}
+
+struct Boom: Error {
+  let printDeinit: PrintDeinit
+
+  init(id: String) {
+    self.printDeinit = PrintDeinit(id: id)
+  }
+}
+
+final class BoomClass: Error {
+  let id: String
+
+  init(id: String) {
+    self.id = id
+  }
+
+  deinit {
+    print("deinit, id: \(id)")
+  }
+}
+
+final class SomeClass: @unchecked Sendable {
+  let id: String
+  init(id: String) {
+    self.id = id
+  }
+
+  deinit {
+    print("deinit, id: \(id)")
+  }
+}
+
+// NOTE: Not as StdlibUnittest/TestSuite since these types of tests are unreasonably slow to load/debug.
+
+@main struct Main {
+  static func main() async {
+    _ = try? await withThrowingDiscardingTaskGroup() { group in
+      group.addTask {
+        throw Boom(id: "race-boom-class")
+      }
+      group.addTask {
+        SomeClass(id: "race-boom-class") // will be discarded
+      }
+      // since values may deinit in any order, we just assert their count basically
+      // CHECK: deinit, id: race-boom-class
+      // CHECK: deinit, id: race-boom-class
+
+      return 12
+    }
+
+    // many ok
+    _ = try? await withThrowingDiscardingTaskGroup() { group in
+      for i in 0..<6 {
+        group.addTask {
+          SomeClass(id: "many-ok") // will be discarded
+        }
+        // since values may deinit in any order, we just assert their count basically
+        // CHECK: deinit, id: many-ok
+        // CHECK: deinit, id: many-ok
+        // CHECK: deinit, id: many-ok
+        // CHECK: deinit, id: many-ok
+        // CHECK: deinit, id: many-ok
+        // CHECK: deinit, id: many-ok
+      }
+
+      return 12
+    }
+
+    // many throws
+    do {
+      let value = try await withThrowingDiscardingTaskGroup() { group in
+        for i in 0..<6 {
+          group.addTask {
+            throw BoomClass(id: "many-error") // will be rethrown
+          }
+        }
+
+        // since values may deinit in any order, we just assert their count basically
+        // CHECK: deinit, id: many-error
+        // CHECK: deinit, id: many-error
+        // CHECK: deinit, id: many-error
+        // CHECK: deinit, id: many-error
+        // CHECK: deinit, id: many-error
+        // CHECK: deinit, id: many-error
+
+        12 // must be ignored
+      }
+      preconditionFailure("Should throw")
+    } catch {
+      precondition("\(error)" == "main.BoomClass", "error was: \(error)")
+    }
+
+    // many errors, many values
+    _ = try? await withThrowingDiscardingTaskGroup() { group in
+      group.addTask {
+        SomeClass(id: "mixed-ok") // will be discarded
+      }
+      group.addTask {
+        SomeClass(id: "mixed-ok") // will be discarded
+      }
+      group.addTask {
+        SomeClass(id: "mixed-ok") // will be discarded
+      }
+      group.addTask {
+        throw Boom(id: "mixed-error")
+      }
+      group.addTask {
+        throw Boom(id: "mixed-error")
+      }
+      group.addTask {
+        throw Boom(id: "mixed-error")
+      }
+
+      // since values may deinit in any order, we just assert their count basically
+      // three ok's
+      // CHECK: deinit, id: mixed
+      // CHECK: deinit, id: mixed
+      // CHECK: deinit, id: mixed
+      // three errors
+      // CHECK: deinit, id: mixed
+      // CHECK: deinit, id: mixed
+      // CHECK: deinit, id: mixed
+
+      return 12
+    }
+  }
+}
diff --git a/test/Concurrency/Runtime/async_taskgroup_discarding_dontLeak_class_error.swift b/test/Concurrency/Runtime/async_taskgroup_discarding_dontLeak_class_error.swift
new file mode 100644
index 0000000000000..8227662a7b375
--- /dev/null
+++ b/test/Concurrency/Runtime/async_taskgroup_discarding_dontLeak_class_error.swift
@@ -0,0 +1,46 @@
+// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking -parse-as-library) | %FileCheck %s --dump-input=always
+
+// REQUIRES: concurrency
+// REQUIRES: executable_test
+// REQUIRES: concurrency_runtime
+
+// UNSUPPORTED: back_deployment_runtime
+// UNSUPPORTED: freestanding
+
+// FIXME: enable discarding taskgroup on windows; rdar://104762037
+// UNSUPPORTED: OS=windows-msvc
+
+import _Concurrency
+
+final class ClassBoom: Error {
+  let id: String
+
+  init(file: String = #fileID, line: UInt = #line) {
+    self.id = "\(file):\(line)"
+    print("INIT OF ClassBoom from \(id)")
+  }
+
+  deinit {
+    print("DEINIT OF ClassBoom from \(id)")
+  }
+}
+
+@main struct Main {
+  static func main() async {
+
+    // many errors
+    _ = try? await withThrowingDiscardingTaskGroup() { group in
+      group.addTask { throw ClassBoom() }
+      group.addTask { throw ClassBoom() }
+      group.addTask { throw ClassBoom() }
+      group.addTask { throw ClassBoom() }
+      group.addTask { 12 }
+      return 12
+
+      // CHECK: DEINIT OF ClassBoom
+      // CHECK: DEINIT OF ClassBoom
+      // CHECK: DEINIT OF ClassBoom
+      // CHECK: DEINIT OF ClassBoom
+    }
+  }
+}
diff --git a/test/Concurrency/Runtime/async_taskgroup_dontLeakTasks.swift b/test/Concurrency/Runtime/async_taskgroup_dontLeakTasks.swift
index c31397e782ad9..1386a1e760242 100644
--- a/test/Concurrency/Runtime/async_taskgroup_dontLeakTasks.swift
+++ b/test/Concurrency/Runtime/async_taskgroup_dontLeakTasks.swift
@@ -1,52 +1,51 @@
-// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking -parse-as-library) 2>&1 | %FileCheck %s --dump-input=always
+// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking -parse-as-library) | %FileCheck %s
+// TODO: move to target-run-simple-leaks-swift once CI is using at least Xcode 14.3
+
+// This test uses `leaks` which is only available on apple platforms; limit it to macOS:
+// REQUIRES: OS=macosx
+
 // REQUIRES: executable_test
 // REQUIRES: concurrency
-// REQUIRES: swift_task_debug_log
 
 // REQUIRES: concurrency_runtime
 // UNSUPPORTED: back_deployment_runtime
 
-#if os(Linux)
-import Glibc
-#elseif os(Windows)
-import MSVCRT
-#else
 import Darwin
-#endif
+
+final class Something {
+  let int: Int
+  init(int: Int) {
+    self.int = int
+  }
+
+  deinit {
+    print("deinit, Something, int: \(int)")
+  }
+}
 
 func test_taskGroup_next() async {
-  // CHECK: creating task [[MAIN_TASK:0x.*]] with parent 0x0
-  // CHECK: creating task [[GROUP_TASK_1:0x.*]] with parent [[MAIN_TASK]]
-  // CHECK: creating task [[GROUP_TASK_2:0x.*]] with parent [[MAIN_TASK]]
-  // CHECK: creating task [[GROUP_TASK_3:0x.*]] with parent [[MAIN_TASK]]
-  // CHECK: creating task [[GROUP_TASK_4:0x.*]] with parent [[MAIN_TASK]]
-  // CHECK: creating task [[GROUP_TASK_5:0x.*]] with parent [[MAIN_TASK]]
-
-  _ = await withTaskGroup(of: Int.self, returning: Int.self) { group in
-    for n in 0..<5 {
+  let tasks = 5
+  _ = await withTaskGroup(of: Something.self, returning: Int.self) { group in
+    for n in 0..<tasks {
       group.spawn {
-        return n
+        Something(int: n)
       }
     }
-    await Task.sleep(2_000_000)
 
     var sum = 0
     for await value in group {
-      sum += 1
+      sum += value.int
     }
 
+
+    // CHECK-DAG: deinit, Something, int: 0
+    // CHECK-DAG: deinit, Something, int: 1
+    // CHECK-DAG: deinit, Something, int: 2
+    // CHECK-DAG: deinit, Something, int: 3
+    // CHECK-DAG: deinit, Something, int: 4
+
     return sum
   }
-  // as we exit the group, it must be guaranteed that its child tasks were destroyed
-  //
-  // NOTE: there is no great way to express "any of GROUP_TASK_n",
-  //       so we just check that 5 tasks were destroyed
-  //
-  // CHECK: destroy task [[DESTROY_GROUP_TASK_1:0x.*]]
-  // CHECK: destroy task [[DESTROY_GROUP_TASK_2:0x.*]]
-  // CHECK: destroy task [[DESTROY_GROUP_TASK_3:0x.*]]
-  // CHECK: destroy task [[DESTROY_GROUP_TASK_4:0x.*]]
-  // CHECK: destroy task [[DESTROY_GROUP_TASK_5:0x.*]]
 }
 
 @main struct Main {
diff --git a/test/Concurrency/Runtime/async_taskgroup_throw_recover.swift b/test/Concurrency/Runtime/async_taskgroup_throw_recover.swift
index f428b50d1ef47..28a741955582f 100644
--- a/test/Concurrency/Runtime/async_taskgroup_throw_recover.swift
+++ b/test/Concurrency/Runtime/async_taskgroup_throw_recover.swift
@@ -1,4 +1,4 @@
-// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking -parse-as-library) | %FileCheck %s --dump-input=always
+// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking -parse-as-library) | %FileCheck %s
 
 // REQUIRES: executable_test
 // REQUIRES: concurrency
diff --git a/test/lit.cfg b/test/lit.cfg
index f79b94e31f63e..83a45c4ee5419 100644
--- a/test/lit.cfg
+++ b/test/lit.cfg
@@ -1006,7 +1006,7 @@ else:
 swift_reflection_test_name = 'swift-reflection-test' + variant_suffix
 
 def use_interpreter_for_simple_runs():
-    def make_simple_target_run(gyb=False, stdlib=False, parameterized=False):
+    def make_simple_target_run(gyb=False, stdlib=False, parameterized=False, leaks=False):
         result = ''
         if gyb:
             result += ('%empty-directory(%t) && '
@@ -1034,6 +1034,7 @@ def use_interpreter_for_simple_runs():
     config.target_run_stdlib_swift = make_simple_target_run(stdlib=True)
     config.target_run_simple_swift = make_simple_target_run()
     config.target_run_simple_swift_parameterized = make_simple_target_run(parameterized=True)
+    config.target_run_simple_leaks_swift_parameterized = make_simple_leaks_target_run(parameterized=True)
     config.target_run_stdlib_swift_parameterized = make_simple_target_run(stdlib=True, parameterized=True)
     config.target_run_simple_swiftgyb_parameterized = make_simple_target_run(gyb=True, parameterized=True)
     config.available_features.add('interpret')
@@ -2272,6 +2273,11 @@ elif not kIsWindows:
         lit_config.note('Testing with the just-built libraries')
 
     lit_config.note('Library load path: {0}'.format(os.path.pathsep.join(target_stdlib_path)))
+    config.target_run_with_leaks = (
+        "/usr/bin/env " +
+        construct_library_path_env(target_stdlib_path) +
+        " xcrun leaks -atExit -- " +
+        config.target_run)
     config.target_run = (
         "/usr/bin/env " +
         construct_library_path_env(target_stdlib_path) +
@@ -2307,6 +2313,17 @@ if not getattr(config, 'target_run_simple_swift', None):
            escape_for_substitute_captures(config.target_codesign),
            escape_for_substitute_captures(config.target_run))
     )
+    if not kIsWindows:
+        config.target_run_simple_leaks_swift_parameterized = SubstituteCaptures(
+            r"%%empty-directory(%%t) && "
+            r"%s %s %%s \1 -o %%t/a.out -module-name main  && "
+            r"%s %%t/a.out && "
+            r"%s %%t/a.out"
+            % (escape_for_substitute_captures(config.target_build_swift),
+               escape_for_substitute_captures(mcp_opt),
+               escape_for_substitute_captures(config.target_codesign),
+               escape_for_substitute_captures(config.target_run_with_leaks))
+        )
     config.target_fail_simple_swift_parameterized = SubstituteCaptures(
         r"%%empty-directory(%%t) && "
         r"%s %s %%s \1 -o %%t/a.out -module-name main  && "
@@ -2482,6 +2499,9 @@ config.substitutions.append(('%target-run-simple-swiftgyb\(([^)]+)\)',
 config.substitutions.append(('%target-run-simple-swiftgyb', config.target_run_simple_swiftgyb))
 config.substitutions.append(('%target-run-simple-swift\(([^)]+)\)',
                             config.target_run_simple_swift_parameterized))
+if not kIsWindows:
+    config.substitutions.append(('%target-run-simple-leaks-swift\(([^)]+)\)',
+                                config.target_run_simple_leaks_swift_parameterized))
 config.substitutions.append(('%target-fail-simple-swift\(([^)]+)\)',
                             config.target_fail_simple_swift_parameterized))
 config.substitutions.append(('%target-run-stdlib-swift\(([^)]+)\)',