diff --git a/kernel/fs/fs.h b/kernel/fs/fs.h
index f63fd3a5..ba4989d3 100644
--- a/kernel/fs/fs.h
+++ b/kernel/fs/fs.h
@@ -30,6 +30,7 @@ constexpr int32_t ERR_XDEV      = -15;
 constexpr int32_t ERR_SPIPE     = -16;
 constexpr int32_t ERR_INTR      = -17;
 constexpr int32_t ERR_NODATA    = -18;
+constexpr int32_t ERR_PIPE      = -19;
 
 /**
  * @brief Initialize the filesystem subsystem. Registers ramfs,
diff --git a/kernel/mm/mm.cpp b/kernel/mm/mm.cpp
index f2689c33..5f51ca32 100644
--- a/kernel/mm/mm.cpp
+++ b/kernel/mm/mm.cpp
@@ -6,6 +6,10 @@
 #include "mm/heap.h"
 #include "mm/paging.h"
 #include "mm/shmem.h"
+#include "sched/sched.h"
+#include "sched/task.h"
+#include "clock/clock.h"
+#include "trace/ktrace_events.h"
 #include "common/string.h"
 #include "common/logging.h"
 #include "hw/cache.h"
@@ -83,9 +87,17 @@ int32_t handle_user_pf(
         return MM_CTX_ERR_INVALID_ARG;
     }
 
+    uint64_t trace_start_ns = ktrace::is_recording(ktrace::EVENT_PAGE_FAULT) ? clock::now_ns() : 0;
+
     sync::mutex_lock(mm_ctx->lock);
     int32_t result = handle_user_pf_locked(mm_ctx, fault_address, pf_flags);
     sync::mutex_unlock(mm_ctx->lock);
+
+    if (trace_start_ns) {
+        ktrace::record_page_fault(sched::current()->tid, fault_address, pf_flags,
+                                  clock::now_ns() - trace_start_ns, result);
+    }
+
     return result;
 }
 
diff --git a/kernel/resource/providers/file_provider.cpp b/kernel/resource/providers/file_provider.cpp
index 384c3d7e..dd263b53 100644
--- a/kernel/resource/providers/file_provider.cpp
+++ b/kernel/resource/providers/file_provider.cpp
@@ -39,6 +39,10 @@ static int32_t map_fs_error_to_resource(int32_t fs_err) {
             return ERR_BUSY;
         case fs::ERR_NODATA:
             return ERR_NODATA;
+        case fs::ERR_INTR:
+            return ERR_INTR;
+        case fs::ERR_PIPE:
+            return ERR_PIPE;
         default:
             return ERR_IO;
     }
diff --git a/kernel/sched/sched.cpp b/kernel/sched/sched.cpp
index cc521cba..af0e2c6b 100644
--- a/kernel/sched/sched.cpp
+++ b/kernel/sched/sched.cpp
@@ -22,6 +22,7 @@
 #include "hw/cpu_features.h"
 #include "clock/clock.h"
 #include "timer/timer.h"
+#include "trace/ktrace_events.h"
 #include "rc/reaper.h"
 #include "exec/elf.h"
 #include "random/random.h"
@@ -362,6 +363,18 @@ __PRIVILEGED_CODE static uint32_t load_balance_select_cpu() {
     return 0;
 }
 
+static uint8_t switch_reason(bool prev_dead, bool preempted, uint32_t prev_state) {
+    if (prev_dead) {
+        return ktrace::SWITCH_REASON_EXITED;
+    }
+
+    if (preempted) {
+        return ktrace::SWITCH_REASON_PREEMPTED;
+    }
+
+    return prev_state == TASK_STATE_BLOCKED ? ktrace::SWITCH_REASON_BLOCKED : ktrace::SWITCH_REASON_YIELDED;
+}
+
 /**
  * @note Privilege: **required**
  */
@@ -425,6 +438,11 @@ __PRIVILEGED_CODE task* pick_next_and_switch(task* prev, bool preempted) {
         next->state.store_relaxed(TASK_STATE_RUNNING);
     }
 
+    if (next != prev) {
+        ktrace::record_sched_switch(prev->tid, prev->name, next->tid, next->name,
+                                    switch_reason(prev_dead, preempted, prev_state));
+    }
+
     // One timestamp ends prev's charge and starts next's, so no time goes uncharged
     uint64_t now_ns = clock::now_ns();
     charge_cpu_time(prev, now_ns);
@@ -563,6 +581,9 @@ __PRIVILEGED_CODE void wake(task* t) {
         return;
     }
 
