diff --git a/meta/lib/oeqa/core/tests/cases/streams.py b/meta/lib/oeqa/core/tests/cases/streams.py
new file mode 100644
index 00000000000..f165e7124a1
--- /dev/null
+++ b/meta/lib/oeqa/core/tests/cases/streams.py
@@ -0,0 +1,24 @@
+#
+# Copyright OpenEmbedded Contributors
+#
+# SPDX-License-Identifier: MIT
+#
+
+import sys
+
+from oeqa.core.case import OETestCase
+
+class StreamsTest(OETestCase):
+
+    def testStreams1(self):
+        print("Output from testStreams1")
+        sys.stderr.write("Error output from testStreams1\n")
+
+    def testStreams2(self):
+        print("Output from testStreams2")
+        sys.stderr.write("Error output from testStreams2\n")
+        self.fail("Failed on purpose")
+
+    def testStreams3(self):
+        print("Output from testStreams3")
+        self.fail("Failed on purpose")
diff --git a/meta/lib/oeqa/core/tests/test_concurrency.py b/meta/lib/oeqa/core/tests/test_concurrency.py
index 16e6619ad7f..b44c006d532 100644
--- a/meta/lib/oeqa/core/tests/test_concurrency.py
+++ b/meta/lib/oeqa/core/tests/test_concurrency.py
@@ -19,7 +19,7 @@ except ImportError:
 
 @unittest.skipIf(ConcurrentTestSuite is None, "concurrencytest needs testtools, subunit and bitbake")
 class TestConcurrency(TestBase):
-    def _statuses(self, modules, processes=None):
+    def _run(self, modules, processes=None):
         tc = self._testLoader(modules=modules)
         if processes:
             setup = lambda suffix, selftestdir, suite: (None, None)
@@ -27,6 +27,10 @@ class TestConcurrency(TestBase):
         # fork_for_tests() only looks up the selftest layer with BUILDDIR set
         with mock.patch.dict(os.environ, {'BUILDDIR': '/nonexistent'}):
             result = tc.runTests(processes=processes)
+        return tc, result
+
+    def _statuses(self, modules, processes=None):
+        tc, result = self._run(modules, processes)
         return {case_id: result._getTestResultDetails(case)[0]
                 for case_id, case in tc._registry['cases'].items()}
 
@@ -38,5 +42,16 @@ class TestConcurrency(TestBase):
         self.assertNotIn('UNKNOWN', sequential.values())
         self.assertEqual(self._statuses(modules, processes=1), sequential)
 
+    def test_concurrent_output(self):
+        # Each test's results must only have the output from that test, not
+        # everything its process has output so far
+        tc, result = self._run(['streams'], processes=1)
+        for n in (1, 2, 3):
+            stdout, stderr = result.logged_output['streams.StreamsTest.testStreams%d' % n]
+            output = stdout + stderr
+            self.assertIn("Output from testStreams%d" % n, output)
+            for other in {1, 2, 3} - {n}:
+                self.assertNotIn("testStreams%d" % other, output)
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/meta/lib/oeqa/core/utils/concurrencytest.py b/meta/lib/oeqa/core/utils/concurrencytest.py
index 09d621446ee..455b70faff2 100644
--- a/meta/lib/oeqa/core/utils/concurrencytest.py
+++ b/meta/lib/oeqa/core/utils/concurrencytest.py
@@ -57,6 +57,9 @@ class BBThreadsafeForwardingResult(ThreadsafeForwardingResult):
         self.totaltests = totaltests
         self.buffer = True
         self.outputbuf = output
+        # The process' output accumulates in outputbuf; this is how much of
+        # it was given to earlier results
+        self.outputoffset = 0
         self.finalresult = finalresult
         self.finalresult.buffer = True
         self.target = target
@@ -79,8 +82,13 @@ class BBThreadsafeForwardingResult(ThreadsafeForwardingResult):
                     test.id())
         finally:
             self.semaphore.release()
-        self.finalresult._stderr_buffer = io.StringIO(initial_value=self.outputbuf.getvalue().decode("utf-8"))
+        # The process runs its tests one at a time, so the output since the
+        # previous result is this test's. stdout and stderr share one pipe so
+        # it is all given as stderr.
+        output = self.outputbuf.getvalue()
+        self.finalresult._stderr_buffer = io.StringIO(initial_value=output[self.outputoffset:].decode("utf-8"))
         self.finalresult._stdout_buffer = io.StringIO()
+        self.outputoffset = len(output)
         super()._add_result_with_semaphore(method, test, *args, **kwargs)
 
 class ProxyTestResult: