diff --git a/doc/neotest.txt b/doc/neotest.txt
index 6b6530f4..e3d4adac 100644
--- a/doc/neotest.txt
+++ b/doc/neotest.txt
@@ -76,7 +76,7 @@ neotest.setup({user_config})                                 *neotest.setup()*
           skipped = "NeotestSkipped",
           target = "NeotestTarget",
           test = "NeotestTest",
-          unknown = "NeotestUnknown",
+          unknown = "NeotestUnknown"
         },
         icons = {
           child_indent = "│",
diff --git a/lua/neotest/async.lua b/lua/neotest/async.lua
index becbed23..81cc55b9 100644
--- a/lua/neotest/async.lua
+++ b/lua/neotest/async.lua
@@ -18,6 +18,25 @@ end
 local async_wrapper = {
   api = proxy_vim("api"),
   fn = proxy_vim("fn"),
+  lib = {
+    first = function(...)
+      local functions = { ... }
+      local send_ran, await_ran = plen_async.control.channel.oneshot()
+      local result, ran
+      for _, func in ipairs(functions) do
+        plen_async.run(function()
+          local func_result = func()
+          if not ran then
+            result = func_result
+            ran = true
+            send_ran()
+          end
+        end)
+      end
+      await_ran()
+      return result
+    end,
+  },
 }
 if false then
   -- For type checking
diff --git a/lua/neotest/client/init.lua b/lua/neotest/client/init.lua
index c30aebd8..5fa9daed 100644
--- a/lua/neotest/client/init.lua
+++ b/lua/neotest/client/init.lua
@@ -55,33 +55,24 @@ function NeotestClient:run_tree(tree, args)
     return
   end
   self._state:update_running(adapter_id, root.id, pos_ids)
-  local all_results = {}
-  local success, error = pcall(
+  local success, all_results = pcall(
     self._runner.run_tree,
     self._runner,
     tree,
     args,
     adapter,
     function(results)
-      for pos_id, result in pairs(results) do
-        all_results[pos_id] = result
-      end
-      self._state:update_results(adapter_id, results)
+      self._state:update_results(adapter_id, results, true)
     end
   )
   if not success then
-    lib.notify(("%s: %s"):format(adapter.name, error), "warn")
+    lib.notify(("%s: %s"):format(adapter.name, all_results), "warn")
     all_results = {}
     for _, pos in tree:iter() do
       all_results[pos.id] = { status = "skipped" }
     end
   end
-  if root.type ~= "test" then
-    self._runner:fill_results(tree, all_results)
-  end
-  if root.type == "test" or root.type == "namespace" then
-    all_results[root.path] = nil
-  end
+
   self._state:update_results(adapter_id, all_results)
 end
 
@@ -299,7 +290,7 @@ function NeotestClient:_get_adapter(position_id, adapter_id, refresh)
   else
     for _, adapter in ipairs(self._adapters) do
       local root = self._state:positions(adapter.name)
-      if vim.startswith(position_id, root:data().path) then
+      if root and vim.startswith(position_id, root:data().path) then
         return adapter.name, adapter
       end
     end
diff --git a/lua/neotest/client/runner.lua b/lua/neotest/client/runner.lua
index fe880285..a35ef08d 100644
--- a/lua/neotest/client/runner.lua
+++ b/lua/neotest/client/runner.lua
@@ -20,18 +20,46 @@ end
 ---@param args table
 ---@param adapter neotest.Adapter
 function TestRunner:run_tree(tree, args, adapter, on_results)
-  local results = {}
-  local results_callback = function(results_)
-    on_results(results_)
-    for pos_id, result in ipairs(results_) do
-      results[pos_id] = result
+  local all_results = {}
+  local results_callback = function(root, results, output_path)
+    local function fill_results(missing_results)
+      for pos_id, result in pairs(missing_results) do
+        results[pos_id] = result
+      end
+    end
+
+    fill_results(self:_missing_results(root, results, not output_path))
+
+    if output_path then
+      for _, pos in root:iter() do
+        if not results[pos.id] and not all_results[pos.id] then
+          results[pos.id] = {
+            status = "failed",
+            errors = {},
+            output = output_path,
+          }
+        end
+      end
+
+      for _, result in pairs(results) do
+        if not result.output then
+          result.output = output_path
+        end
+      end
+    end
+
+    for pos_id, result in pairs(results) do
+      all_results[pos_id] = result
     end
+    fill_results(self:_missing_results(tree, all_results, true))
+    on_results(results)
   end
 
   args = vim.tbl_extend("keep", args or {}, { strategy = "integrated" })
 
   self:_run_tree(tree, args, adapter, results_callback)
-  on_results(results)
+
+  return all_results
 end
 
 function TestRunner:_run_tree(tree, args, adapter, results_callback)
@@ -44,92 +72,31 @@ function TestRunner:_run_tree(tree, args, adapter, results_callback)
   self:_run_spec(spec, tree, args, adapter, results_callback)
 end
 
-function TestRunner:_stream_queue()
-  local sender, receiver = async.control.channel.mpsc()
-
-  local producer = function(output_stream)
-    local orig = ""
-    local pending_data = nil
-    for data in output_stream do
-      orig = orig .. data
-      local ends_with_newline = vim.endswith(data, "\n")
-      local next_lines = vim.split(data, "\n", { plain = true, trimempty = true })
-      if pending_data then
-        next_lines[1] = pending_data .. next_lines[1]
-        pending_data = nil
-      end
-      if not ends_with_newline then
-        pending_data = table.remove(next_lines, #next_lines)
-      end
-      for _, line in ipairs(next_lines) do
-        sender.send(line)
-      end
-    end
-  end
-
-  local consumer = function()
-    return receiver.recv()
-  end
-  return producer, consumer
-end
-
 ---@param spec neotest.RunSpec
 ---@param adapter neotest.Adapter
 function TestRunner:_run_spec(spec, tree, args, adapter, results_callback)
   local position = tree:data()
   spec.strategy =
     vim.tbl_extend("force", spec.strategy or {}, config.strategies[args.strategy] or {})
-    spec.env = vim.tbl_extend("force", spec.env or {}, args.env or {})
-    spec.cwd = args.cwd or spec.cwd
-    if vim.tbl_isempty(spec.env or {}) then
-      spec.env = nil
-    end
-
+  spec.env = vim.tbl_extend("force", spec.env or {}, args.env or {})
+  spec.cwd = args.cwd or spec.cwd
+  if vim.tbl_isempty(spec.env or {}) then
+    spec.env = nil
+  end
 
   local proc_key = self:_create_process_key(adapter.name, position.id)
-  local producer, consumer = self:_stream_queue()
 
-  local process_result = self._processes:run(proc_key, spec, args, spec.stream and producer)
-  if spec.stream then
-    async.run(function()
-      for stream_results in spec.stream(consumer) do
-        results_callback(stream_results)
+  local stream_processor = spec.stream
+    and function(stream)
+      for stream_results in spec.stream(stream) do
+        results_callback(tree, stream_results)
       end
-    end)
-  end
-
-  local results = adapter.results(spec, process_result, tree)
-
-  if vim.tbl_isempty(results) then
-    results_callback(self:_fill_empty_results(tree, process_result.output))
-    return
-  end
-
-  self:fill_results(tree, results)
-
-  for _, result in pairs(results) do
-    if not result.output then
-      result.output = process_result.output
     end
-  end
+  local process_result = self._processes:run(proc_key, spec, args, stream_processor)
 
-  results_callback(results)
-end
+  local results = adapter.results(spec, process_result, tree)
 
-function TestRunner:_fill_empty_results(tree, output_path)
-  if #tree:children() == 0 then
-    return { [tree:data().id] = { status = "skipped", output = output_path } }
-  end
-  local results = {}
-  logger.warn("Results returned were empty, setting all positions to failed")
-  for _, pos in tree:iter() do
-    results[pos.id] = {
-      status = "failed",
-      errors = {},
-      output = output_path,
-    }
-  end
-  return results
+  results_callback(tree, results, process_result.output)
 end
 
 function TestRunner:_run_broken_down_tree(tree, args, adapter, results_callback)
@@ -211,7 +178,13 @@ end
 ---@async
 ---@param tree neotest.Tree
 ---@param results table<string, neotest.Result>
-function TestRunner:fill_results(tree, results)
+---@param partial? boolean
+function TestRunner:_missing_results(tree, results, partial)
+  local new_results = setmetatable({}, {
+    __index = function(_, key)
+      return results[key]
+    end,
+  })
   local root = tree:data()
   local missing_tests = {}
   for _, node in tree:iter_nodes() do
@@ -223,7 +196,6 @@ function TestRunner:fill_results(tree, results)
         if not lib.positions.contains(root, parent_pos) then
           break
         end
-
         local parent_result = results[parent_pos.id]
         local pos_result = results[pos.id]
         if not parent_result then
@@ -240,7 +212,7 @@ function TestRunner:fill_results(tree, results)
           parent_result.errors = vim.list_extend(parent_result.errors or {}, pos_result.errors)
         end
 
-        results[parent_pos.id] = parent_result
+        new_results[parent_pos.id] = parent_result
       end
     else
       if pos.type == "test" then
@@ -248,6 +220,13 @@ function TestRunner:fill_results(tree, results)
       end
     end
   end
+  if partial then
+    for _, test_id in ipairs(missing_tests) do
+      for parent in tree:get_key(test_id):iter_parents() do
+        new_results[parent:data().id] = nil
+      end
+    end
+  end
 
   local root_result = results[root.id]
   for _, node in tree:iter_nodes() do
@@ -256,22 +235,18 @@ function TestRunner:fill_results(tree, results)
       if pos.type == "file" then
         -- Files not being present means that they were skipped (probably)
         if not results[pos.id] and root_result then
-          results[pos.id] = { status = "skipped", output = root_result.output }
+          new_results[pos.id] = { status = "skipped", output = root_result.output }
         end
       else
         -- Tests and namespaces not being present means that they failed to even start, count as root result
         if not results[pos.id] and root_result then
-          results[pos.id] = { status = root_result.status, output = root_result.output }
+          new_results[pos.id] = { status = root_result.status, output = root_result.output }
         end
       end
     end
   end
 
-  for _, test_id in ipairs(missing_tests) do
-    for parent in tree:get_key(test_id):iter_parents() do
-      results[parent:data().id] = nil
-    end
-  end
+  return new_results
 end
 
 return function(processes)
diff --git a/lua/neotest/client/state/init.lua b/lua/neotest/client/state/init.lua
index 22d0dafd..675bfc97 100644
--- a/lua/neotest/client/state/init.lua
+++ b/lua/neotest/client/state/init.lua
@@ -61,15 +61,17 @@ function NeotestClientState:update_positions(adapter_id, tree)
 end
 
 ---@param results table<string, neotest.Result>
-function NeotestClientState:update_results(adapter_id, results)
+function NeotestClientState:update_results(adapter_id, results, partial)
   logger.debug("New results for adapter", adapter_id)
   logger.trace(results)
   self._results[adapter_id] = vim.tbl_extend("force", self._results[adapter_id] or {}, results)
   if not self._running[adapter_id] then
     self._running[adapter_id] = {}
   end
-  for id, _ in pairs(results) do
-    self._running[adapter_id][id] = nil
+  if not partial then
+    for id, _ in pairs(results) do
+      self._running[adapter_id][id] = nil
+    end
   end
   self._events:emit(NeotestEvents.RESULTS, adapter_id, results)
 end
diff --git a/lua/neotest/client/strategies/dap/init.lua b/lua/neotest/client/strategies/dap/init.lua
index 8046a75f..567e25e5 100644
--- a/lua/neotest/client/strategies/dap/init.lua
+++ b/lua/neotest/client/strategies/dap/init.lua
@@ -1,7 +1,8 @@
 local async = require("neotest.async")
+local FanoutAccum = require("neotest.types").FanoutAccum
 
 ---@param spec neotest.RunSpec
----@return neotest.StrategyResult
+---@return neotest.StrategyResult?
 return function(spec)
   if vim.tbl_isempty(spec.strategy) then
     return
@@ -9,23 +10,40 @@ return function(spec)
   local dap = require("dap")
 
   local handler_id = "neotest_" .. async.fn.localtime()
+  local data_accum = FanoutAccum(function(prev, new)
+    if not prev then
+      return new
+    end
+    return prev .. new
+  end, nil)
 
   local output_path = vim.fn.tempname()
-  local output_file = assert(io.open(output_path, "w"))
+  local open_err, output_fd = async.uv.fs_open(output_path, "w", 438)
+  assert(not open_err, open_err)
+
+  data_accum:subscribe(function(data)
+    local write_err, _ = async.uv.fs_write(output_fd, data)
+    assert(not write_err, write_err)
+  end)
 
   local finish_cond = async.control.Condvar.new()
   local result_code
 
+  async.util.scheduler()
   dap.run(vim.tbl_extend("keep", spec.strategy, { env = spec.env, cwd = spec.cwd }), {
     before = function(config)
       dap.listeners.after.event_output[handler_id] = function(_, body)
         if vim.tbl_contains({ "stdout", "stderr" }, body.category) then
-          output_file:write(body.output)
+          async.run(function()
+            data_accum:push(body.output)
+          end)
         end
       end
       dap.listeners.after.event_exited[handler_id] = function(_, info)
         result_code = info.exitCode
-        finish_cond:notify_all()
+        async.run(function()
+          pcall(finish_cond.notify_all, finish_cond)
+        end)
       end
 
       return config
@@ -38,6 +56,17 @@ return function(spec)
     is_complete = function()
       return result_code ~= nil
     end,
+    output_stream = function()
+      local sender, receiver = async.control.channel.mpsc()
+      data_accum:subscribe(function(d)
+        sender.send(d)
+      end)
+      return function()
+        return async.lib.first(function()
+          finish_cond:wait()
+        end, receiver.recv)
+      end
+    end,
     output = function()
       return output_path
     end,
diff --git a/lua/neotest/client/strategies/init.lua b/lua/neotest/client/strategies/init.lua
index 323d678a..17f95799 100644
--- a/lua/neotest/client/strategies/init.lua
+++ b/lua/neotest/client/strategies/init.lua
@@ -10,9 +10,9 @@ end)
 
 ---@class neotest.ProcessTracker
 ---@field _instances table<integer, neotest.Process>
-local NeotestProcessTracker = {}
+local ProcessTracker = {}
 
-function NeotestProcessTracker:new()
+function ProcessTracker:new()
   local tracker = {
     _instances = {},
   }
@@ -24,10 +24,10 @@ end
 ---@async
 ---@param pos_id string
 ---@param spec neotest.RunSpec
----@param args? table
+---@param args table
+---@param stream_processor async fun(data_iter: async fun(): string)
 ---@return neotest.StrategyResult
-function NeotestProcessTracker:run(pos_id, spec, args, process_stream)
-  --TODO Break this up so we can use instance.output_stream before awaiting finish
+function ProcessTracker:run(pos_id, spec, args, stream_processor)
   local strategy = self:_get_strategy(args)
   logger.info("Starting process", pos_id, "with strategy", args.strategy)
   logger.debug("Strategy spec", spec)
@@ -39,11 +39,10 @@ function NeotestProcessTracker:run(pos_id, spec, args, process_stream)
     return { code = 1, output = output_path }
   end
   self._instances[pos_id] = instance
-  if process_stream then
+  if stream_processor then
+    local iterator = lib.files.split_lines(instance.output_stream())
     async.run(function()
-      for data in instance.output_stream() do
-        process_stream(data)
-      end
+      stream_processor(iterator)
     end)
   end
   local code = instance.result()
@@ -54,7 +53,7 @@ function NeotestProcessTracker:run(pos_id, spec, args, process_stream)
   return { code = code, output = output }
 end
 
-function NeotestProcessTracker:stop(pos_id)
+function ProcessTracker:stop(pos_id)
   local instance = self._instances[pos_id]
   if not instance then
     return false
@@ -64,7 +63,7 @@ function NeotestProcessTracker:stop(pos_id)
 end
 
 ---@return neotest.Strategy
-function NeotestProcessTracker:_get_strategy(args)
+function ProcessTracker:_get_strategy(args)
   if type(args.strategy) == "string" then
     return get_strategy(args.strategy)
   end
@@ -73,7 +72,7 @@ end
 
 ---@async
 ---@param pos_id string
-function NeotestProcessTracker:attach(pos_id)
+function ProcessTracker:attach(pos_id)
   local instance = self._instances[pos_id]
   if not instance then
     return false
@@ -82,10 +81,10 @@ function NeotestProcessTracker:attach(pos_id)
   return true
 end
 
-function NeotestProcessTracker:exists(proc_key)
+function ProcessTracker:exists(proc_key)
   return self._instances[proc_key] ~= nil
 end
 
 return function()
-  return NeotestProcessTracker:new()
+  return ProcessTracker:new()
 end
diff --git a/lua/neotest/client/strategies/integrated/init.lua b/lua/neotest/client/strategies/integrated/init.lua
index 6913d73c..4adffadf 100644
--- a/lua/neotest/client/strategies/integrated/init.lua
+++ b/lua/neotest/client/strategies/integrated/init.lua
@@ -2,24 +2,6 @@ local async = require("neotest.async")
 local lib = require("neotest.lib")
 local FanoutAccum = require("neotest.types").FanoutAccum
 
-local function first(...)
-  local functions = { ... }
-  local send_ran, await_ran = async.control.channel.oneshot()
-  local result, ran
-  for _, func in ipairs(functions) do
-    async.run(function()
-      local func_result = func()
-      if not ran then
-        result = func_result
-        ran = true
-        send_ran()
-      end
-    end)
-  end
-  await_ran()
-  return result
-end
-
 ---@class integratedStrategyConfig
 ---@field height integer
 ---@field width integer
@@ -84,19 +66,27 @@ return function(spec)
     end,
     output_stream = function()
       local sender, receiver = async.control.channel.mpsc()
-      data_accum:subscribe(sender.send)
+      data_accum:subscribe(function(d)
+        sender.send(d)
+      end)
       return function()
-        return first(finish_cond:wait(), receiver.recv)
+        return async.lib.first(function()
+          finish_cond:wait()
+        end, receiver.recv)
       end
     end,
     attach = function()
-      attach_buf = attach_buf or vim.api.nvim_create_buf(false, true)
-      attach_chan = attach_chan
-        or vim.api.nvim_open_term(attach_buf, {
+      if not attach_buf then
+        attach_buf = vim.api.nvim_create_buf(false, true)
+        attach_chan = vim.api.nvim_open_term(attach_buf, {
           on_input = function(_, _, _, data)
             pcall(async.api.nvim_chan_send, job, data)
           end,
         })
+        data_accum:subscribe(function(data)
+          async.api.nvim_chan_send(attach_chan, data)
+        end)
+      end
       attach_win = lib.ui.float.open({
         height = spec.strategy.height,
         width = spec.strategy.width,
@@ -110,10 +100,6 @@ return function(spec)
         end,
       })
       attach_win:jump_to()
-
-      data_accum:subscribe(function(data)
-        async.api.nvim_chan_send(attach_chan, data)
-      end)
     end,
     result = function()
       if result_code == nil then
diff --git a/lua/neotest/consumers/output.lua b/lua/neotest/consumers/output.lua
index d1c02ceb..a181135c 100644
--- a/lua/neotest/consumers/output.lua
+++ b/lua/neotest/consumers/output.lua
@@ -91,6 +91,9 @@ local client
 local init = function()
   if config.output.open_on_run then
     client.listeners.results = function(_, results)
+      if win then
+        return
+      end
       local cur_pos = async.fn.getpos(".")
       local line = cur_pos[2] - 1
       local buf_path = vim.fn.expand("%:p")
diff --git a/lua/neotest/lib/file/init.lua b/lua/neotest/lib/file/init.lua
index 507b0aa3..1f2981ce 100644
--- a/lua/neotest/lib/file/init.lua
+++ b/lua/neotest/lib/file/init.lua
@@ -22,13 +22,97 @@ function M.read(file_path)
   return data
 end
 
+---@async
 function M.read_lines(file_path)
   local data = M.read(file_path)
-  local lines = {}
-  for line in vim.gsplit(data, "[\r]?\n", false) do
-    lines[#lines + 1] = line
+  return vim.split(data, "[\r]?\n", { trimempty = true })
+end
+
+---@param data_iterator fun(): string
+---@return fun(): string[]
+function M.split_lines(data_iterator)
+  local sender, receiver = async.control.channel.mpsc()
+
+  local producer = function()
+    local orig = ""
+    local pending_data = nil
+    for data in data_iterator do
+      orig = orig .. data
+      local ends_with_newline = vim.endswith(data, "\n")
+      local next_lines = vim.split(data, "[\r]?\n", { trimempty = true })
+      if pending_data then
+        if vim.startswith(data, "\r\n") or vim.startswith(data, "\n") then
+          table.insert(next_lines, 1, pending_data)
+        else
+          next_lines[1] = pending_data .. next_lines[1]
+        end
+        pending_data = nil
+      end
+      if not ends_with_newline then
+        pending_data = table.remove(next_lines, #next_lines)
+      end
+      sender.send(next_lines)
+    end
   end
-  return lines
+
+  async.run(producer)
+
+  return receiver.recv
+end
+
+---@return fun(): string, fun()
+function M.stream(file_path)
+  local sender, receiver = async.control.channel.mpsc()
+  local read_semaphore = async.control.Semaphore.new(1)
+
+  local open_err, file_fd = async.uv.fs_open(file_path, "r", 438)
+  assert(not open_err, open_err)
+  local data_read = 0
+
+  local send_exit, await_exit = async.control.channel.oneshot()
+  local read = function()
+    local permit = read_semaphore:acquire()
+    local stat_err, stat = async.uv.fs_fstat(file_fd)
+    assert(not stat_err, stat_err)
+    if data_read == stat.size then
+      permit:forget()
+      return
+    end
+    if data_read > stat.size then
+      send_exit()
+      error("Data deleted from file while streaming")
+    end
+    local read_err, data = async.uv.fs_read(file_fd, stat.size - data_read, data_read)
+    assert(not read_err, read_err)
+    data_read = #data + data_read
+    permit:forget()
+    sender.send(data)
+  end
+
+  read()
+  local event = vim.loop.new_fs_event()
+  event:start(file_path, {}, function(err, _, _)
+    assert(not err)
+    async.run(read)
+  end)
+
+  local function stop()
+    await_exit()
+    event:stop()
+    local close_err = async.uv.fs_close(file_fd)
+    assert(not close_err, close_err)
+  end
+
+  async.run(stop)
+
+  return receiver.recv, send_exit
+end
+
+---@param file_path str
+---@return fun(): string[], fun()
+function M.stream_lines(file_path)
+  local stream, stop = M.stream(file_path)
+  return M.split_lines(stream), stop
 end
 
 function M.exists(path)
@@ -73,6 +157,7 @@ function M.parse_dir_from_files(root, files)
     local function dir_contains(dir, child)
       return vim.startswith(child.path, dir.path .. M.sep)
     end
+
     local current_level = { parent }
     while true do
       local next_pos = dirs:peek()
diff --git a/lua/neotest/types/init.lua b/lua/neotest/types/init.lua
index fe6b7a97..030879f7 100644
--- a/lua/neotest/types/init.lua
+++ b/lua/neotest/types/init.lua
@@ -16,12 +16,12 @@
 ---@field line? integer
 
 ---@class neotest.Process
----@field output async fun()string Output data
+---@field output async fun(): string Path to file containing output data
 ---@field is_complete fun() boolean Is process complete
 ---@field result async fun() integer Get result code of process (async)
 ---@field attach async fun() Attach to the running process for user input
 ---@field stop async fun() Stop the running process
----@field output_stream fun(): string | nil Async iterator of process output
+---@field output_stream async fun(): async fun(): string Async iterator of process output
 
 ---@alias neotest.Strategy async fun(spec: neotest.RunSpec): neotest.Process
 
diff --git a/tests/unit/client/init_spec.lua b/tests/unit/client/init_spec.lua
index df548da1..54b2fa77 100644
--- a/tests/unit/client/init_spec.lua
+++ b/tests/unit/client/init_spec.lua
@@ -12,10 +12,16 @@ end
 describe("neotest client", function()
   ---@type neotest.InternalClient
   local client
-  local mock_adapter, mock_strategy, attached, stopped, exit_test, provided_spec
+  ---@type neotest.Adapter
+  local mock_adapter
+  local mock_strategy, attached, stopped, exit_test, provided_spec
   local dir = async.fn.getcwd()
   local files
   local dirs = { dir }
+  ---@return neotest.Tree
+  local get_pos = function(...)
+    return client:get_position(...)
+  end
   before_each(function()
     dirs = { dir }
     files = { dir .. "/test_file_1", dir .. "/test_file_2" }
@@ -66,7 +72,7 @@ describe("neotest client", function()
         end)
       end,
       build_spec = function()
-        return {}
+        return { strategy = { output = "not_a_file" } }
       end,
       results = function(_, _, tree)
         local results = {}
@@ -94,6 +100,14 @@ describe("neotest client", function()
         stop = function()
           send_exit()
         end,
+        output_stream = function()
+          local data = { "1\n", "2\n3", "\n4\n", "5\n" }
+          local i = 0
+          return function()
+            i = i + 1
+            return data[i]
+          end
+        end,
         attach = function()
           attached = true
         end,
@@ -111,14 +125,14 @@ describe("neotest client", function()
 
   describe("reading positions", function()
     a.it("reads all tests files", function()
-      local tree = client:get_position(dir)
+      local tree = get_pos(dir)
       assert.Not.Nil(tree:get_key(dir .. "/test_file_1"))
       assert.Not.Nil(tree:get_key(dir .. "/test_file_2"))
       assert.Nil(tree:get_key(dir .. "/test_file_3"))
     end)
 
     a.it("updates files when first requested", function()
-      local tree = client:get_position(dir)
+      local tree = get_pos(dir)
       local file_tree = tree:get_key(dir .. "/test_file_1")
       assert.Not.same(file_tree:children(), {})
     end)
@@ -135,7 +149,7 @@ describe("neotest client", function()
         files[#files + 1] = dir .. "/test_file_3"
         client:_update_positions(dir)
         client:_update_positions(dir .. "/test_file_3")
-        local file_tree = client:get_position(dir .. "/test_file_3")
+        local file_tree = get_pos(dir .. "/test_file_3")
         assert.Not.same(file_tree:children(), {})
       end)
     end)
@@ -146,7 +160,7 @@ describe("neotest client", function()
         client:_update_positions(dir)
         files[#files + 1] = dir .. "/new_dir/test_file_3"
         client:_update_positions(dir .. "/new_dir")
-        local file_tree = client:get_position(dir .. "/new_dir")
+        local file_tree = get_pos(dir .. "/new_dir")
         assert.Not.same(file_tree:children(), {})
       end)
     end)
@@ -157,9 +171,9 @@ describe("neotest client", function()
           adapters = { mock_adapter },
           discovery = { enabled = false },
         })
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         assert.Nil(tree)
-        tree = client:get_position(dir .. "/test_file_1")
+        tree = get_pos(dir .. "/test_file_1")
         assert.Nil(tree)
       end)
 
@@ -170,13 +184,13 @@ describe("neotest client", function()
         })
         local bufnr = async.fn.bufadd(dir .. "/test_file_1")
         async.fn.bufload(bufnr)
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         assert.Not.Nil(tree)
         assert.Not.same(tree, {})
-        tree = client:get_position(dir .. "/test_file_1")
+        tree = get_pos(dir .. "/test_file_1")
         assert.Not.Nil(tree)
         assert.Not.same(tree, {})
-        tree = client:get_position(dir .. "/test_file_2")
+        tree = get_pos(dir .. "/test_file_2")
         assert.Nil(tree)
         async.api.nvim_buf_delete(bufnr, {})
       end)
@@ -186,40 +200,109 @@ describe("neotest client", function()
   describe("running tests", function()
     describe("using args", function()
       a.it("provides env", function()
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         exit_test()
         client:run_tree(tree, { strategy = mock_strategy, env = { TEST = "test" } })
         assert.equal(provided_spec.env.TEST, "test")
       end)
 
       a.it("provides cwd", function()
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         exit_test()
         client:run_tree(tree, { strategy = mock_strategy, cwd = "new_cwd" })
         assert.equal(provided_spec.cwd, "new_cwd")
       end)
     end)
+
     describe("with unsupported roots", function()
-      a.it("breaks up directories to files", function()
-        local positions_run = {}
-        mock_adapter.build_spec = function(args)
-          local tree = args.tree
-          local pos = tree:data()
-          if pos.type == "dir" then
-            return
+      describe("supporting files", function()
+        a.it("breaks up directories to files", function()
+          local positions_run = {}
+          mock_adapter.build_spec = function(args)
+            local tree = args.tree
+            local pos = tree:data()
+            if pos.type == "dir" then
+              return
+            end
+            positions_run[pos.id] = true
+            return {}
           end
-          positions_run[pos.id] = true
-          return {}
-        end
 
-        local tree = client:get_position(dir)
-        exit_test()
-        client:run_tree(tree)
+          local tree = get_pos(dir)
+          exit_test()
+          client:run_tree(tree)
 
-        assert.same({
-          [dir .. "/test_file_1"] = true,
-          [dir .. "/test_file_2"] = true,
-        }, positions_run)
+          assert.same({
+            [dir .. "/test_file_1"] = true,
+            [dir .. "/test_file_2"] = true,
+          }, positions_run)
+        end)
+
+        a.it("sets results of directories as results are streamed", function()
+          local positions_run = {}
+
+          dirs = { dir, dir .. "/new_dir" }
+          client:_update_positions(dir)
+          files[#files + 1] = dir .. "/new_dir/test_file_3"
+          client:_update_positions(dir .. "/new_dir")
+          local child_file = client:get_position(dir .. "/new_dir/test_file_3")
+
+          mock_adapter.build_spec = function(args)
+            local tree = args.tree
+            local pos = tree:data()
+            if pos.type == "dir" then
+              return
+            end
+            positions_run[pos.id] = true
+            return {
+              stream = function()
+                local sent = false
+                return function()
+                  if sent then
+                    return nil
+                  end
+                  if pos.id == child_file:data().id then
+                    sent = true
+                    local results = {}
+                    for _, pos in child_file:iter() do
+                      results[pos.id] = { status = "passed" }
+                    end
+                    return results
+                  end
+                end
+              end,
+            }
+          end
+
+          local tree = get_pos(dir)
+
+          async.run(function()
+            client:run_tree(tree, { strategy = mock_strategy })
+          end)
+          async.util.sleep(10)
+
+          local results = client:get_results(mock_adapter.name)
+
+          assert.same({
+            [dir .. "/new_dir"] = {
+              status = "passed",
+            },
+            [dir .. "/new_dir/test_file_3"] = {
+              status = "passed",
+            },
+            [dir .. "/new_dir/test_file_3::namespace"] = {
+              status = "passed",
+            },
+            [dir .. "/new_dir/test_file_3::test_a"] = {
+              status = "passed",
+            },
+            [dir .. "/new_dir/test_file_3::test_b"] = {
+              status = "passed",
+            },
+          }, results)
+
+          exit_test()
+        end)
       end)
 
       a.it("breaks up files to tests", function()
@@ -234,7 +317,7 @@ describe("neotest client", function()
           return {}
         end
 
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         exit_test()
         client:run_tree(tree)
 
@@ -258,7 +341,7 @@ describe("neotest client", function()
           return {}
         end
 
-        local tree = client:get_position(dir .. "/test_file_1::namespace")
+        local tree = get_pos(dir .. "/test_file_1::namespace")
         exit_test()
         client:run_tree(tree)
 
@@ -271,7 +354,7 @@ describe("neotest client", function()
 
     describe("attaching", function()
       a.it("with position", function()
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         async.run(function()
           client:run_tree(tree, { strategy = mock_strategy })
         end)
@@ -281,7 +364,7 @@ describe("neotest client", function()
       end)
 
       a.it("with child", function()
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         async.run(function()
           client:run_tree(tree, { strategy = mock_strategy })
         end)
@@ -293,7 +376,7 @@ describe("neotest client", function()
 
     describe("stopping", function()
       a.it("with position", function()
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         local stopped
         async.run(function()
           client:run_tree(tree, { strategy = mock_strategy })
@@ -304,7 +387,7 @@ describe("neotest client", function()
       end)
 
       a.it("with child", function()
-        local tree = client:get_position(dir)
+        local tree = get_pos(dir)
         async.run(function()
           client:run_tree(tree, { strategy = mock_strategy })
           stopped = true
@@ -314,8 +397,106 @@ describe("neotest client", function()
       end)
     end)
 
+    describe("with streamed results", function()
+      a.it("streams output data", function()
+        local streamed_data = {}
+        mock_adapter.build_spec = function()
+          return {
+            stream = function(data)
+              for lines in data do
+                vim.list_extend(streamed_data, lines)
+              end
+            end,
+          }
+        end
+        local tree = get_pos(dir)
+        async.run(function()
+          client:run_tree(tree, { strategy = mock_strategy })
+        end)
+        async.util.sleep(10)
+
+        assert.same({ "1", "2", "3", "4", "5" }, streamed_data)
+        exit_test()
+      end)
+
+      a.it("emits streamed results", function()
+        local tree = get_pos(dir .. "/test_file_1")
+        mock_adapter.build_spec = function()
+          return {
+            stream = function()
+              local results = {}
+              for i, pos in tree:iter() do
+                if pos.type == "test" and i % 2 == 0 then
+                  results[pos.id] = { status = "passed" }
+                end
+              end
+              local i, result
+              return function()
+                i, result = next(results, i)
+                if i then
+                  return { [i] = result }
+                end
+              end
+            end,
+          }
+        end
+        async.run(function()
+          client:run_tree(tree, { strategy = mock_strategy })
+        end)
+        async.util.sleep(10)
+
+        local results = client:get_results(mock_adapter.name)
+        for i, pos in tree:iter() do
+          if i % 2 == 0 and pos.type == "test" then
+            assert.same({ status = "passed" }, results[pos.id])
+          else
+            assert.Nil(results[pos.id])
+          end
+        end
+
+        exit_test()
+      end)
+
+      a.it("attaches position", function()
+        local tree = get_pos(dir .. "/test_file_1")
+        mock_adapter.build_spec = function()
+          return {
+            stream = function()
+              local results = {}
+              for i, pos in tree:iter() do
+                if pos.type == "test" and i % 2 == 0 then
+                  results[pos.id] = { status = "passed" }
+                end
+              end
+              local i, result
+              return function()
+                i, result = next(results, i)
+                if i then
+                  return { [i] = result }
+                end
+              end
+            end,
+          }
+        end
+        async.run(function()
+          client:run_tree(tree, { strategy = mock_strategy })
+        end)
+        async.util.sleep(10)
+
+        for i, pos in tree:iter_nodes() do
+          if i % 2 == 0 and pos:data().type == "test" then
+            client:attach(pos)
+            break
+          end
+        end
+        assert.True(attached)
+
+        exit_test()
+      end)
+    end)
+
     a.it("fills results for dir from child files", function()
-      local tree = client:get_position(dir)
+      local tree = get_pos(dir)
       exit_test()
       client:run_tree(tree, { strategy = mock_strategy })
       local results = client:get_results(mock_adapter.name)
@@ -325,7 +506,7 @@ describe("neotest client", function()
     end)
 
     a.it("fills results for namespaces from child tests", function()
-      local tree = client:get_position(dir .. "/test_file_1")
+      local tree = get_pos(dir .. "/test_file_1")
       exit_test()
       client:run_tree(tree, { strategy = mock_strategy })
       local results = client:get_results(mock_adapter.name)
@@ -349,7 +530,7 @@ describe("neotest client", function()
         return results
       end
 
-      local tree = client:get_position(dir)
+      local tree = get_pos(dir)
       exit_test()
       client:run_tree(tree, { strategy = mock_strategy })
       local results = client:get_results(mock_adapter.name)
diff --git a/tests/unit/client/strategies/integrated_spec.lua b/tests/unit/client/strategies/integrated_spec.lua
new file mode 100644
index 00000000..4c4b793f
--- /dev/null
+++ b/tests/unit/client/strategies/integrated_spec.lua
@@ -0,0 +1,73 @@
+local async = require("neotest.async")
+local a = async.tests
+local lib = require("neotest.lib")
+local strategy = require("neotest.client.strategies.integrated")
+
+A = function(...)
+  print(vim.inspect(...))
+end
+describe("integrated strategy", function()
+  a.it("produces output", function()
+    local process = strategy({
+      command = { "printf", "hello" },
+      strategy = {
+        height = 10,
+        width = 10,
+      },
+    })
+    process.result()
+    local output = lib.files.read(process.output())
+    assert.equal(output, "hello")
+  end)
+
+  a.it("returns exit code", function()
+    local process = strategy({
+      command = { "bash", "-c", "exit 100" },
+      strategy = {
+        height = 10,
+        width = 10,
+      },
+    })
+    local code = process.result()
+    assert.equal(code, 100)
+  end)
+
+  a.it("stops the job", function()
+    local process = strategy({
+      command = { "bash", "-c", "sleep 1" },
+      strategy = {
+        height = 10,
+        width = 10,
+      },
+    })
+    process.stop()
+    local code = process.result()
+    assert.Not.equal(0, code)
+  end)
+
+  a.it("streams output", function()
+    local process = strategy({
+      command = { "bash", "-c", "printf hello; sleep 0; printf world" },
+      strategy = {
+        height = 10,
+        width = 10,
+      },
+    })
+    local stream = process.output_stream()
+    assert.equal("hello", stream())
+    assert.equal("world", stream())
+  end)
+
+  a.it("opens attach window", function()
+    local process = strategy({
+      command = { "echo", "hello" },
+      strategy = {
+        height = 10,
+        width = 10,
+      },
+    })
+    async.util.sleep(100)
+    process.attach()
+    assert.Not.equal(async.api.nvim_win_get_config(0), "")
+  end)
+end)
diff --git a/tests/unit/lib/files/init_spec.lua b/tests/unit/lib/files/init_spec.lua
index b5e3b803..c33dd1d7 100644
--- a/tests/unit/lib/files/init_spec.lua
+++ b/tests/unit/lib/files/init_spec.lua
@@ -1,3 +1,5 @@
+local async = require("neotest.async")
+local a = async.tests
 local files = require("neotest.lib").files
 A = function(...)
   print(vim.inspect(...))
@@ -117,4 +119,69 @@ describe("files library", function()
       })
     end)
   end)
+
+  describe("reading files", function()
+    local path, file
+    before_each(function()
+      path = vim.fn.tempname()
+      file = io.open(path, "w")
+    end)
+    after_each(function()
+      file:close()
+    end)
+
+    a.it("reads data", function()
+      file:write("some data")
+      file:flush()
+      local read_data = files.read(path)
+      assert.equal("some data", read_data)
+    end)
+
+    a.it("reads lines", function()
+      file:write("first\r\nsecond\nthird\n")
+      file:flush()
+      local read_data = files.read_lines(path)
+      assert.same({ "first", "second", "third" }, read_data)
+    end)
+
+    a.it("stream lines", function()
+      file:write("first\r\nsecond\nthird\n")
+      file:flush()
+      local lines_iter, stop_reading = files.stream_lines(path)
+      local result = {}
+      async.run(function()
+        for lines in lines_iter do
+          for _, line in ipairs(lines) do
+            result[#result + 1] = line
+          end
+        end
+      end)
+      async.util.sleep(0.1)
+      stop_reading()
+      assert.same({ "first", "second", "third" }, result)
+    end)
+
+    a.it("stream lines after new data written", function()
+      file:write("first\r\nsecond\nthird\n")
+      file:flush()
+      local lines_iter, stop_reading = files.stream_lines(path)
+      local result = {}
+      async.run(function()
+        for lines in lines_iter do
+          for _, line in ipairs(lines) do
+            result[#result + 1] = line
+          end
+        end
+      end)
+      async.util.sleep(10)
+      file:write("fourth")
+      file:flush()
+      async.util.sleep(10)
+      file:write("\nfifth\n")
+      file:flush()
+      async.util.sleep(10)
+      stop_reading()
+      assert.same({ "first", "second", "third", "fourth", "fifth" }, result)
+    end)
+  end)
 end)