+    task* waker = current();
+    ktrace::record_sched_wakeup(t->tid, t->name, task_cpu, waker->tid, waker->name);
+
     // If queued or running, its own switch-out keeps a READY task on the runqueue
     if (t->exec.on_runqueue) {
         sync::spin_unlock_irqrestore(rq.lock, irq);
diff --git a/kernel/syscall/syscall.cpp b/kernel/syscall/syscall.cpp
index 0a811ca0..ef73b924 100644
--- a/kernel/syscall/syscall.cpp
+++ b/kernel/syscall/syscall.cpp
@@ -8,6 +8,8 @@
 #include "dynpriv/dynpriv.h"
 #include "percpu/percpu.h"
 #include "hw/cpu.h"
+#include "clock/clock.h"
+#include "trace/ktrace_events.h"
 #include "common/logging.h"
 
 constexpr uint32_t ELEVATION_CONTEXT_MASK = sched::TASK_FLAG_ELEVATED | sched::TASK_FLAG_IN_SYSCALL;
@@ -46,6 +48,11 @@ extern "C" __PRIVILEGED_CODE int64_t stlx_syscall_handler(
     this_cpu(percpu_is_elevated) = true;
     cpu::irq_enable();
 
+    uint64_t trace_start_ns = 0;
+    if (ktrace::is_recording(ktrace::EVENT_SYSCALL) && syscall_num != syscall::SYS_ELEVATE) {
+        trace_start_ns = clock::now_ns();
+    }
+
     int64_t result;
 
     if (syscall_num < syscall::MAX_SYSCALL_NUM && syscall::g_syscall_table[syscall_num]) {
@@ -74,6 +81,12 @@ extern "C" __PRIVILEGED_CODE int64_t stlx_syscall_handler(
         }
     }
 
+    if (trace_start_ns && self) {
+        uint32_t pid = self->group ? self->group->pid : 0;
+        ktrace::record_syscall(self->tid, pid, static_cast<uint32_t>(syscall_num),
+                               clock::now_ns() - trace_start_ns, result);
+    }
+
     // Return-boundary restore: dynamic runtime elevation follows the selected
     // task mode once syscall handling and switch teardown are complete.
     cpu::irq_disable();
diff --git a/kernel/tests/trace/ktrace.test.cpp b/kernel/tests/trace/ktrace.test.cpp
index a3379c38..5867f093 100644
--- a/kernel/tests/trace/ktrace.test.cpp
+++ b/kernel/tests/trace/ktrace.test.cpp
@@ -16,19 +16,36 @@
 TEST_SUITE(ktrace);
 
 struct cpu_line_totals {
-    uint64_t kept;
-    uint32_t wrapped_rings;
+    uint64_t records;
+    uint64_t lost;
+};
+
+struct stream_summary {
+    bool     header_valid;
+    bool     ended;
+    bool     malformed;
+    uint64_t session_start_ns;
+    uint64_t stop_ns;
+    uint64_t records;
+    uint64_t lost;
+    uint64_t out_of_order;
+    uint64_t before_session;
+    uint64_t bytes_after_end;
+    uint64_t cpus_with_records_mask;
 };
 
 static constexpr uint64_t MISSING_STATUS_VALUE = ~0ull;
 static constexpr uint64_t RING_RECORDS = 65536;
 
-// Odd sizes, so reads cross section and record boundaries
-static constexpr size_t SMALL_READ_BYTES = 100;
-static constexpr size_t LARGE_READ_BYTES = 64 * 1024 + 7;
+// Sessions here record only this id, so kernel events never mix into the counts
+static constexpr uint16_t TEST_EVENT_ID = ktrace::EVENT_ID_COUNT - 1;
+static constexpr uint64_t TEST_EVENT_MASK = 1ull << TEST_EVENT_ID;
+
+static constexpr size_t SMALL_STREAM_BYTES = 64 * 1024;
+static constexpr size_t FULL_RING_STREAM_BYTES = RING_RECORDS * sizeof(ktrace::trace_record) + SMALL_STREAM_BYTES;
 
 static char g_status_text[4096];
-alignas(64) static uint8_t g_trace_bytes[4096];
+alignas(64) static uint8_t g_stream_bytes[SMALL_STREAM_BYTES];
 
 static bool read_status() {
     fs::file* status = fs::open("/dev/ktrace/status", fs::O_RDONLY);
@@ -91,7 +108,7 @@ static uint64_t status_value(const char* key) {
     return MISSING_STATUS_VALUE;
 }
 
-// Totals the `cpu<N> <kept> <flags>` lines
+// Totals the `cpu<N> <records> <lost>` lines
 static cpu_line_totals sum_cpu_lines() {
     cpu_line_totals totals = {};
     for (const char* line = g_status_text; *line; line = next_line(line)) {
@@ -102,11 +119,9 @@ static cpu_line_totals sum_cpu_lines() {
         const char* field = line + 3;
         parse_u64(&field);
         field++;
-        totals.kept += parse_u64(&field);
+        totals.records += parse_u64(&field);
         field++;
-        if (parse_u64(&field) & ktrace::CPU_FLAG_RING_WRAPPED) {
-            totals.wrapped_rings++;
-        }
+        totals.lost += parse_u64(&field);
     }
 
     return totals;
@@ -115,32 +130,30 @@ static cpu_line_totals sum_cpu_lines() {
 static void record_events(uint64_t count) {
     for (uint64_t i = 0; i < count; i++) {
         ktrace::trace_record rec = {};
+        rec.hdr.event_id = TEST_EVENT_ID;
         rec.payload[0] = i;
         ktrace::record_event(rec);
     }
 }
 
-static int32_t try_open_trace() {
-    int32_t err = fs::OK;
-    fs::file* trace = fs::open("/dev/ktrace/trace", fs::O_RDONLY, &err);
-    if (!trace) {
-        return err;
-    }
-
-    fs::close(trace);
-    return fs::OK;
+static fs::file* open_reader(int32_t* err = nullptr) {
+    return fs::open("/dev/ktrace/trace", fs::O_RDONLY, err);
 }
 
-static size_t read_trace(uint8_t* buf, size_t cap, size_t chunk) {
-    fs::file* trace = fs::open("/dev/ktrace/trace", fs::O_RDONLY);
-    if (!trace) {
-        return 0;
+static fs::file* start_session_with_reader() {
+    fs::file* reader = open_reader();
+    if (reader && ktrace::start(TEST_EVENT_MASK) != ktrace::OK) {
+        fs::close(reader);
+        return nullptr;
     }
 
+    return reader;
+}
+
+static size_t read_until_end(fs::file* reader, uint8_t* buf, size_t cap) {
     size_t length = 0;
     while (length < cap) {
-        size_t room = cap - length;
-        ssize_t rc = fs::read(trace, buf + length, room < chunk ? room : chunk);
+        ssize_t rc = fs::read(reader, buf + length, cap - length);
         if (rc <= 0) {
             break;
         }
@@ -148,10 +161,88 @@ static size_t read_trace(uint8_t* buf, size_t cap, size_t chunk) {
         length += static_cast<size_t>(rc);
     }
 
-    fs::close(trace);
     return length;
 }
 
+static bool header_matches_session(const ktrace::file_header* header) {
+    return string::memcmp(header->magic, ktrace::FILE_MAGIC, sizeof(header->magic)) == 0 &&
+           header->version == ktrace::FILE_VERSION &&
+           header->header_size == sizeof(ktrace::file_header) &&
+           header->arch == exec::ELF_EXPECTED_MACHINE &&
+           header->cpu_count == smp::cpu_count() &&
+           header->record_size == sizeof(ktrace::trace_record) &&
+           header->session_start_ns > 0 &&
+           header->event_mask == TEST_EVENT_MASK;
+}
+
+// Expects the record_events() payloads in order from 0
+static stream_summary summarize_stream(const uint8_t* bytes, size_t length) {
+    stream_summary summary = {};
+    if (length < sizeof(ktrace::file_header)) {
+        return summary;
+    }
+
+    const auto* header = reinterpret_cast<const ktrace::file_header*>(bytes);
+    summary.header_valid = header_matches_session(header);
+    summary.session_start_ns = header->session_start_ns;
+
+    size_t pos = sizeof(ktrace::file_header);
+    uint64_t next_payload = 0;
+
+    while (pos + sizeof(ktrace::chunk_header) <= length) {
+        const auto* chunk = reinterpret_cast<const ktrace::chunk_header*>(bytes + pos);
+        pos += sizeof(ktrace::chunk_header);
+
+        if (chunk->kind == ktrace::CHUNK_END) {
+            summary.ended = true;
+            summary.stop_ns = chunk->stop_ns;
+            break;
+        }
+
+        uint64_t records_bytes = chunk->record_count * sizeof(ktrace::trace_record);
+        if (chunk->kind != ktrace::CHUNK_RECORDS || chunk->cpu_id >= header->cpu_count ||
+            pos + records_bytes > length) {
+            summary.malformed = true;
+            break;
+        }
+
+        const auto* records = reinterpret_cast<const ktrace::trace_record*>(bytes + pos);
+        for (uint64_t i = 0; i < chunk->record_count; i++) {
+            if (records[i].payload[0] != next_payload) {
+                summary.out_of_order++;
+            }
+
+            if (records[i].hdr.timestamp < header->session_start_ns) {
+                summary.before_session++;
+            }
+
+            next_payload = records[i].payload[0] + 1;
+        }
+
+        if (chunk->record_count > 0) {
+            summary.cpus_with_records_mask |= 1ull << chunk->cpu_id;
+        }
+
+        summary.records += chunk->record_count;
+        summary.lost += chunk->lost_records;
+        pos += records_bytes;
+    }
+
+    summary.bytes_after_end = length - pos;
+    return summary;
+}
+
+static ssize_t write_to_control(const char* command) {
+    fs::file* control = fs::open("/dev/ktrace/control", fs::O_WRONLY);
+    if (!control) {
+        return fs::ERR_NOENT;
+    }
+
+    ssize_t rc = fs::write(control, command, string::strlen(command));
+    fs::close(control);
+    return rc;
+}
+
 // Runs before any other test starts a session
 TEST(ktrace, status_reports_idle_before_the_first_session) {
     ASSERT_TRUE(read_status());
@@ -159,57 +250,97 @@ TEST(ktrace, status_reports_idle_before_the_first_session) {
     EXPECT_EQ(status_value("session_start_ns"), 0u);
     EXPECT_EQ(status_value("session_stop_ns"), 0u);
     EXPECT_EQ(status_value("capacity"), RING_RECORDS);
-    EXPECT_EQ(sum_cpu_lines().kept, 0u);
+    EXPECT_EQ(status_value("trace_open"), 0u);
+    EXPECT_EQ(sum_cpu_lines().records, 0u);
+}
+
+TEST(ktrace, start_fails_without_a_reader) {
+    EXPECT_EQ(ktrace::start(TEST_EVENT_MASK), ktrace::ERR_NO_READER);
+    EXPECT_EQ(write_to_control("start"), fs::ERR_PIPE);
 }
 
-TEST(ktrace, opening_the_trace_before_the_first_session_fails_with_nodata) {
-    EXPECT_EQ(try_open_trace(), fs::ERR_NODATA);
+TEST(ktrace, a_second_reader_is_refused) {
+    fs::file* reader = open_reader();
+    ASSERT_NOT_NULL(reader);
+
+    int32_t err = fs::OK;
+    EXPECT_NULL(open_reader(&err));
+    EXPECT_EQ(err, fs::ERR_BUSY);
+
+    fs::close(reader);
 }
 
 TEST(ktrace, start_fails_while_a_session_is_recording) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
-    EXPECT_EQ(ktrace::start(), ktrace::ERR_BUSY);
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
+
+    EXPECT_EQ(ktrace::start(TEST_EVENT_MASK), ktrace::ERR_BUSY);
     EXPECT_EQ(ktrace::stop(), ktrace::OK);
+
+    fs::close(reader);
 }
 
 TEST(ktrace, stop_fails_when_no_session_is_recording) {
     EXPECT_EQ(ktrace::stop(), ktrace::ERR_NOT_RECORDING);
 }
 
-TEST(ktrace, a_stopped_session_can_be_started_again) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
+TEST(ktrace, a_reader_streams_only_one_session) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
+
     ASSERT_EQ(ktrace::stop(), ktrace::OK);
-    EXPECT_EQ(ktrace::start(), ktrace::OK);
+    EXPECT_EQ(ktrace::start(TEST_EVENT_MASK), ktrace::ERR_NO_READER);
+    fs::close(reader);
+
+    reader = start_session_with_reader();
+    EXPECT_NOT_NULL(reader);
+
     EXPECT_EQ(ktrace::stop(), ktrace::OK);
+    fs::close(reader);
 }
 
-static ssize_t write_to_control(const char* command) {
-    fs::file* control = fs::open("/dev/ktrace/control", fs::O_WRONLY);
-    if (!control) {
-        return fs::ERR_NOENT;
-    }
+TEST(ktrace, closing_the_reader_stops_the_session) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
 
-    ssize_t rc = fs::write(control, command, string::strlen(command));
-    fs::close(control);
-    return rc;
+    fs::close(reader);
+
+    ASSERT_TRUE(read_status());
+    EXPECT_TRUE(status_state_is("stopped"));
+    EXPECT_EQ(status_value("trace_open"), 0u);
 }
 
 TEST(ktrace, writing_start_and_stop_to_control_drives_a_session) {
+    fs::file* reader = open_reader();
+    ASSERT_NOT_NULL(reader);
+
     ASSERT_EQ(write_to_control("start"), 5);
-    EXPECT_EQ(ktrace::start(), ktrace::ERR_BUSY);
+    EXPECT_EQ(ktrace::start(TEST_EVENT_MASK), ktrace::ERR_BUSY);
     EXPECT_EQ(write_to_control("stop"), 4);
     EXPECT_EQ(ktrace::stop(), ktrace::ERR_NOT_RECORDING);
+
+    fs::close(reader);
 }
 
 TEST(ktrace, control_accepts_a_command_ending_in_a_newline) {
+    fs::file* reader = open_reader();
+    ASSERT_NOT_NULL(reader);
+
     ASSERT_EQ(write_to_control("start\n"), 6);
     EXPECT_EQ(write_to_control("stop\n"), 5);
+
+    fs::close(reader);
 }
 
 TEST(ktrace, control_refuses_start_while_a_session_is_recording) {
+    fs::file* reader = open_reader();
+    ASSERT_NOT_NULL(reader);
+
     ASSERT_EQ(write_to_control("start"), 5);
     EXPECT_EQ(write_to_control("start"), fs::ERR_BUSY);
     EXPECT_EQ(write_to_control("stop"), 4);
+
+    fs::close(reader);
 }
 
 TEST(ktrace, control_refuses_stop_when_no_session_is_recording) {
@@ -222,156 +353,178 @@ TEST(ktrace, control_refuses_unknown_commands) {
     EXPECT_EQ(write_to_control("start\n\n"), fs::ERR_INVAL);
 }
 
-TEST(ktrace, status_counts_records_while_recording_and_after_stop) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
+TEST(ktrace, the_stream_holds_the_session_and_ends_with_its_stop_time) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
+
+    record_events(3);
+    ASSERT_EQ(ktrace::stop(), ktrace::OK);
+
+    size_t length = read_until_end(reader, g_stream_bytes, sizeof(g_stream_bytes));
+    fs::close(reader);
+
+    stream_summary summary = summarize_stream(g_stream_bytes, length);
+    EXPECT_TRUE(summary.header_valid);
+    EXPECT_FALSE(summary.malformed);
+    EXPECT_EQ(length % 64, 0u);
+
+    EXPECT_EQ(summary.records, 3u);
+    EXPECT_EQ(summary.out_of_order, 0u);
+    EXPECT_EQ(summary.before_session, 0u);
+    EXPECT_EQ(summary.lost, 0u);
+
+    EXPECT_TRUE(summary.ended);
+    EXPECT_GE(summary.stop_ns, summary.session_start_ns);
+    EXPECT_EQ(summary.bytes_after_end, 0u);
+}
+
+TEST(ktrace, status_counts_the_records_and_bytes_streamed) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
+
     record_events(3);
 
     ASSERT_TRUE(read_status());
     EXPECT_TRUE(status_state_is("recording"));
-    EXPECT_EQ(sum_cpu_lines().kept, 3u);
+    EXPECT_EQ(status_value("trace_open"), 1u);
     EXPECT_GT(status_value("session_start_ns"), 0u);
     EXPECT_EQ(status_value("session_stop_ns"), 0u);
 
     ASSERT_EQ(ktrace::stop(), ktrace::OK);
+    size_t length = read_until_end(reader, g_stream_bytes, sizeof(g_stream_bytes));
 
     ASSERT_TRUE(read_status());
     EXPECT_TRUE(status_state_is("stopped"));
-    EXPECT_EQ(sum_cpu_lines().kept, 3u);
+    EXPECT_EQ(sum_cpu_lines().records, 3u);
+    EXPECT_EQ(sum_cpu_lines().lost, 0u);
+    EXPECT_EQ(status_value("bytes_streamed"), length);
     EXPECT_GE(status_value("session_stop_ns"), status_value("session_start_ns"));
+
+    fs::close(reader);
 }
 
-TEST(ktrace, start_empties_the_rings) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
-    record_events(2);
-    ASSERT_EQ(ktrace::stop(), ktrace::OK);
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
+TEST(ktrace, a_read_while_recording_returns_the_records_written_so_far) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
 
-    ASSERT_TRUE(read_status());
-    EXPECT_EQ(sum_cpu_lines().kept, 0u);
-    EXPECT_EQ(ktrace::stop(), ktrace::OK);
-}
+    record_events(3);
+    ssize_t first = fs::read(reader, g_stream_bytes, sizeof(g_stream_bytes));
+    ASSERT_TRUE(first > 0);
+
+    stream_summary so_far = summarize_stream(g_stream_bytes, static_cast<size_t>(first));
+    EXPECT_TRUE(so_far.header_valid);
+    EXPECT_EQ(so_far.records, 3u);
+    EXPECT_FALSE(so_far.ended);
 
-TEST(ktrace, events_outside_a_session_are_dropped) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
     ASSERT_EQ(ktrace::stop(), ktrace::OK);
-    record_events(5);
+    size_t rest = read_until_end(reader, g_stream_bytes + first, sizeof(g_stream_bytes) - first);
+    fs::close(reader);
 
-    ASSERT_TRUE(read_status());
-    EXPECT_EQ(sum_cpu_lines().kept, 0u);
+    stream_summary whole = summarize_stream(g_stream_bytes, static_cast<size_t>(first) + rest);
+    EXPECT_EQ(whole.records, 3u);
+    EXPECT_TRUE(whole.ended);
+    EXPECT_EQ(whole.bytes_after_end, 0u);
 }
 
-TEST(ktrace, status_flags_a_ring_that_wrapped) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
-    record_events(RING_RECORDS + 10);
+TEST(ktrace, reading_past_the_end_returns_end_of_file) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
+
     ASSERT_EQ(ktrace::stop(), ktrace::OK);
+    read_until_end(reader, g_stream_bytes, sizeof(g_stream_bytes));
+    EXPECT_EQ(fs::read(reader, g_stream_bytes, sizeof(g_stream_bytes)), 0);
 
-    ASSERT_TRUE(read_status());
-    cpu_line_totals totals = sum_cpu_lines();
-    EXPECT_EQ(totals.kept, RING_RECORDS);
-    EXPECT_EQ(totals.wrapped_rings, 1u);
+    fs::close(reader);
 }
 
-TEST(ktrace, opening_the_trace_while_recording_fails_with_busy) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
-    EXPECT_EQ(try_open_trace(), fs::ERR_BUSY);
+TEST(ktrace, a_read_too_small_for_a_chunk_is_refused) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
+
+    constexpr size_t too_small = sizeof(ktrace::chunk_header) + sizeof(ktrace::trace_record) - 1;
+    EXPECT_EQ(fs::read(reader, g_stream_bytes, too_small), fs::ERR_INVAL);
+
     EXPECT_EQ(ktrace::stop(), ktrace::OK);
+    fs::close(reader);
 }
 
-TEST(ktrace, start_fails_while_the_trace_is_open) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
+TEST(ktrace, start_empties_the_rings) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
+
+    record_events(2);
     ASSERT_EQ(ktrace::stop(), ktrace::OK);
+    fs::close(reader);
 
-    fs::file* trace = fs::open("/dev/ktrace/trace", fs::O_RDONLY);
-    ASSERT_NOT_NULL(trace);
-    EXPECT_EQ(ktrace::start(), ktrace::ERR_BUSY);
-    EXPECT_TRUE(read_status());
-    EXPECT_EQ(status_value("trace_open"), 1u);
+    reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
 
-    fs::close(trace);
-    EXPECT_EQ(ktrace::start(), ktrace::OK);
-    EXPECT_EQ(ktrace::stop(), ktrace::OK);
+    ASSERT_EQ(ktrace::stop(), ktrace::OK);
+    size_t length = read_until_end(reader, g_stream_bytes, sizeof(g_stream_bytes));
+    fs::close(reader);
+
+    stream_summary summary = summarize_stream(g_stream_bytes, length);
+    EXPECT_EQ(summary.records, 0u);
+    EXPECT_TRUE(summary.ended);
 }
 
-TEST(ktrace, the_trace_describes_the_stopped_session) {
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
-    record_events(3);
-    ASSERT_EQ(ktrace::stop(), ktrace::OK);
+TEST(ktrace, events_outside_the_session_mask_are_dropped) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
 
-    size_t length = read_trace(g_trace_bytes, sizeof(g_trace_bytes), SMALL_READ_BYTES);
-    ASSERT_TRUE(length >= sizeof(ktrace::file_header));
-
-    const auto* header = reinterpret_cast<const ktrace::file_header*>(g_trace_bytes);
-    EXPECT_EQ(string::memcmp(header->magic, ktrace::FILE_MAGIC, sizeof(header->magic)), 0);
-    EXPECT_EQ(header->version, ktrace::FILE_VERSION);
-    EXPECT_EQ(header->header_size, sizeof(ktrace::file_header));
-    EXPECT_EQ(header->arch, exec::ELF_EXPECTED_MACHINE);
-    EXPECT_EQ(header->cpu_count, smp::cpu_count());
-    EXPECT_EQ(header->record_size, sizeof(ktrace::trace_record));
-    EXPECT_EQ(header->cpu_table_offset, sizeof(ktrace::file_header));
-    EXPECT_GE(header->session_stop_ns, header->session_start_ns);
-
-    const auto* entries = reinterpret_cast<const ktrace::cpu_table_entry*>(g_trace_bytes + header->cpu_table_offset);
-    const ktrace::cpu_table_entry* recorded = nullptr;
-    uint64_t next_records_offset = entries[0].records_offset;
-    uint64_t total_records = 0;
-    EXPECT_EQ(next_records_offset % 64, 0u);
-
-    for (uint32_t cpu = 0; cpu < header->cpu_count; cpu++) {
-        EXPECT_EQ(entries[cpu].cpu_id, cpu);
-        EXPECT_EQ(entries[cpu].records_offset, next_records_offset);
-        next_records_offset += entries[cpu].record_count * sizeof(ktrace::trace_record);
-        total_records += entries[cpu].record_count;
-        if (entries[cpu].record_count > 0) {
-            recorded = &entries[cpu];
-        }
-    }
+    record_events(2);
 
-    EXPECT_EQ(total_records, 3u);
-    EXPECT_EQ(length, next_records_offset);
-    ASSERT_NOT_NULL(recorded);
+    ktrace::trace_record unselected = {};
+    unselected.hdr.event_id = TEST_EVENT_ID - 1;
+    ktrace::record_event(unselected);
 
-    const auto* records = reinterpret_cast<const ktrace::trace_record*>(g_trace_bytes + recorded->records_offset);
-    for (uint64_t i = 0; i < 3; i++) {
-        EXPECT_EQ(records[i].payload[0], i);
-        EXPECT_GE(records[i].hdr.timestamp, header->session_start_ns);
-    }
+    ktrace::trace_record out_of_range = {};
+    out_of_range.hdr.event_id = ktrace::EVENT_ID_COUNT;
+    ktrace::record_event(out_of_range);
+
+    ASSERT_EQ(ktrace::stop(), ktrace::OK);
+    size_t length = read_until_end(reader, g_stream_bytes, sizeof(g_stream_bytes));
+    fs::close(reader);
+
+    EXPECT_EQ(summarize_stream(g_stream_bytes, length).records, 2u);
 }
 
-TEST(ktrace, the_trace_lists_a_wrapped_ring_oldest_first) {
-    constexpr uint64_t overwritten = 10;
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
-    record_events(RING_RECORDS + overwritten);
+TEST(ktrace, events_outside_a_session_are_dropped) {
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
+
     ASSERT_EQ(ktrace::stop(), ktrace::OK);
+    record_events(5);
 
-    size_t cap = sizeof(g_trace_bytes) + RING_RECORDS * sizeof(ktrace::trace_record);
-    auto* trace = static_cast<uint8_t*>(heap::uzalloc(cap));
-    ASSERT_NOT_NULL(trace);
+    size_t length = read_until_end(reader, g_stream_bytes, sizeof(g_stream_bytes));
+    fs::close(reader);
 
-    read_trace(trace, cap, LARGE_READ_BYTES);
-    const auto* header = reinterpret_cast<const ktrace::file_header*>(trace);
-    const auto* entries = reinterpret_cast<const ktrace::cpu_table_entry*>(trace + header->cpu_table_offset);
+    EXPECT_EQ(summarize_stream(g_stream_bytes, length).records, 0u);
+}
 
-    const ktrace::cpu_table_entry* wrapped = nullptr;
-    for (uint32_t cpu = 0; cpu < header->cpu_count; cpu++) {
-        if (entries[cpu].flags & ktrace::CPU_FLAG_RING_WRAPPED) {
-            wrapped = &entries[cpu];
-        }
-    }
+TEST(ktrace, a_full_ring_drops_new_records_and_counts_them) {
+    constexpr uint64_t dropped = 10;
+    auto* stream = static_cast<uint8_t*>(heap::uzalloc(FULL_RING_STREAM_BYTES));
+    ASSERT_NOT_NULL(stream);
 
-    uint64_t out_of_order = 0;
-    if (wrapped) {
-        const auto* records = reinterpret_cast<const ktrace::trace_record*>(trace + wrapped->records_offset);
-        for (uint64_t i = 0; i < wrapped->record_count; i++) {
-            if (records[i].payload[0] != overwritten + i) {
-                out_of_order++;
-            }
-        }
-    }
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
 
-    EXPECT_NOT_NULL(wrapped);
-    EXPECT_EQ(wrapped ? wrapped->record_count : 0, RING_RECORDS);
-    EXPECT_EQ(out_of_order, 0u);
-    heap::ufree(trace);
+    record_events(RING_RECORDS + dropped);
+    ASSERT_EQ(ktrace::stop(), ktrace::OK);
+
+    size_t length = read_until_end(reader, stream, FULL_RING_STREAM_BYTES);
+    stream_summary summary = summarize_stream(stream, length);
+    EXPECT_EQ(summary.records, RING_RECORDS);
+    EXPECT_EQ(summary.lost, dropped);
+    EXPECT_EQ(summary.out_of_order, 0u);
+    EXPECT_TRUE(summary.ended);
+
+    ASSERT_TRUE(read_status());
+    EXPECT_EQ(sum_cpu_lines().lost, dropped);
+
+    fs::close(reader);
+    heap::ufree(stream);
 }
 
 static constexpr uint32_t RECORDING_CPU = 1;
@@ -385,11 +538,13 @@ static void record_on_cpu_task(void*) {
     sched::exit(0);
 }
 
-// CPU 0 records nothing, so a single read crosses its empty range on the way to CPU 1's records
-TEST(ktrace, the_trace_reads_past_a_cpu_that_recorded_nothing) {
+// CPU 0 records nothing, so its ring yields no chunk
+TEST(ktrace, the_stream_reads_past_a_cpu_that_recorded_nothing) {
     ASSERT_TRUE(smp::cpu_count() > RECORDING_CPU);
     g_recorder_done.store_relaxed(0);
-    ASSERT_EQ(ktrace::start(), ktrace::OK);
+
+    fs::file* reader = start_session_with_reader();
+    ASSERT_NOT_NULL(reader);
 
     bool created = false;
     RUN_ELEVATED({
@@ -403,17 +558,12 @@ TEST(ktrace, the_trace_reads_past_a_cpu_that_recorded_nothing) {
     EXPECT_TRUE(test_helpers::spin_wait(g_recorder_done));
     ASSERT_EQ(ktrace::stop(), ktrace::OK);
 
-    size_t length = read_trace(g_trace_bytes, sizeof(g_trace_bytes), LARGE_READ_BYTES);
-    const auto* header = reinterpret_cast<const ktrace::file_header*>(g_trace_bytes);
-    const auto* entries = reinterpret_cast<const ktrace::cpu_table_entry*>(g_trace_bytes + header->cpu_table_offset);
-    const auto& recorded = entries[RECORDING_CPU];
-    const auto* records = reinterpret_cast<const ktrace::trace_record*>(g_trace_bytes + recorded.records_offset);
+    size_t length = read_until_end(reader, g_stream_bytes, sizeof(g_stream_bytes));
+    fs::close(reader);
 
-    EXPECT_EQ(entries[0].record_count, 0u);
-    EXPECT_EQ(recorded.record_count, RECORDING_CPU_EVENTS);
-    EXPECT_EQ(length, recorded.records_offset + RECORDING_CPU_EVENTS * sizeof(ktrace::trace_record));
-
-    for (uint64_t i = 0; i < RECORDING_CPU_EVENTS; i++) {
-        EXPECT_EQ(records[i].payload[0], i);
-    }
+    stream_summary summary = summarize_stream(g_stream_bytes, length);
+    EXPECT_EQ(summary.records, RECORDING_CPU_EVENTS);
+    EXPECT_EQ(summary.cpus_with_records_mask, 1ull << RECORDING_CPU);
+    EXPECT_EQ(summary.out_of_order, 0u);
+    EXPECT_TRUE(summary.ended);
 }
diff --git a/kernel/trace/ktrace.cpp b/kernel/trace/ktrace.cpp
index df2c0c84..0730b0e7 100644
--- a/kernel/trace/ktrace.cpp
+++ b/kernel/trace/ktrace.cpp
@@ -34,20 +34,28 @@ constexpr uint32_t SESSION_RECORDING = 1;
 constexpr uint32_t SESSION_STOPPING  = 2;
 constexpr uint32_t SESSION_STOPPED   = 3;
 
-constexpr uint64_t SECTION_ALIGNMENT = 64;
+constexpr uint64_t DRAIN_INTERVAL_NS    = 100000000ULL;
+constexpr size_t   SMALLEST_CHUNK_BYTES = sizeof(chunk_header) + sizeof(trace_record);
 
 namespace {
 
 struct trace_ring {
     trace_record*          records;
-    sync::atomic<uint64_t> head;
-    sync::atomic<uint64_t> committed;
+    sync::atomic<uint64_t> head;      // Slots reserved, RING_CLOSED once stopped
+    sync::atomic<uint64_t> committed; // Records completely written
+    sync::atomic<uint64_t> drained;   // Records copied to the reader
+    sync::atomic<uint64_t> lost;      // Records dropped while the ring was full
+    uint64_t               lost_reported;
 };
 
-struct ring_summary {
-    uint64_t kept;
-    uint64_t oldest_slot;
-    uint32_t flags;
+struct trace_reader {
+    bool     session_started;
+    bool     header_sent;
+    bool     end_sent;
+    bool     read_in_progress;
+    bool     last_read_was_full;
+    uint32_t drain_start_cpu;
+    uint64_t last_drain_ns;
 };
 
 // /dev/ktrace/control, where each write is one command, `start` or `stop`
@@ -60,7 +68,7 @@ class control_node : public fs::node {
 
         int32_t rc;
         if (is_command(text, count, "start")) {
-            rc = start();
+            rc = start(ALL_EVENTS);
         } else if (is_command(text, count, "stop")) {
             rc = stop();
         } else {
@@ -71,6 +79,10 @@ class control_node : public fs::node {
             return fs::ERR_BUSY;
         }
 
+        if (rc == ERR_NO_READER) {
+            return fs::ERR_PIPE;
+        }
+
         if (rc != OK) {
             return fs::ERR_INVAL;
         }
@@ -106,7 +118,10 @@ static sync::spinlock g_session_lock = sync::SPINLOCK_INIT;
 static sync::atomic<uint32_t> g_session_state;
 static sync::atomic<uint64_t> g_session_start_ns;
 static sync::atomic<uint64_t> g_session_stop_ns;
+static sync::atomic<uint64_t> g_session_event_mask;
+static sync::atomic<uint64_t> g_bytes_streamed;
 static uint32_t g_trace_open_count;
+static trace_reader g_reader;
 
 /**
  * @note Privilege: **required**
@@ -143,26 +158,7 @@ static const char* session_state_name(uint32_t state) {
     }
 }
 
-static ring_summary summarize_ring(trace_ring& ring) {
-    uint64_t committed = ring.committed.load_acquire();
-    bool wrapped = committed > PERCPU_RECORD_BUFFER_RECORDS;
-
-    ring_summary summary = {};
-    summary.kept = wrapped ? PERCPU_RECORD_BUFFER_RECORDS : committed;
-    summary.oldest_slot = wrapped ? committed % PERCPU_RECORD_BUFFER_RECORDS : 0;
-
-    if (wrapped) {
-        summary.flags |= CPU_FLAG_RING_WRAPPED;
-    }
-
-    if (!ring.records) {
-        summary.flags |= CPU_FLAG_NO_RING;
-    }
-
-    return summary;
-}
-
-// `key value` lines, then one `cpu<N> <kept> <flags>` line per CPU with the CPU table flags
+// `key value` lines, then one `cpu<N> <records> <lost>` line per CPU
 static size_t generate_status(char* buf, size_t cap) {
     uint32_t state;
     uint64_t start_ns;
@@ -191,155 +187,246 @@ static size_t generate_status(char* buf, size_t cap) {
 
     pos = devfs::append_str(buf, cap, pos, "\ntrace_open ");
     pos = devfs::append_u64(buf, cap, pos, trace_open);
+    pos = devfs::append_str(buf, cap, pos, "\nbytes_streamed ");
+    pos = devfs::append_u64(buf, cap, pos, g_bytes_streamed.load_relaxed());
     pos = devfs::append_str(buf, cap, pos, "\n");
 
     uint32_t cpu_count = smp::cpu_count();
 
     for (uint32_t cpu = 0; cpu < cpu_count; cpu++) {
-        ring_summary summary = summarize_ring(per_cpu_on(ktrace_percpu_ring, cpu));
+        trace_ring& ring = per_cpu_on(ktrace_percpu_ring, cpu);
 
         pos = devfs::append_str(buf, cap, pos, "cpu");
         pos = devfs::append_u64(buf, cap, pos, cpu);
         pos = devfs::append_str(buf, cap, pos, " ");
-        pos = devfs::append_u64(buf, cap, pos, summary.kept);
+        pos = devfs::append_u64(buf, cap, pos, ring.drained.load_relaxed());
         pos = devfs::append_str(buf, cap, pos, " ");
-        pos = devfs::append_u64(buf, cap, pos, summary.flags);
+        pos = devfs::append_u64(buf, cap, pos, ring.lost.load_relaxed());
         pos = devfs::append_str(buf, cap, pos, "\n");
     }
 
     return pos;
 }
 
-static uint64_t records_start(uint32_t cpu_count) {
-    uint64_t table_end = sizeof(file_header) + cpu_count * sizeof(cpu_table_entry);
-    return (table_end + SECTION_ALIGNMENT - 1) / SECTION_ALIGNMENT * SECTION_ALIGNMENT;
-}
-
-static uint64_t trace_file_size(uint32_t cpu_count) {
-    uint64_t size = records_start(cpu_count);
-
-    for (uint32_t cpu = 0; cpu < cpu_count; cpu++) {
-        size += summarize_ring(per_cpu_on(ktrace_percpu_ring, cpu)).kept * sizeof(trace_record);
-    }
-
-    return size;
-}
-
-// Copies the bytes of a region that fall inside the read window [pos, end)
-static void copy_overlap(uint8_t* dst, uint64_t pos, uint64_t end,
-                         uint64_t region_offset, const void* region, uint64_t region_size) {
-    uint64_t from = pos > region_offset ? pos : region_offset;
-    uint64_t to = end < region_offset + region_size ? end : region_offset + region_size;
-    if (from >= to) {
-        return;
-    }
-
-    string::memcpy(dst + (from - pos), static_cast<const uint8_t*>(region) + (from - region_offset), to - from);
-}
-
-static file_header build_file_header(uint32_t cpu_count) {
+static file_header build_file_header() {
     file_header header = {};
     string::memcpy(header.magic, FILE_MAGIC, sizeof(header.magic));
+
     header.version = FILE_VERSION;
     header.header_size = sizeof(file_header);
     header.arch = exec::ELF_EXPECTED_MACHINE;
-    header.cpu_count = cpu_count;
+    header.cpu_count = smp::cpu_count();
     header.record_size = sizeof(trace_record);
-    header.cpu_table_offset = sizeof(file_header);
     header.boot_unix_ns = clock::boot_realtime_ns();
     header.session_start_ns = g_session_start_ns.load_relaxed();
-    header.session_stop_ns = g_session_stop_ns.load_relaxed();
+    header.event_mask = g_session_event_mask.load_relaxed();
+
     return header;
 }
 
-static void copy_records(uint8_t* dst, uint64_t pos, uint64_t end, const trace_ring& ring,
-                         const ring_summary& summary, uint64_t records_offset) {
-    uint64_t records_end = records_offset + summary.kept * sizeof(trace_record);
-    uint64_t from = pos > records_offset ? pos : records_offset;
-    uint64_t to = end < records_end ? end : records_end;
-    if (from >= to) {
-        return;
+// Only a moment with no write in progress shows where the written records end
+static bool find_written_end(trace_ring& ring, uint64_t* end) {
+    uint64_t reserved = ring.head.load_acquire() & ~RING_CLOSED;
+    uint64_t written = ring.committed.load_acquire();
+    if (written != reserved || (ring.head.load_acquire() & ~RING_CLOSED) != reserved) {
+        return false;
     }
 
-    uint64_t first = (from - records_offset) / sizeof(trace_record);
-    uint64_t last = (to - records_offset - 1) / sizeof(trace_record);
+    *end = reserved;
+    return true;
+}
 
-    for (uint64_t i = first; i <= last; i++) {
-        const trace_record& rec = ring.records[(summary.oldest_slot + i) % PERCPU_RECORD_BUFFER_RECORDS];
-        copy_overlap(dst, pos, end, records_offset + i * sizeof(trace_record), &rec, sizeof(rec));
+static size_t drain_ring_chunk(uint8_t* dst, size_t room, uint32_t cpu, trace_ring& ring) {
+    uint64_t end = 0;
+    if (!ring.records || room < sizeof(chunk_header) || !find_written_end(ring, &end)) {
+        return 0;
     }
-}
 
-static void copy_trace_bytes(uint8_t* dst, uint64_t pos, uint64_t end, uint32_t cpu_count) {
-    file_header header = build_file_header(cpu_count);
-    copy_overlap(dst, pos, end, 0, &header, sizeof(header));
+    uint64_t start = ring.drained.load_relaxed();
+    uint64_t lost = ring.lost.load_relaxed();
+    uint64_t fits = (room - sizeof(chunk_header)) / sizeof(trace_record);
+    uint64_t count = end - start < fits ? end - start : fits;
+
+    if (count == 0 && lost == ring.lost_reported) {
+        return 0;
+    }
+
+    chunk_header chunk = {};
+    chunk.kind = CHUNK_RECORDS;
+    chunk.cpu_id = cpu;
+    chunk.record_count = count;
+    chunk.lost_records = lost - ring.lost_reported;
+
+    uint64_t first = start % PERCPU_RECORD_BUFFER_RECORDS;
+    uint64_t before_wrap = PERCPU_RECORD_BUFFER_RECORDS - first;
+    uint64_t contiguous = count < before_wrap ? count : before_wrap;
 
-    uint64_t records_offset = records_start(cpu_count);
+    uint8_t* records = dst + sizeof(chunk);
+    string::memcpy(dst, &chunk, sizeof(chunk));
+    string::memcpy(records, &ring.records[first], contiguous * sizeof(trace_record));
+    string::memcpy(records + contiguous * sizeof(trace_record), ring.records,
+                   (count - contiguous) * sizeof(trace_record));
 
+    ring.lost_reported = lost;
+    ring.drained.store_release(start + count);
+
+    return sizeof(chunk) + count * sizeof(trace_record);
+}
+
+static bool all_rings_drained(uint32_t cpu_count) {
     for (uint32_t cpu = 0; cpu < cpu_count; cpu++) {
         trace_ring& ring = per_cpu_on(ktrace_percpu_ring, cpu);
-        ring_summary summary = summarize_ring(ring);
+        uint64_t reserved = ring.head.load_acquire() & ~RING_CLOSED;
+
+        bool records_left = ring.drained.load_relaxed() != reserved;
+        bool losses_unreported = ring.lost.load_relaxed() != ring.lost_reported;
+        if (records_left || losses_unreported) {
+            return false;
+        }
+    }
+
+    return true;
+}
+
+static int32_t wait_for_next_drain() {
+    while (true) {
+        uint32_t state;
+        bool started;
 
-        cpu_table_entry entry = {};
-        entry.cpu_id = cpu;
-        entry.flags = summary.flags;
-        entry.record_count = summary.kept;
-        entry.records_offset = records_offset;
+        {
+            sync::lock_guard guard(g_session_lock);
+
+            state = g_session_state.load_relaxed();
+            started = g_reader.session_started;
+        }
 
-        uint64_t entry_offset = sizeof(file_header) + cpu * sizeof(cpu_table_entry);
-        copy_overlap(dst, pos, end, entry_offset, &entry, sizeof(entry));
-        copy_records(dst, pos, end, ring, summary, records_offset);
+        uint64_t wait_ns = DRAIN_INTERVAL_NS;
+        if (started) {
+            if (state == SESSION_STOPPED || !g_reader.header_sent || g_reader.last_read_was_full) {
+                return fs::OK;
+            }
+
+            uint64_t since_drain = clock::now_ns() - g_reader.last_drain_ns;
+            if (since_drain >= DRAIN_INTERVAL_NS) {
+                return fs::OK;
+            }
+
+            wait_ns = DRAIN_INTERVAL_NS - since_drain;
+        }
 
-        records_offset += summary.kept * sizeof(trace_record);
+        if (sched::sleep_ns(wait_ns) != 0) {
+            return fs::ERR_INTR;
+        }
     }
 }
 
-int32_t trace_node::open(fs::file*, uint32_t) {
-    sync::lock_guard guard(g_session_lock);
+static ssize_t read_stream(uint8_t* dst, size_t count) {
+    if (g_reader.end_sent) {
+        return 0;
+    }
 
-    uint32_t state = g_session_state.load_acquire();
-    if (state == SESSION_IDLE) {
-        return fs::ERR_NODATA;
+    while (true) {
+        int32_t rc = wait_for_next_drain();
+        if (rc != fs::OK) {
+            return rc;
+        }
+
+        size_t pos = 0;
+        if (!g_reader.header_sent) {
+            file_header header = build_file_header();
+            string::memcpy(dst, &header, sizeof(header));
+
+            pos = sizeof(header);
+            g_reader.header_sent = true;
+        }
+
+        uint32_t cpu_count = smp::cpu_count();
+
+        for (uint32_t i = 0; i < cpu_count; i++) {
+            uint32_t cpu = (g_reader.drain_start_cpu + i) % cpu_count;
+
+            trace_ring& ring = per_cpu_on(ktrace_percpu_ring, cpu);
+            pos += drain_ring_chunk(dst + pos, count - pos, cpu, ring);
+        }
+
+        g_reader.drain_start_cpu = (g_reader.drain_start_cpu + 1) % cpu_count;
+
+        bool stopped = g_session_state.load_acquire() == SESSION_STOPPED;
+        if (stopped && count - pos >= sizeof(chunk_header) && all_rings_drained(cpu_count)) {
+            chunk_header end_chunk = {};
+            end_chunk.kind = CHUNK_END;
+            end_chunk.stop_ns = g_session_stop_ns.load_relaxed();
+
+            string::memcpy(dst + pos, &end_chunk, sizeof(end_chunk));
+            pos += sizeof(end_chunk);
+
+            g_reader.end_sent = true;
+        }
+
+        g_reader.last_read_was_full = !g_reader.end_sent && count - pos < SMALLEST_CHUNK_BYTES;
+        if (!g_reader.last_read_was_full) {
+            g_reader.last_drain_ns = clock::now_ns();
+        }
+
+        if (pos > 0) {
+            g_bytes_streamed.fetch_add_relaxed(pos);
+            return static_cast<ssize_t>(pos);
+        }
     }
+}
 
-    if (state != SESSION_STOPPED) {
+int32_t trace_node::open(fs::file*, uint32_t) {
+    sync::lock_guard guard(g_session_lock);
+
+    if (g_trace_open_count != 0) {
         return fs::ERR_BUSY;
     }
 
-    g_trace_open_count++;
+    g_trace_open_count = 1;
+    g_reader = {};
+
     return fs::OK;
 }
 
 int32_t trace_node::on_close(fs::file*) {
-    sync::lock_guard guard(g_session_lock);
+    bool recording;
+
+    {
+        sync::lock_guard guard(g_session_lock);
+
+        g_trace_open_count = 0;
+        recording = g_session_state.load_relaxed() == SESSION_RECORDING;
+    }
+
+    // Nothing else drains the rings, so the session ends with its reader
+    if (recording) {
+        stop();
+    }
 
-    g_trace_open_count--;
     return fs::OK;
 }
 
-ssize_t trace_node::read(fs::file* f, void* buf, size_t count, uint32_t) {
-    int64_t offset = f->offset();
-    if (offset < 0) {
+ssize_t trace_node::read(fs::file*, void* buf, size_t count, uint32_t) {
+    if (count < SMALLEST_CHUNK_BYTES) {
         return fs::ERR_INVAL;
     }
 
-    uint32_t cpu_count = smp::cpu_count();
-    uint64_t size = trace_file_size(cpu_count);
-    uint64_t pos = static_cast<uint64_t>(offset);
-    if (pos >= size) {
-        return 0;
-    }
+    {
+        sync::lock_guard guard(g_session_lock);
+
+        if (g_reader.read_in_progress) {
+            return fs::ERR_BUSY;
+        }
 
-    if (count > size - pos) {
-        count = size - pos;
+        g_reader.read_in_progress = true;
     }
 
-    auto* dst = static_cast<uint8_t*>(buf);
-    string::memset(dst, 0, count);
-    copy_trace_bytes(dst, pos, pos + count, cpu_count);
+    ssize_t rc = read_stream(static_cast<uint8_t*>(buf), count);
+
+    sync::lock_guard guard(g_session_lock);
 
-    f->set_offset(offset + static_cast<int64_t>(count));
-    return static_cast<ssize_t>(count);
+    g_reader.read_in_progress = false;
+    return rc;
 }
 
 __PRIVILEGED_CODE static int32_t add_node(fs::node* dir, fs::node* node, const char* path) {
@@ -350,6 +437,7 @@ __PRIVILEGED_CODE static int32_t add_node(fs::node* dir, fs::node* node, const c
 
     if (devfs::add_char_device_at(dir, node) != devfs::OK) {
         log::error("ktrace: failed to register %s", path);
+
         heap::kfree_delete(node);
         return ERR_DEVFS;
     }
@@ -361,6 +449,10 @@ __PRIVILEGED_CODE static int32_t add_node(fs::node* dir, fs::node* node, const c
  * @note Privilege: **required**
  */
 __PRIVILEGED_CODE int32_t register_devfs_nodes() {
+#if !defined(KTRACE_ENABLED) || KTRACE_ENABLED == 0
+    return OK;
+#endif
+
     fs::node* dir = devfs::ensure_dir("ktrace");
     if (!dir) {
         log::error("ktrace: failed to create /dev/ktrace");
@@ -381,7 +473,7 @@ __PRIVILEGED_CODE int32_t register_devfs_nodes() {
     return add_node(dir, heap::kalloc_new<trace_node>(), "/dev/ktrace/trace");
 }
 
-int32_t start() {
+int32_t start(uint64_t event_mask) {
     sync::lock_guard guard(g_session_lock);
 
     uint32_t state = g_session_state.load_acquire();
@@ -389,8 +481,8 @@ int32_t start() {
         return ERR_BUSY;
     }
 
-    if (g_trace_open_count != 0) {
-        return ERR_BUSY;
+    if (g_trace_open_count == 0 || g_reader.session_started) {
+        return ERR_NO_READER;
     }
 
     uint32_t cpu_count = smp::cpu_count();
@@ -398,15 +490,24 @@ int32_t start() {
     for (uint32_t cpu = 0; cpu < cpu_count; cpu++) {
         trace_ring& ring = per_cpu_on(ktrace_percpu_ring, cpu);
 
-        // `committed` is zeroed before the head reopens, so
-        // a writer that reserves right away keeps its commit.
+        // The counters are zeroed before the head reopens, so a writer
+        // that reserves right away sees an empty ring and keeps its commit.
         ring.committed.store_relaxed(0);
+        ring.drained.store_relaxed(0);
+        ring.lost.store_relaxed(0);
+        ring.lost_reported = 0;
         ring.head.store_release(0);
     }
 
     g_session_start_ns.store_relaxed(clock::now_ns());
     g_session_stop_ns.store_relaxed(0);
+    g_session_event_mask.store_relaxed(event_mask);
+
+    g_bytes_streamed.store_relaxed(0);
+    g_reader.session_started = true;
+
     g_session_state.store_release(SESSION_RECORDING);
+
     return OK;
 }
 
@@ -442,8 +543,13 @@ int32_t stop() {
 }
 
 #if defined(KTRACE_ENABLED) && KTRACE_ENABLED == 1
+bool is_recording(uint16_t event_id) {
+    return g_session_state.load_acquire() == SESSION_RECORDING && event_id < EVENT_ID_COUNT &&
+           (g_session_event_mask.load_relaxed() & (1ull << event_id));
+}
+
 void record_event(const trace_record& rec) {
-    if (g_session_state.load_acquire() != SESSION_RECORDING) {
+    if (!is_recording(rec.hdr.event_id)) {
         return;
     }
 
@@ -455,10 +561,17 @@ void record_event(const trace_record& rec) {
     }
 
     // Atomically reserve a slot in the ring buffer
-    uint64_t slot = ring.head.fetch_add_acquire(1);
-    if (slot & RING_CLOSED) {
-        return;
-    }
+    uint64_t slot = ring.head.load_acquire();
+    do {
+        if (slot & RING_CLOSED) {
+            return;
+        }
+
+        if (slot - ring.drained.load_acquire() >= PERCPU_RECORD_BUFFER_RECORDS) {
+            ring.lost.fetch_add_relaxed(1);
+            return;
+        }
+    } while (!ring.head.cmpxchg_weak_acquire(slot, slot + 1));
 
     // Insert the event record
     trace_record& entry = ring.records[slot % PERCPU_RECORD_BUFFER_RECORDS];
diff --git a/kernel/trace/ktrace.h b/kernel/trace/ktrace.h
index 01f5a63a..3774a467 100644
--- a/kernel/trace/ktrace.h
+++ b/kernel/trace/ktrace.h
@@ -4,28 +4,26 @@
 #include "common/types.h"
 
 /*
- * A .ktrace file holds one stopped session. Fields are little-endian and
- * offsets are relative to the start of the file. Each section starts on a
- * 64-byte boundary.
+ * A .ktrace file holds one session, streamed while it records. Fields are
+ * little-endian and every section is a multiple of 64 bytes.
  *
- *   offset 0           +-------------------------+
- *                      | file_header             |  64 bytes
- *   cpu_table_offset   +-------------------------+
- *                      | cpu_table_entry CPU 0   |  32 bytes each, cpu_count
- *                      | cpu_table_entry CPU 1   |  entries in CPU id order
- *                      | ...                     |
- *   records_offset     +-------------------------+
- *   of CPU 0           | trace_record            |  record_count records,
- *                      | ...                     |  oldest first
- *   records_offset     +-------------------------+
- *   of CPU 1           | trace_record            |
- *                      | ...                     |
- *                      +-------------------------+
+ *   +-------------------------+
+ *   | file_header             |  64 bytes
+ *   +-------------------------+
+ *   | chunk_header            |  64 bytes, CHUNK_RECORDS
+ *   | trace_record            |  record_count records of one CPU,
+ *   | ...                     |  oldest first
+ *   +-------------------------+
+ *   | ...                     |  more chunks, CPUs interleaved
+ *   +-------------------------+
+ *   | chunk_header            |  CHUNK_END, last in the file
+ *   +-------------------------+
  *
- * Readers sort records by timestamp, since an interrupt can record between a
- * writer's clock read and its slot reservation, and they drop records older
- * than session_start_ns. Fields are only appended to the header, growing
- * header_size, and version changes only for incompatible layouts.
+ * A file without a CHUNK_END chunk was cut short. Readers sort records by
+ * timestamp, since an interrupt can record between a writer's clock read and
+ * its slot reservation, and they drop records older than session_start_ns.
+ * Fields are only appended to the headers, and version changes only for
+ * incompatible layouts.
  */
 
 namespace ktrace {
@@ -35,12 +33,17 @@ constexpr int32_t ERR_NO_MEMORY     = -1;
 constexpr int32_t ERR_BUSY          = -2;
 constexpr int32_t ERR_NOT_RECORDING = -3;
 constexpr int32_t ERR_DEVFS         = -4;
+constexpr int32_t ERR_NO_READER     = -5;
 
 constexpr char     FILE_MAGIC[]  = "STLXKTRC";
 constexpr uint16_t FILE_VERSION  = 1;
 
-constexpr uint32_t CPU_FLAG_RING_WRAPPED = 1 << 0;
-constexpr uint32_t CPU_FLAG_NO_RING      = 1 << 1;
+constexpr uint32_t CHUNK_RECORDS = 1;
+constexpr uint32_t CHUNK_END     = 2;
+
+// An event mask holds one bit per event id
+constexpr uint16_t EVENT_ID_COUNT = 64;
+constexpr uint64_t ALL_EVENTS     = ~0ull;
 
 struct trace_record_header {
     uint64_t    timestamp;     // Nanoseconds since boot, set by record_event()
@@ -59,33 +62,32 @@ struct trace_record {
 } __attribute__((aligned(64)));
 
 static_assert(sizeof(trace_record) == 64, "trace_record size must be 64 bytes");
-
 struct file_header {
     char        magic[8];           // FILE_MAGIC without its terminator
     uint16_t    version;            // FILE_VERSION
     uint16_t    header_size;        // sizeof(file_header)
     uint16_t    arch;               // ELF machine number of the kernel
     uint16_t    reserved0;
-    uint32_t    cpu_count;          // Entries in the CPU table
+    uint32_t    cpu_count;
     uint32_t    record_size;        // sizeof(trace_record)
-    uint64_t    cpu_table_offset;
     uint64_t    boot_unix_ns;       // Wall-clock time at boot, 0 without an RTC
     uint64_t    session_start_ns;   // Nanoseconds since boot
-    uint64_t    session_stop_ns;    // Nanoseconds since boot
-    uint64_t    reserved1;
+    uint64_t    event_mask;         // Event ids the session recorded, one bit each
+    uint64_t    reserved1[2];
 } __attribute__((packed));
 
 static_assert(sizeof(file_header) == 64, "file_header size must be 64 bytes");
 
-struct cpu_table_entry {
+struct chunk_header {
+    uint32_t    kind;               // CHUNK_RECORDS or CHUNK_END
     uint32_t    cpu_id;
-    uint32_t    flags;              // CPU_FLAG_* bits
     uint64_t    record_count;
-    uint64_t    records_offset;
-    uint64_t    reserved;
+    uint64_t    lost_records;       // Records the CPU dropped since its previous chunk
+    uint64_t    stop_ns;            // Nanoseconds since boot, set in CHUNK_END
+    uint64_t    reserved[4];
 } __attribute__((packed));
 
-static_assert(sizeof(cpu_table_entry) == 32, "cpu_table_entry size must be 32 bytes");
+static_assert(sizeof(chunk_header) == 64, "chunk_header size must be 64 bytes");
 
 /**
  * @brief Initializes the kernel tracing and profiling subsystem for a given CPU.
@@ -94,35 +96,47 @@ static_assert(sizeof(cpu_table_entry) == 32, "cpu_table_entry size must be 32 by
 __PRIVILEGED_CODE int32_t init();
 
 /**
- * @brief Registers the /dev/ktrace nodes. Must be called after devfs is mounted.
+ * @brief Registers the /dev/ktrace nodes if ktrace is enabled.
  * @note Privilege: **required**
  */
 __PRIVILEGED_CODE int32_t register_devfs_nodes();
 
 /**
- * @brief Starts a session on every CPU, emptying the rings first.
- * @return OK, or ERR_BUSY while a session is recording.
+ * @brief Starts a session on every CPU that records the events set in `event_mask`,
+ * emptying the rings first. It streams to the reader of /dev/ktrace/trace.
+ * @return OK, ERR_BUSY while a session is recording, or ERR_NO_READER when the trace has
+ * no reader or its reader already streamed a session.
  */
-int32_t start();
+int32_t start(uint64_t event_mask);
 
 /**
  * @brief Stops the recording session. Returns once every record reserved in the session
- * is completely written, yielding the CPU while it waits. The rings keep the session's
- * records until the next start().
+ * is completely written, yielding the CPU while it waits. The reader then gets the rest.
  * @return OK, or ERR_NOT_RECORDING when no session is recording.
  */
 int32_t stop();
 
 /**
  * @brief Append a record to this CPU's trace ring, stamped with `clock::now_ns()`.
- * Does nothing unless a session is recording. Any timestamp already in `rec` is
- * replaced. Callable from any context at either privilege level.
+ * Does nothing unless a session is recording `rec`'s event id, and drops the record while
+ * the ring is full. Any timestamp already in `rec` is replaced. Callable from any context
+ * at either privilege level.
  */
 #if defined(KTRACE_ENABLED) && KTRACE_ENABLED == 1
     void record_event(const trace_record& rec);
 #else
     inline void record_event(const trace_record&) {}
 #endif
+
+/**
+ * @brief Whether a session is recording `event_id`, so typed recorders can return
+ * before building a record.
+ */
+#if defined(KTRACE_ENABLED) && KTRACE_ENABLED == 1
+    bool is_recording(uint16_t event_id);
+#else
+    inline bool is_recording(uint16_t) { return false; }
+#endif
 } // namespace ktrace
 
 #endif // STELLUX_TRACE_KTRACE_H
diff --git a/kernel/trace/ktrace_events.cpp b/kernel/trace/ktrace_events.cpp
new file mode 100644
index 00000000..e7fea286
--- /dev/null
+++ b/kernel/trace/ktrace_events.cpp
@@ -0,0 +1,114 @@
+#include "trace/ktrace_events.h"
+#include "common/string.h"
+
+namespace ktrace {
+
+#if defined(KTRACE_ENABLED) && KTRACE_ENABLED == 1
+
+static void copy_task_name(char* dst, const char* name) {
+    string::memcpy(dst, name, string::strnlen(name, TASK_NAME_BYTES));
+}
+
+void record_sched_switch(
+    uint32_t prev_tid,
+    const char* prev_name,
+    uint32_t next_tid,
+    const char* next_name,
+    uint8_t reason
+) {
+    if (!is_recording(EVENT_SCHED_SWITCH)) {
+        return;
+    }
+
+    sched_switch_payload payload = {};
+    payload.prev_tid = prev_tid;
+    payload.next_tid = next_tid;
+    payload.reason = reason;
+    copy_task_name(payload.prev_name, prev_name);
+    copy_task_name(payload.next_name, next_name);
+
+    trace_record rec = {};
+    rec.hdr.event_id = EVENT_SCHED_SWITCH;
+    rec.hdr.length = sizeof(trace_record) / sizeof(uint64_t);
+    string::memcpy(rec.payload, &payload, sizeof(payload));
+    record_event(rec);
+}
+
+void record_sched_wakeup(
+    uint32_t woken_tid,
+    const char* woken_name,
+    uint32_t target_cpu,
+    uint32_t waker_tid,
+    const char* waker_name
+) {
+    if (!is_recording(EVENT_SCHED_WAKEUP)) {
+        return;
+    }
+
+    sched_wakeup_payload payload = {};
+    payload.woken_tid = woken_tid;
+    payload.target_cpu = target_cpu;
+    payload.waker_tid = waker_tid;
+    copy_task_name(payload.woken_name, woken_name);
+    copy_task_name(payload.waker_name, waker_name);
+
+    trace_record rec = {};
+    rec.hdr.event_id = EVENT_SCHED_WAKEUP;
+    rec.hdr.length = sizeof(trace_record) / sizeof(uint64_t);
+    string::memcpy(rec.payload, &payload, sizeof(payload));
+    record_event(rec);
+}
+
+void record_syscall(
+    uint32_t tid,
+    uint32_t pid,
+    uint32_t number,
+    uint64_t duration_ns,
+    int64_t result
+) {
+    if (!is_recording(EVENT_SYSCALL)) {
+        return;
+    }
+
+    syscall_payload payload = {};
+    payload.duration_ns = duration_ns;
+    payload.result = result;
+    payload.number = number;
+    payload.tid = tid;
+    payload.pid = pid;
+
+    trace_record rec = {};
+    rec.hdr.event_id = EVENT_SYSCALL;
+    rec.hdr.length = sizeof(trace_record) / sizeof(uint64_t);
+    string::memcpy(rec.payload, &payload, sizeof(payload));
+    record_event(rec);
+}
+
+void record_page_fault(
+    uint32_t tid,
+    uint64_t address,
+    uint64_t flags,
+    uint64_t duration_ns,
+    int32_t result
+) {
+    if (!is_recording(EVENT_PAGE_FAULT)) {
+        return;
+    }
+
+    page_fault_payload payload = {};
+    payload.address = address;
+    payload.flags = flags;
+    payload.duration_ns = duration_ns;
+    payload.result = result;
+    payload.tid = tid;
+
+    trace_record rec = {};
+    rec.hdr.event_id = EVENT_PAGE_FAULT;
+    rec.hdr.length = sizeof(trace_record) / sizeof(uint64_t);
+    string::memcpy(rec.payload, &payload, sizeof(payload));
+    record_event(rec);
+}
+
+#endif
+
+} // namespace ktrace
diff --git a/kernel/trace/ktrace_events.h b/kernel/trace/ktrace_events.h
new file mode 100644
index 00000000..370ea4db
--- /dev/null
+++ b/kernel/trace/ktrace_events.h
@@ -0,0 +1,102 @@
+#ifndef STELLUX_TRACE_KTRACE_EVENTS_H
+#define STELLUX_TRACE_KTRACE_EVENTS_H
+
+#include "trace/ktrace.h"
+
+namespace ktrace {
+
+constexpr uint16_t EVENT_SCHED_SWITCH = 1;
+constexpr uint16_t EVENT_SCHED_WAKEUP = 2;
+constexpr uint16_t EVENT_SYSCALL      = 3;
+constexpr uint16_t EVENT_PAGE_FAULT   = 4;
+
+constexpr uint8_t SWITCH_REASON_PREEMPTED = 0;
+constexpr uint8_t SWITCH_REASON_YIELDED   = 1;
+constexpr uint8_t SWITCH_REASON_BLOCKED   = 2;
+constexpr uint8_t SWITCH_REASON_EXITED    = 3;
+
+constexpr size_t TASK_NAME_BYTES = 16;
+
+struct sched_switch_payload {
+    uint32_t    prev_tid;
+    uint32_t    next_tid;
+    uint8_t     reason;
+    uint8_t     reserved[7];
+    char        prev_name[TASK_NAME_BYTES];
+    char        next_name[TASK_NAME_BYTES];
+} __attribute__((packed));
+
+static_assert(sizeof(sched_switch_payload) == sizeof(trace_record::payload),
+              "sched_switch_payload must fill a record payload");
+
+struct sched_wakeup_payload {
+    uint32_t    woken_tid;
+    uint32_t    target_cpu;
+    uint32_t    waker_tid;
+    uint8_t     reserved[4];
+    char        woken_name[TASK_NAME_BYTES];
+    char        waker_name[TASK_NAME_BYTES];
+} __attribute__((packed));
+
+static_assert(sizeof(sched_wakeup_payload) == sizeof(trace_record::payload),
+              "sched_wakeup_payload must fill a record payload");
+
+struct syscall_payload {
+    uint64_t    duration_ns;
+    int64_t     result;
+    uint32_t    number;
+    uint32_t    tid;
+    uint32_t    pid;
+    uint8_t     reserved[20];
+} __attribute__((packed));
+
+static_assert(sizeof(syscall_payload) == sizeof(trace_record::payload),
+              "syscall_payload must fill a record payload");
+
+struct page_fault_payload {
+    uint64_t    address;
+    uint64_t    flags;
+    uint64_t    duration_ns;
+    int32_t     result;
+    uint32_t    tid;
+    uint8_t     reserved[16];
+} __attribute__((packed));
+
+static_assert(sizeof(page_fault_payload) == sizeof(trace_record::payload),
+              "page_fault_payload must fill a record payload");
+
+#if defined(KTRACE_ENABLED) && KTRACE_ENABLED == 1
+    /**
+    * @brief Records this CPU switching from task `prev_tid` to `next_tid` when the session
+    * records EVENT_SCHED_SWITCH.
+    */
+    void record_sched_switch(uint32_t prev_tid, const char* prev_name,
+                             uint32_t next_tid, const char* next_name, uint8_t reason);
+
+    /**
+    * @brief Records task `woken_tid` becoming runnable on `target_cpu`.
+    */
+    void record_sched_wakeup(uint32_t woken_tid, const char* woken_name, uint32_t target_cpu,
+                             uint32_t waker_tid, const char* waker_name);
+
+    /**
+    * @brief Records one completed syscall of task `tid`.
+    */
+    void record_syscall(uint32_t tid, uint32_t pid, uint32_t number, uint64_t duration_ns,
+                        int64_t result);
+
+    /**
+    * @brief Records one page fault of task `tid` at `address`.
+    */
+    void record_page_fault(uint32_t tid, uint64_t address, uint64_t flags, uint64_t duration_ns,
+                           int32_t result);
+#else
+    inline void record_sched_switch(uint32_t, const char*, uint32_t, const char*, uint8_t) {}
+    inline void record_sched_wakeup(uint32_t, const char*, uint32_t, uint32_t, const char*) {}
+    inline void record_syscall(uint32_t, uint32_t, uint32_t, uint64_t, int64_t) {}
+    inline void record_page_fault(uint32_t, uint64_t, uint64_t, uint64_t, int32_t) {}
+#endif
+
+} // namespace ktrace
+
+#endif // STELLUX_TRACE_KTRACE_EVENTS_H
diff --git a/userland/apps/ktrace/src/ktrace.c b/userland/apps/ktrace/src/ktrace.c
index 05f2b897..84f37d3d 100644
--- a/userland/apps/ktrace/src/ktrace.c
+++ b/userland/apps/ktrace/src/ktrace.c
@@ -2,6 +2,7 @@
 
 #include <errno.h>
 #include <fcntl.h>
+#include <signal.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdio.h>
@@ -9,21 +10,26 @@
 #include <string.h>
 #include <time.h>
 #include <unistd.h>
+#include <stlx/proc.h>
 
 #define CONTROL_PATH "/dev/ktrace/control"
 #define STATUS_PATH  "/dev/ktrace/status"
 #define TRACE_PATH   "/dev/ktrace/trace"
+#define NULL_PATH    "/dev/null"
+#define SELF_PATH    "/bin/ktrace"
 
-#define DUMP_CHUNK_BYTES (64 * 1024)
-
-#define CPU_FLAG_RING_WRAPPED 0x1
-#define CPU_FLAG_NO_RING      0x2
+#define TRACE_READ_BYTES (1024 * 1024)
 
 #define NS_PER_SEC  1000000000ull
 #define NS_PER_MSEC 1000000ull
 
+#define STATUS_POLL_NS   (10 * NS_PER_MSEC)
+#define START_TIMEOUT_NS (3 * NS_PER_SEC)
+#define STOP_TIMEOUT_NS  (30 * NS_PER_SEC)
+
 static char g_status[16384];
-static char g_dump_buffer[DUMP_CHUNK_BYTES];
+static char g_trace_buffer[TRACE_READ_BYTES];
+static volatile sig_atomic_t g_interrupted;
 
 static int read_status(void) {
     int fd = open(STATUS_PATH, O_RDONLY);
@@ -73,6 +79,36 @@ static bool status_state_is(const char* state) {
     return field && strncmp(field, state, length) == 0 && field[length] == '\n';
 }
 
+// Parses a `cpu<N> <records> <lost>` status line
+static bool parse_cpu_line(const char* line, unsigned long* cpu, unsigned long long* records,
+                           unsigned long long* lost) {
+    if (strncmp(line, "cpu", 3) != 0) {
+        return false;
+    }
+
+    char* end = NULL;
+    *cpu = strtoul(line + 3, &end, 10);
+    *records = strtoull(end, &end, 10);
+    *lost = strtoull(end, &end, 10);
+
+    return true;
+}
+
+static unsigned long long sum_lost_records(void) {
+    unsigned long long total = 0;
+
+    for (const char* line = g_status; *line; line = next_line(line)) {
+        unsigned long cpu;
+        unsigned long long records;
+        unsigned long long lost;
+        if (parse_cpu_line(line, &cpu, &records, &lost)) {
+            total += lost;
+        }
+    }
+
+    return total;
+}
+
 static uint64_t monotonic_ns(void) {
     struct timespec now;
     clock_gettime(CLOCK_MONOTONIC, &now);
@@ -111,24 +147,15 @@ static void print_session(void) {
 }
 
 static void print_cpus(void) {
-    printf("\ncpu  records  flags\n");
+    printf("\ncpu  records  lost\n");
 
     for (const char* line = g_status; *line; line = next_line(line)) {
-        if (strncmp(line, "cpu", 3) != 0) {
-            continue;
-        }
-
-        char* end = NULL;
-        unsigned long cpu = strtoul(line + 3, &end, 10);
-        unsigned long long kept = strtoull(end, &end, 10);
-        unsigned long flags = strtoul(end, &end, 10);
-
-        if (flags == 0) {
-            printf("%-4lu %llu\n", cpu, kept);
-            continue;
+        unsigned long cpu;
+        unsigned long long records;
+        unsigned long long lost;
+        if (parse_cpu_line(line, &cpu, &records, &lost)) {
+            printf("%-4lu %-8llu %llu\n", cpu, records, lost);
         }
-
-        printf("%-4lu %-8llu %s\n", cpu, kept, (flags & CPU_FLAG_RING_WRAPPED) ? "wrapped" : "no ring");
     }
 }
 
@@ -142,6 +169,7 @@ static int show_status(void) {
     printf("state         %.*s\n", state ? (int)strcspn(state, "\n") : 0, state ? state : "");
     print_session();
     printf("trace open    %llu\n", (unsigned long long)status_u64("trace_open"));
+    printf("streamed      %llu bytes\n", (unsigned long long)status_u64("bytes_streamed"));
     printf("capacity      %llu records per CPU\n", (unsigned long long)status_u64("capacity"));
     print_cpus();
     return 0;
@@ -159,36 +187,40 @@ static int send_command(const char* command) {
     return err;
 }
 
-static int start_session(void) {
-    int err = send_command("start");
-    if (err == 0) {
-        return 0;
-    }
-
-    if (err == EBUSY && read_status() == 0 && status_u64("trace_open") > 0) {
-        fprintf(stderr, "ktrace: cannot start: the trace is still open\n");
-    } else if (err == EBUSY) {
+static void report_start_error(int err) {
+    if (err == EBUSY) {
         fprintf(stderr, "ktrace: cannot start: a session is already recording\n");
+    } else if (err == EPIPE) {
+        fprintf(stderr, "ktrace: cannot start: nothing is reading the trace\n");
     } else {
         fprintf(stderr, "ktrace: cannot start: %s\n", strerror(err));
     }
+}
 
-    return 1;
+static bool session_is_recording(void) {
+    return status_state_is("recording") && status_u64("trace_open") == 1;
 }
 
-static int stop_session(void) {
-    int err = send_command("stop");
-    if (err == 0) {
-        return 0;
-    }
+static bool reader_is_closed(void) {
+    return status_u64("trace_open") == 0;
+}
 
-    if (err == EINVAL) {
-        fprintf(stderr, "ktrace: cannot stop: no session is recording\n");
-    } else {
-        fprintf(stderr, "ktrace: cannot stop: %s\n", strerror(err));
-    }
+// Rereads the status until `condition` holds, leaving the last read in g_status
+static bool poll_status(bool (*condition)(void), uint64_t timeout_ns) {
+    uint64_t deadline_ns = monotonic_ns() + timeout_ns;
 
-    return 1;
+    while (true) {
+        if (read_status() == 0 && condition()) {
+            return true;
+        }
+
+        if (monotonic_ns() >= deadline_ns) {
+            return false;
+        }
+
+        struct timespec pause = { 0, STATUS_POLL_NS };
+        nanosleep(&pause, NULL);
+    }
 }
 
 static bool write_all(int fd, const char* data, size_t length) {
@@ -205,17 +237,58 @@ static bool write_all(int fd, const char* data, size_t length) {
     return true;
 }
 
-static int dump_trace(const char* path) {
-    int trace = open(TRACE_PATH, O_RDONLY);
-    if (trace < 0) {
-        if (errno == ENODATA) {
-            fprintf(stderr, "ktrace: cannot dump: no session has been recorded\n");
-        } else if (errno == EBUSY) {
-            fprintf(stderr, "ktrace: cannot dump: a session is still recording\n");
-        } else {
-            fprintf(stderr, "ktrace: %s: %s\n", TRACE_PATH, strerror(errno));
+static void on_interrupt(int signal_number) {
+    (void)signal_number;
+    g_interrupted = 1;
+}
+
+// Without SA_RESTART, Ctrl-C ends the blocked read so the session can be stopped
+static bool install_interrupt_handler(void) {
+    struct sigaction action;
+    memset(&action, 0, sizeof(action));
+    action.sa_handler = on_interrupt;
+    sigemptyset(&action.sa_mask);
+
+    return sigaction(SIGINT, &action, NULL) == 0;
+}
+
+// Returns 0 once the stream ends, or the errno value that cut it short
+static int drain_trace(int trace, int out, unsigned long long* written) {
+    bool stop_sent = false;
+
+    while (true) {
+        ssize_t n = read(trace, g_trace_buffer, sizeof(g_trace_buffer));
+        if (n == 0) {
+            return 0;
+        }
+
+        if (n < 0 && errno == EINTR) {
+            if (g_interrupted && !stop_sent) {
+                // Fails harmlessly when `ktrace stop` ended the session first
+                send_command("stop");
+                stop_sent = true;
+            }
+
+            continue;
+        }
+
+        if (n < 0 || !write_all(out, g_trace_buffer, (size_t)n)) {
+            return errno;
         }
 
+        *written += (unsigned long long)n;
+    }
+}
+
+static int record_session(const char* path) {
+    if (!install_interrupt_handler()) {
+        fprintf(stderr, "ktrace: cannot handle Ctrl-C: %s\n", strerror(errno));
+        return 1;
+    }
+
+    int trace = open(TRACE_PATH, O_RDONLY);
+    if (trace < 0) {
+        report_start_error(errno);
         return 1;
     }
 
@@ -226,39 +299,138 @@ static int dump_trace(const char* path) {
         return 1;
     }
 
-    unsigned long long total = 0;
-    int err = 0;
-    while (true) {
-        ssize_t n = read(trace, g_dump_buffer, sizeof(g_dump_buffer));
-        if (n == 0) {
-            break;
-        }
+    int err = send_command("start");
+    if (err != 0) {
+        report_start_error(err);
+        close(out);
+        close(trace);
+        return 1;
+    }
+
+    printf("recording to %s\n", path);
+
+    unsigned long long written = 0;
+    err = drain_trace(trace, out, &written);
+
+    close(out);
+    close(trace);
+
+    if (err != 0) {
+        fprintf(stderr, "ktrace: cannot record to %s: %s\n", path, strerror(err));
+        return 1;
+    }
+
+    unsigned long long lost = read_status() == 0 ? sum_lost_records() : 0;
+    printf("wrote %llu bytes to %s, %llu records lost\n", written, path, lost);
+
+    return 0;
+}
+
+// The process must not have started yet
+static int detach_standard_streams(int process) {
+    int null_fd = open(NULL_PATH, O_RDWR);
+    if (null_fd < 0) {
+        return errno;
+    }
 
-        if (n < 0 || !write_all(out, g_dump_buffer, (size_t)n)) {
+    int err = 0;
+    for (int slot = STDIN_FILENO; slot <= STDERR_FILENO && err == 0; slot++) {
+        if (proc_set_handle(process, slot, null_fd) != 0) {
             err = errno;
-            break;
         }
+    }
+
+    close(null_fd);
+    return err;
+}
 
-        total += (unsigned long long)n;
+// Runs `ktrace record -o path` on its own. It must hold none of this command's
+// streams, or an SSH session that ran `ktrace start` cannot close.
+static int spawn_detached_recorder(const char* path) {
+    const char* argv[] = { "record", "-o", path, NULL };
+    int recorder = proc_create(SELF_PATH, argv);
+    if (recorder < 0) {
+        return errno;
+    }
+
+    int err = detach_standard_streams(recorder);
+    if (err == 0 && proc_start(recorder) != 0) {
+        err = errno;
+    }
+
+    if (err != 0) {
+        close(recorder);
+        return err;
+    }
+
+    proc_detach(recorder);
+    return 0;
+}
+
+static int start_detached_recording(const char* path) {
+    if (read_status() != 0) {
+        fprintf(stderr, "ktrace: %s: %s\n", STATUS_PATH, strerror(errno));
+        return 1;
+    }
+
+    if (status_u64("trace_open") > 0) {
+        report_start_error(EBUSY);
+        return 1;
+    }
+
+    int out = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0644);
+    if (out < 0) {
+        fprintf(stderr, "ktrace: %s: %s\n", path, strerror(errno));
+        return 1;
     }
 
     close(out);
-    close(trace);
 
-    // A partial dump would pass for a complete one, so failures remove it
+    int err = spawn_detached_recorder(path);
     if (err != 0) {
-        fprintf(stderr, "ktrace: cannot dump to %s: %s\n", path, strerror(err));
-        unlink(path);
+        fprintf(stderr, "ktrace: cannot start the recorder: %s\n", strerror(err));
         return 1;
     }
 
-    printf("wrote %llu bytes to %s\n", total, path);
+    if (!poll_status(session_is_recording, START_TIMEOUT_NS)) {
+        fprintf(stderr, "ktrace: cannot start: the recorder did not start a session\n");
+        return 1;
+    }
+
+    printf("recording to %s\n", path);
+    return 0;
+}
+
+static int stop_session(void) {
+    int err = send_command("stop");
+    if (err == EINVAL) {
+        fprintf(stderr, "ktrace: cannot stop: no session is recording\n");
+        return 1;
+    }
+
+    if (err != 0) {
+        fprintf(stderr, "ktrace: cannot stop: %s\n", strerror(err));
+        return 1;
+    }
+
+    if (!poll_status(reader_is_closed, STOP_TIMEOUT_NS)) {
+        fprintf(stderr, "ktrace: stopped, but the recorder has not finished the file\n");
+        return 1;
+    }
+
+    printf("streamed %llu bytes, %llu records lost\n",
+           (unsigned long long)status_u64("bytes_streamed"), sum_lost_records());
+
     return 0;
 }
 
 int main(int argc, char* argv[]) {
-    if (argc == 2 && strcmp(argv[1], "start") == 0) {
-        return start_session();
+    if (argc == 4 && strcmp(argv[1], "start") == 0 && strcmp(argv[2], "-o") == 0) {
+        return start_detached_recording(argv[3]);
+    }
+
+    if (argc == 4 && strcmp(argv[1], "record") == 0 && strcmp(argv[2], "-o") == 0) {
+        return record_session(argv[3]);
     }
 
     if (argc == 2 && strcmp(argv[1], "stop") == 0) {
@@ -269,10 +441,6 @@ int main(int argc, char* argv[]) {
         return show_status();
     }
 
-    if (argc == 4 && strcmp(argv[1], "dump") == 0 && strcmp(argv[2], "-o") == 0) {
-        return dump_trace(argv[3]);
-    }
-
-    fprintf(stderr, "usage: ktrace start | stop | status | dump -o FILE\n");
+    fprintf(stderr, "usage: ktrace start -o FILE | record -o FILE | stop | status\n");
     return 1;
 }