kernel: tagged log ring — per-line pid/name/level records, klog_status

Replace the linear keep-earliest RAM buffer with a 512 KiB ring of
framed records (log-ring.zig, host-tested): every debug_write becomes
one record per payload line, stamped by the kernel with the sender's
pid, task name (its binary path), level, per-boot sequence number, and
monotonic timestamp. Attribution is structural — a payload cannot forge
another sender's tag, and newline injection lands inside the forger's
own next record. Oldest records are overwritten when full; sequence
gaps make the loss countable.

debug_write gains a level argument (err/warn/info/debug/raw; old
two-arg callers clamp to raw). klog_read becomes a stream-offset read
that fails once the cursor falls behind the ring's tail; the new
klog_status (#45) returns the cursors plus the boot wall-clock anchor —
what the logger service will name per-boot log directories with.

The log now guards itself with a dedicated spinlock (BKL -> log lock
order, never the reverse); panic paths use a bounded try-acquire and
fall back to sinks-only. Serial rendering keeps the historical
transcript byte-identical for kernel and legacy raw output; leveled
records get a kernel-rendered name prefix. log-flush/init's interim
drains start at the ring tail and write framed records until the logger
service replaces them.

build.zig

@@ -875,6 +875,7 @@ pub fn build(b: *std.Build) void {
     for ([_][]const u8{
         "system/boot-handoff.zig",
         "system/abi.zig",
+        "system/initial-ramdisk.zig", // v2 path-named entries: find/basename/magic
         "system/devices/device-abi.zig",
         "system/devices/pci-class.zig", // class/subclass/prog-IF name decoding
         "system/devices/acpi-ids.zig", // _HID name decoding
@@ -921,6 +922,21 @@ pub fn build(b: *std.Build) void {
     });
     test_step.dependOn(&b.addRunArtifact(xkb_tests).step);
 
+    // The tagged kernel log ring: append/wrap/reclaim/sequence-gap behavior over
+    // a RAM buffer. Needs the `abi` module (record header layout), so it doesn't
+    // fit the plain loop above.
+    const log_ring_tests = b.addTest(.{
+        .root_module = b.createModule(.{
+            .root_source_file = b.path("system/kernel/log-ring.zig"),
+            .target = target,
+            .optimize = optimize,
+            .imports = &.{
+                .{ .name = "abi", .module = abi_module },
+            },
+        }),
+    });
+    test_step.dependOn(&b.addRunArtifact(log_ring_tests).step);
+
     // runtime.time's Instant/Duration arithmetic. time.zig pulls in system.zig (the
     // syscall wrappers), which needs the `abi` module, so it doesn't fit the plain
     // loop above.

library/runtime/system.zig

@@ -21,10 +21,24 @@ pub fn yield() void {
     _ = sc.systemCall0(.yield);
 }
 
-/// Write raw bytes to the kernel log (a bring-up diagnostic; real output goes
-/// through the console/VFS later). Returns the byte count, or a wrapped -1.
+/// The tagged-log level of a record — re-exported so runtime.log and the logger
+/// service don't import `abi` themselves.
+pub const KlogLevel = abi.KlogLevel;
+pub const KlogStatus = abi.KlogStatus;
+pub const KlogRecordHeader = abi.KlogRecordHeader;
+
+/// Write raw bytes to the kernel log (bring-up/panic diagnostics; ordinary
+/// output goes through std.log -> writeRecord). The kernel stamps the record
+/// with this process's id and name. Returns the byte count, or a wrapped -1.
 pub fn write(message: []const u8) usize {
-    return sc.systemCall2(.debug_write, @intFromPtr(message.ptr), message.len);
+    return writeRecord(.raw, message);
+}
+
+/// Emit one leveled record into the tagged kernel log ring. The kernel stamps
+/// pid/name/sequence/timestamp; the payload should be a single line (embedded
+/// newlines split into further records).
+pub fn writeRecord(level: KlogLevel, message: []const u8) usize {
+    return sc.systemCall3(.debug_write, @intFromPtr(message.ptr), message.len, @intFromEnum(level));
 }
 
 /// Block the caller for `ms` milliseconds.
@@ -59,14 +73,25 @@ pub fn wallClock() u64 {
     return @intCast(sc.systemCall0(.wall_clock));
 }
 
-/// Copy bytes out of the kernel's in-memory diagnostic log — the accumulated
-/// stream of everything `write` (and the kernel itself) has emitted — starting at
-/// `offset`, into `out`. Returns the number of bytes copied (0 at end of buffer).
-/// A program reads the whole log by looping from offset 0, advancing by the return
-/// value, until it gets 0. This is how the boot log is persisted to disk on a
-/// headless/real machine where serial output is otherwise lost.
-pub fn klogRead(offset: usize, out: []u8) usize {
-    return sc.systemCall3(.klog_read, offset, @intFromPtr(out.ptr), out.len);
+/// Copy bytes out of the tagged kernel log ring — framed records of everything
+/// every process (and the kernel) has emitted — starting at stream offset
+/// `offset`, into `out`. Returns the byte count (0 = caught up), or null when
+/// `offset` fell behind the ring's tail (those records were overwritten) or
+/// lies past its head; re-sync via `klogStatus`. A reader parses
+/// [KlogRecordHeader][name][message] frames (8-byte aligned) from the bytes.
+pub fn klogRead(offset: u64, out: []u8) ?usize {
+    const r = sc.systemCall3(.klog_read, offset, @intFromPtr(out.ptr), out.len);
+    if (@as(isize, @bitCast(r)) < 0) return null;
+    return r;
+}
+
+/// The log ring's live cursors (oldest retained offset, end of stream, next
+/// sequence number) plus the wall-clock time of boot — how a log reader starts,
+/// detects loss, and names a per-boot log directory.
+pub fn klogStatus() ?KlogStatus {
+    var status: KlogStatus = undefined;
+    if (@as(isize, @bitCast(sc.systemCall1(.klog_status, @intFromPtr(&status)))) != 0) return null;
+    return status;
 }
 
 /// End the process. Never returns.

system/abi.zig

@@ -58,7 +58,7 @@ pub const SystemCall = enum(u64) {
     signal_bind = 29, // signal_bind(endpoint) -> 0/-errno: nominate the endpoint this process's signals arrive on
     process_signal = 30, // process_signal(id, signal) -> 0/-errno: post a signal to a child (or to yourself)
     timer_bind = 31, // timer_bind(endpoint, ms) -> 0/-errno: one-shot timer — posts a notification when ms elapse
-    klog_read = 32, // klog_read(offset, ptr, len) -> bytes copied: copy the kernel RAM log buffer out to a user buffer (for persisting the boot log to disk)
+    klog_read = 32, // klog_read(offset, ptr, len) -> bytes copied: copy tagged log-ring stream bytes from `offset` out to a user buffer; fails once `offset` falls behind the ring's tail (re-sync via klog_status)
     wall_clock = 33, // wall_clock() -> Unix epoch seconds (UTC): the RTC wall-clock time, for filesystem timestamps (mtime). Monotonic time is `clock`.
     shm_create = 34, // shm_create(len) -> virtual_address (rax), handle (rdx): a shareable, zeroed, cacheable RAM region mapped into this AS; the handle is a capability passed to another process as an ipc_call send_cap (docs/display-v2.md)
     shm_map = 35, // shm_map(cap) -> virtual_address: map the shared region named by a received capability into this address space (the same physical pages the creator sees)
@@ -71,6 +71,7 @@ pub const SystemCall = enum(u64) {
     thread_self = 42, // thread_self() -> tid: the calling thread's kernel task id (runtime.Thread.getCurrentId)
     thread_join = 43, // thread_join(tid) -> 0: block until the thread with id `tid` has exited (runtime.Thread.join; no per-thread IPC endpoint) (docs/threading.md)
     set_thread_pointer = 44, // set_thread_pointer(addr) -> 0: set the caller's thread pointer (user-space TLS base; x86_64 IA32_FS_BASE, aarch64 TPIDR_EL0); restored per task across context switches (docs/threading-plan.md M10)
+    klog_status = 45, // klog_status(ptr) -> 0: copy a KlogStatus (ring cursors + the boot wall-clock anchor) out to a user buffer
     _,
 };
 
@@ -187,6 +188,53 @@ pub const ProcessDescriptor = extern struct {
     name: [maximum_process_name]u8, // argv[0] at spawn; empty for kernel tasks
 };
 
+// --- the tagged kernel log ring (klog) ---------------------------------------
+// Every `debug_write` becomes one RECORD per payload line, stamped by the kernel
+// with the sender's pid, task name (its binary path), level, a per-boot sequence
+// number, and a monotonic timestamp. `klog_read` copies raw stream bytes — a
+// reader parses [KlogRecordHeader][name][message] frames, each padded to
+// `klog_record_alignment`. Sequence gaps tell a reader exactly how many records
+// the ring overwrote while it wasn't looking.
+
+/// Log level of a klog record — std.log's levels plus `raw` (untagged bytes:
+/// kernel prints and legacy runtime.system.write output).
+pub const KlogLevel = enum(u8) { err = 0, warn = 1, info = 2, debug = 3, raw = 4 };
+
+/// "RK" — leads every record; a parser's resync/corruption guard.
+pub const klog_record_magic: u16 = 0x4B52;
+
+/// KlogRecordHeader.flags bit: the emitter truncated the payload to fit.
+pub const klog_flag_truncated: u8 = 1;
+
+/// Header of one ring record, followed by `name_len` bytes of task name and
+/// `message_len` bytes of payload; the whole record is padded to 8 bytes.
+pub const KlogRecordHeader = extern struct {
+    magic: u16, // klog_record_magic
+    level: KlogLevel,
+    name_len: u8, // 0..maximum_process_name
+    pid: u32, // sender process id; 0 = the kernel itself
+    sequence: u64, // per-boot monotonic record number (gaps = lost records)
+    timestamp_ns: u64, // monotonic ns since boot (the `clock` timebase)
+    message_len: u16, // payload bytes (excludes the record's trailing pad)
+    flags: u8, // klog_flag_* bits
+    _reserved: [5]u8,
+};
+
+pub const klog_record_header_size: usize = 32; // @sizeOf(KlogRecordHeader), pinned by a test
+pub const klog_record_alignment: usize = 8;
+/// Per-record payload cap (one line; longer emitter lines are truncated).
+pub const klog_maximum_message: usize = 256;
+
+/// The klog_status copy-out: the ring's live cursors plus the wall-clock time
+/// of boot — the anchor a log persister names its per-boot directory with and
+/// combines with record timestamps for wall-clock line stamps.
+pub const KlogStatus = extern struct {
+    tail: u64, // oldest retained stream offset — always a record boundary
+    head: u64, // next byte to be written (end of stream)
+    next_sequence: u64, // the sequence the next record will get
+    boot_unix_seconds: u64, // wall-clock time of boot (RTC anchor)
+};
+
 /// Well-known IPC service ids for the bootstrap name registry (create_ipc_endpoint +
 /// ipc_register/ipc_lookup). Small integers, so no string interning is needed
 /// during bring-up. The VFS server registers under `vfs`; clients look it up.

system/kernel/kernel.zig

@@ -75,7 +75,7 @@ fn kmain(boot_information: *const BootInformation) noreturn {
     // Retain the whole stream in a RAM buffer too, so a user program can later
     // read it back (klog_read) and persist the boot log to disk — the only way to
     // see it on a headless/real machine with no host capturing serial.
-    log.addSink(log.ramSink);
+    // (Retention is the tagged ring inside log.zig — not a sink.)
 
     // The **framebuffer** is deliberately *not* a log sink. It's a separate output
     // surface — a bootstrap text console today, a graphics device driver later — so
@@ -426,7 +426,7 @@ fn status(message: []const u8) void {
 /// any display service holding the framebuffer. The console is otherwise silent in normal
 /// operation (see `status`); it exists now only for early-boot and fatal output.
 fn fatal(message: []const u8) void {
-    log.write(message);
+    log.appendPanic(message); // bounded lock wait: a panic never deadlocks on the log
     console.setSuppressed(false);
     console.write(message);
 }

system/kernel/log-ring.zig

@@ -0,0 +1,234 @@
+//! The tagged kernel log ring — a circular byte buffer of framed records, each
+//! stamped by the writer (the kernel) with the sender's pid, task name, level,
+//! per-boot sequence number, and monotonic timestamp. Pure code over an
+//! embedded buffer — no architecture or lock imports — so it host-tests
+//! alongside the other pure kernel pieces (`zig build test`).
+//!
+//! `head` and `tail` are free-running u64 positions in a logical byte stream;
+//! the physical wrap is invisible to readers (all copies are modulo the
+//! buffer), so a record never splits logically and no padding records exist.
+//! Reclaim happens record by record: the writer parses the header at `tail`
+//! (which it wrote itself) and advances until the new record fits — `tail`
+//! always sits on a record boundary, and sequence-number gaps tell a reader
+//! exactly how many records it lost.
+//!
+//! Locking is the caller's job (log.zig holds its log lock around every call);
+//! the ring itself is single-writer, snapshot-reader.
+
+const std = @import("std");
+const abi = @import("abi");
+
+pub fn Ring(comptime capacity: usize) type {
+    comptime std.debug.assert(std.math.isPowerOfTwo(capacity));
+    return struct {
+        const Self = @This();
+
+        buffer: [capacity]u8 = undefined,
+        head: u64 = 0,
+        tail: u64 = 0,
+        next_sequence: u64 = 0,
+
+        /// Append one record; returns its sequence number. `name` and `message`
+        /// are clamped to their ABI caps (the syscall clamps earlier too — the
+        /// clamp here makes the ring safe in isolation).
+        pub fn append(
+            self: *Self,
+            pid: u32,
+            name: []const u8,
+            level: abi.KlogLevel,
+            timestamp_ns: u64,
+            message: []const u8,
+            truncated: bool,
+        ) u64 {
+            const name_len: usize = @min(name.len, abi.maximum_process_name);
+            const message_len: usize = @min(message.len, abi.klog_maximum_message);
+            const record_len = recordLength(name_len, message_len);
+
+            // Reclaim whole records until the new one fits.
+            while (self.head + record_len - self.tail > capacity) self.reclaimOne();
+
+            const sequence = self.next_sequence;
+            self.next_sequence += 1;
+
+            const header = abi.KlogRecordHeader{
+                .magic = abi.klog_record_magic,
+                .level = level,
+                .name_len = @intCast(name_len),
+                .pid = pid,
+                .sequence = sequence,
+                .timestamp_ns = timestamp_ns,
+                .message_len = @intCast(message_len),
+                .flags = if (truncated) abi.klog_flag_truncated else 0,
+                ._reserved = @splat(0),
+            };
+            self.put(self.head, std.mem.asBytes(&header));
+            self.put(self.head + abi.klog_record_header_size, name[0..name_len]);
+            self.put(self.head + abi.klog_record_header_size + name_len, message[0..message_len]);
+            // The alignment pad is dead space; zero it so raw dumps stay tidy.
+            var pad = abi.klog_record_header_size + name_len + message_len;
+            while (pad < record_len) : (pad += 1)
+                self.buffer[@intCast((self.head + pad) % capacity)] = 0;
+            self.head += record_len;
+            return sequence;
+        }
+
+        /// Copy stream bytes beginning at `offset` into `out`. Returns null if
+        /// `offset` fell behind `tail` (overwritten) or lies past `head` — the
+        /// reader re-syncs from status(). 0 bytes means caught up.
+        pub fn read(self: *const Self, offset: u64, out: []u8) ?usize {
+            if (offset < self.tail or offset > self.head) return null;
+            const n: usize = @intCast(@min(out.len, self.head - offset));
+            self.get(offset, out[0..n]);
+            return n;
+        }
+
+        /// Cursors for klog_status. boot_unix_seconds is the kernel wrapper's
+        /// to fill — the ring knows nothing of wall clocks.
+        pub fn status(self: *const Self) abi.KlogStatus {
+            return .{
+                .tail = self.tail,
+                .head = self.head,
+                .next_sequence = self.next_sequence,
+                .boot_unix_seconds = 0,
+            };
+        }
+
+        fn reclaimOne(self: *Self) void {
+            var header_bytes: [abi.klog_record_header_size]u8 = undefined;
+            self.get(self.tail, &header_bytes);
+            const header = std.mem.bytesToValue(abi.KlogRecordHeader, &header_bytes);
+            // The writer wrote this header itself: the assert guards against
+            // memory corruption, not bad input.
+            std.debug.assert(header.magic == abi.klog_record_magic);
+            self.tail += recordLength(header.name_len, header.message_len);
+        }
+
+        fn recordLength(name_len: usize, message_len: usize) usize {
+            return std.mem.alignForward(usize, abi.klog_record_header_size + name_len + message_len, abi.klog_record_alignment);
+        }
+
+        // Byte-at-a-time modulo copies keep the wrap logic obviously correct;
+        // if they ever show in a profile, split into two @memcpy spans.
+        fn put(self: *Self, offset: u64, bytes: []const u8) void {
+            for (bytes, 0..) |b, i| self.buffer[@intCast((offset + i) % capacity)] = b;
+        }
+
+        fn get(self: *const Self, offset: u64, out: []u8) void {
+            for (out, 0..) |*b, i| b.* = self.buffer[@intCast((offset + i) % capacity)];
+        }
+    };
+}
+
+// --- tests (host) -----------------------------------------------------------
+
+const TestRing = Ring(4096);
+
+/// Parse the record at `offset` out of `ring`, returning the header plus name
+/// and message copies — the same walk a userspace drainer performs.
+const Parsed = struct {
+    header: abi.KlogRecordHeader,
+    name: [abi.maximum_process_name]u8 = undefined,
+    message: [abi.klog_maximum_message]u8 = undefined,
+
+    fn nameSlice(self: *const Parsed) []const u8 {
+        return self.name[0..self.header.name_len];
+    }
+    fn messageSlice(self: *const Parsed) []const u8 {
+        return self.message[0..self.header.message_len];
+    }
+    fn next(self: *const Parsed, offset: u64) u64 {
+        return offset + std.mem.alignForward(usize, abi.klog_record_header_size + self.header.name_len + self.header.message_len, abi.klog_record_alignment);
+    }
+};
+
+fn parseAt(ring: *const TestRing, offset: u64) Parsed {
+    var p: Parsed = undefined;
+    var header_bytes: [abi.klog_record_header_size]u8 = undefined;
+    std.debug.assert(ring.read(offset, &header_bytes).? == header_bytes.len);
+    p.header = std.mem.bytesToValue(abi.KlogRecordHeader, &header_bytes);
+    std.debug.assert(p.header.magic == abi.klog_record_magic);
+    _ = ring.read(offset + abi.klog_record_header_size, p.name[0..p.header.name_len]);
+    _ = ring.read(offset + abi.klog_record_header_size + p.header.name_len, p.message[0..p.header.message_len]);
+    return p;
+}
+
+test "header size is pinned" {
+    try std.testing.expectEqual(abi.klog_record_header_size, @sizeOf(abi.KlogRecordHeader));
+}
+
+test "append/read round trip" {
+    var ring = std.testing.allocator.create(TestRing) catch unreachable;
+    defer std.testing.allocator.destroy(ring);
+    ring.* = .{};
+
+    _ = ring.append(7, "/system/services/fat", .info, 123, "mounted /mnt/usb", false);
+    _ = ring.append(0, "kernel", .raw, 456, "wall clock online", false);
+
+    const first = parseAt(ring, ring.tail);
+    try std.testing.expectEqual(@as(u32, 7), first.header.pid);
+    try std.testing.expectEqual(abi.KlogLevel.info, first.header.level);
+    try std.testing.expectEqual(@as(u64, 123), first.header.timestamp_ns);
+    try std.testing.expectEqualStrings("/system/services/fat", first.nameSlice());
+    try std.testing.expectEqualStrings("mounted /mnt/usb", first.messageSlice());
+
+    const second = parseAt(ring, first.next(ring.tail));
+    try std.testing.expectEqual(@as(u32, 0), second.header.pid);
+    try std.testing.expectEqualStrings("kernel", second.nameSlice());
+    try std.testing.expectEqual(@as(u64, 1), second.header.sequence);
+}
+
+test "wrap reclaims whole records and keeps tail on a boundary" {
+    var ring = std.testing.allocator.create(TestRing) catch unreachable;
+    defer std.testing.allocator.destroy(ring);
+    ring.* = .{};
+
+    // Fill far past capacity so the ring wraps many times.
+    var i: u32 = 0;
+    while (i < 200) : (i += 1) {
+        var message: [64]u8 = undefined;
+        const m = std.fmt.bufPrint(&message, "line {d} padding padding padding", .{i}) catch unreachable;
+        _ = ring.append(1, "/system/tests/writer", .info, i, m, false);
+    }
+    try std.testing.expect(ring.head - ring.tail <= 4096);
+
+    // The record at tail parses cleanly (boundary held), and walking to head
+    // yields consecutive sequence numbers.
+    var offset = ring.tail;
+    var previous: ?u64 = null;
+    while (offset < ring.head) {
+        const p = parseAt(ring, offset);
+        if (previous) |q| try std.testing.expectEqual(q + 1, p.header.sequence);
+        previous = p.header.sequence;
+        offset = p.next(offset);
+    }
+    try std.testing.expectEqual(ring.head, offset);
+    // Records were lost (sequence at tail > 0), and the count is the gap.
+    try std.testing.expect(parseAt(ring, ring.tail).header.sequence > 0);
+}
+
+test "stale offset returns null; head offset reads zero bytes" {
+    var ring = std.testing.allocator.create(TestRing) catch unreachable;
+    defer std.testing.allocator.destroy(ring);
+    ring.* = .{};
+
+    var i: u32 = 0;
+    while (i < 300) : (i += 1)
+        _ = ring.append(1, "w", .info, i, "0123456789abcdef0123456789abcdef", false);
+
+    var out: [16]u8 = undefined;
+    try std.testing.expect(ring.read(0, &out) == null); // long overwritten
+    try std.testing.expect(ring.read(ring.head + 1, &out) == null); // past the end
+    try std.testing.expectEqual(@as(usize, 0), ring.read(ring.head, &out).?); // caught up
+}
+
+test "truncation flag and clamping" {
+    var ring = std.testing.allocator.create(TestRing) catch unreachable;
+    defer std.testing.allocator.destroy(ring);
+    ring.* = .{};
+
+    const long = "x" ** 300; // past klog_maximum_message
+    _ = ring.append(2, "w", .warn, 0, long, true);
+    const p = parseAt(ring, ring.tail);
+    try std.testing.expectEqual(@as(u16, abi.klog_maximum_message), p.header.message_len);
+    try std.testing.expect(p.header.flags & abi.klog_flag_truncated != 0);
+}

system/kernel/log.zig

@@ -4,22 +4,37 @@
 //! Output is a *diagnostic convenience, never a correctness dependency* — the
 //! kernel must boot and run correctly with zero output channels. So logging fans
 //! out to a set of registered **sinks**, each best-effort and self-guarding: the
-//! serial UART, the 0xE9 debug console, and — later — a file on a ramdisk/USB/SSD.
-//! A message reaches whatever channels exist; if none do, the kernel runs on,
-//! silent but correct.
+//! serial UART and the 0xE9 debug console. A message reaches whatever channels
+//! exist; if none do, the kernel runs on, silent but correct.
+//!
+//! Retention is the tagged RING (log-ring.zig): every emission becomes one
+//! record per line, stamped with the sender's pid, task name (its binary path),
+//! level, sequence number, and monotonic timestamp — attribution is structural,
+//! stamped by the kernel, not a naming convention a process could forge. The
+//! stamping is per LINE: an embedded '\n' ends the record, so a payload cannot
+//! imitate another sender on the line that follows. Oldest records are
+//! overwritten when the ring is full; sequence gaps make the loss countable.
+//! `klog_read`/`klog_status` expose the stream to userspace (the logger service
+//! drains it into per-process files once storage is up).
+//!
+//! Locking: a dedicated log spinlock, NOT the big kernel lock. `print` is
+//! called both inside and outside BKL sections (and from ISRs), so the log
+//! lock is taken with interrupts off and nothing inside it ever takes the BKL —
+//! lock order is strictly BKL -> log lock, never the reverse. Panic paths use a
+//! bounded try-acquire and fall back to sinks-only: a panic must never deadlock
+//! on its own diagnostics.
 //!
 //! The **framebuffer is deliberately not a sink here.** It's a separate output
 //! surface (a bootstrap text console today, a graphics device driver later), so
-//! the log never assumes the machine is text-based. `main.zig` mirrors a few
-//! user-facing status lines and panics to it explicitly; the verbose log does not.
-//!
-//! No allocation: the sink table is fixed, so the log works before the heap is up
-//! and inside a panic. Two channels don't go through the sink list because they
-//! must survive even a total-output failure: `checkpoint` (a one-byte POST code)
-//! and `recordPanic` (a breadcrumb in a fixed record).
+//! the log never assumes the machine is text-based. Two channels bypass the
+//! sink list because they must survive even a total-output failure:
+//! `checkpoint` (a one-byte POST code) and `recordPanic` (a fixed breadcrumb).
 
 const std = @import("std");
 const architecture = @import("architecture");
+const abi = @import("abi");
+const log_ring = @import("log-ring.zig");
+const wall_clock = @import("wall-clock.zig");
 
 pub const SinkFn = *const fn ([]const u8) void;
 
@@ -36,42 +51,120 @@ pub fn addSink(sink: SinkFn) void {
     }
 }
 
-/// Fan `bytes` out to every registered sink.
-pub fn write(bytes: []const u8) void {
-    for (sinks[0..sink_count]) |sink| sink(bytes);
+// --- the log lock ------------------------------------------------------------
+
+var lock_held = std.atomic.Value(u32).init(0);
+
+fn lockAcquire() u64 {
+    const flags = architecture.saveInterrupts();
+    while (lock_held.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) std.atomic.spinLoopHint();
+    return flags;
 }
 
-// --- the RAM sink: a retained copy of the whole diagnostic stream ------------
-//
-// A fixed in-image buffer that accumulates every logged byte, so a user program
-// (`log-flush`, and init at shutdown) can read it back through `klog_read` and
-// persist it to a file — the boot log survives on a headless/real machine that
-// has no host capturing serial. It is a *sink like any other*: register it with
-// `addSink(ramSink)` at boot. No allocation (works pre-heap and in a panic).
-//
-// It fills linearly and stops when full: the earliest output — the most valuable
-// for diagnosing a boot — is kept, and the tail is still on the live serial sink.
-// 256 KiB comfortably holds a full boot plus a long run (a boot is ~15 KiB).
+fn lockTryAcquire(spins: usize) ?u64 {
+    const flags = architecture.saveInterrupts();
+    var i: usize = 0;
+    while (i < spins) : (i += 1) {
+        if (lock_held.cmpxchgWeak(0, 1, .acquire, .monotonic) == null) return flags;
+        std.atomic.spinLoopHint();
+    }
+    architecture.restoreInterrupts(flags);
+    return null;
+}
 
-const ram_capacity = 256 * 1024;
-var ram_buffer: [ram_capacity]u8 = undefined;
-var ram_len: usize = 0;
+fn lockRelease(flags: u64) void {
+    lock_held.store(0, .release);
+    architecture.restoreInterrupts(flags);
+}
 
-/// The RAM sink. Best-effort and self-guarding like every sink: appends what fits
-/// and silently drops the rest once full. (Concurrency matches the other sinks —
-/// the dominant writer, debug_write, already holds the kernel lock; a rare torn
-/// append on a kernel-internal line is an accepted diagnostic imperfection.)
-pub fn ramSink(bytes: []const u8) void {
-    const n = @min(ram_buffer.len - ram_len, bytes.len);
-    if (n != 0) {
-        @memcpy(ram_buffer[ram_len..][0..n], bytes[0..n]);
-        ram_len += n;
+// --- the ring + renderer -----------------------------------------------------
+
+/// 512 KiB: the tagged frames cost ~30% over the raw text, and the ring only
+/// needs to cover the pre-mount backlog (a boot is ~15 KiB of text) — the
+/// logger service tails it continuously once storage is up.
+const ring_capacity = 512 * 1024;
+var ring: log_ring.Ring(ring_capacity) = .{};
+
+/// Renderer state: whether the sinks sit at a line start, and which pid's line
+/// is currently open — when a different sender interleaves mid-line, the
+/// renderer closes the line so serial output can't visually merge two senders.
+var at_line_start: bool = true;
+var open_line_pid: u32 = 0;
+
+/// Append `bytes` as one tagged record per line and render them to the sinks.
+/// The core emission path: `debug_write` calls this with the sender's identity;
+/// kernel-internal `write`/`print` funnel here as pid 0 ("kernel", raw).
+pub fn append(pid: u32, name: []const u8, level: abi.KlogLevel, bytes: []const u8) void {
+    if (bytes.len == 0) return;
+    const now = architecture.nanos();
+    const flags = lockAcquire();
+    defer lockRelease(flags);
+    appendLocked(pid, name, level, now, bytes);
+}
+
+/// The panic-safe variant: bounded lock wait; on failure, sinks only — the ring
+/// entry is lost but the message still reaches serial, and the panic cannot
+/// deadlock on a core that died holding the log lock.
+pub fn appendPanic(bytes: []const u8) void {
+    if (lockTryAcquire(100_000)) |flags| {
+        defer lockRelease(flags);
+        appendLocked(0, "kernel", .raw, architecture.nanos(), bytes);
+    } else {
+        for (sinks[0..sink_count]) |sink| sink(bytes);
     }
 }
 
-/// The accumulated log so far — what `klog_read` copies out.
-pub fn ramSnapshot() []const u8 {
-    return ram_buffer[0..ram_len];
+fn appendLocked(pid: u32, name: []const u8, level: abi.KlogLevel, now: u64, bytes: []const u8) void {
+    var rest = bytes;
+    while (rest.len != 0) {
+        const newline = std.mem.indexOfScalar(u8, rest, '\n');
+        // The record payload excludes the newline: a record IS a line (or the
+        // open tail of one when the emission didn't end in '\n').
+        const line = if (newline) |i| rest[0..i] else rest;
+        const line_complete = newline != null;
+        if (line.len != 0 or line_complete)
+            _ = ring.append(pid, name, level, now, line, line.len > abi.klog_maximum_message);
+        render(pid, name, level, line, line_complete);
+        rest = if (newline) |i| rest[i + 1 ..] else rest[rest.len..];
+    }
+}
+
+/// Serial/debugcon rendering. Kernel output and legacy raw user output pass
+/// through byte-identical to the historical stream (services still write their
+/// own "name: " prefixes until the std.log migration). Leveled (std.log)
+/// records get a kernel-rendered "<name>: " prefix at line start — err/warn/
+/// debug also get their level spelled out.
+fn render(pid: u32, name: []const u8, level: abi.KlogLevel, line: []const u8, line_complete: bool) void {
+    if (sink_count == 0) return;
+    if (line.len == 0 and !line_complete) return;
+    if (!at_line_start and open_line_pid != pid) {
+        fanOut("\n");
+        at_line_start = true;
+    }
+    if (at_line_start and level != .raw) {
+        fanOut(name);
+        fanOut(": ");
+        switch (level) {
+            .err => fanOut("error: "),
+            .warn => fanOut("warning: "),
+            .debug => fanOut("debug: "),
+            .info, .raw => {},
+        }
+    }
+    fanOut(line);
+    if (line_complete) fanOut("\n");
+    at_line_start = line_complete;
+    open_line_pid = pid;
+}
+
+fn fanOut(bytes: []const u8) void {
+    for (sinks[0..sink_count]) |sink| sink(bytes);
+}
+
+/// Kernel-internal write — a raw record from "kernel" (pid 0). The signature is
+/// unchanged so every existing kernel call site stays as it is.
+pub fn write(bytes: []const u8) void {
+    append(0, "kernel", .raw, bytes);
 }
 
 /// A formatted log line. Truncates past 256 bytes; the buffer is on the stack, so
@@ -81,6 +174,24 @@ pub fn print(comptime fmt: []const u8, args: anytype) void {
     write(std.fmt.bufPrint(&buffer, fmt, args) catch return);
 }
 
+/// klog_read: copy ring stream bytes from `offset` into `out`. Null when the
+/// cursor was overwritten or lies past the end — the reader re-syncs via
+/// status(). Zero bytes means caught up.
+pub fn readAt(offset: u64, out: []u8) ?usize {
+    const flags = lockAcquire();
+    defer lockRelease(flags);
+    return ring.read(offset, out);
+}
+
+/// klog_status: the ring cursors plus the boot wall-clock anchor.
+pub fn status() abi.KlogStatus {
+    const flags = lockAcquire();
+    defer lockRelease(flags);
+    var s = ring.status();
+    s.boot_unix_seconds = wall_clock.bootSeconds();
+    return s;
+}
+
 /// Emit a one-byte checkpoint/POST code (I/O port 0x80) — the always-available
 /// progress channel for when there is no text output at all. Independent of the
 /// sink list, so it works even before any sink is registered.

system/kernel/process.zig

@@ -234,6 +234,7 @@ fn system_call(state: *architecture.CpuState) void {
         .process_signal => systemProcessSignal(state),
         .timer_bind => systemTimerBind(state),
         .klog_read => systemKlogRead(state),
+        .klog_status => systemKlogStatus(state),
         .wall_clock => systemWallClock(state),
         .shm_create => systemShmCreate(state),
         .shm_map => systemShmMap(state),
@@ -1210,7 +1211,6 @@ fn systemIrqAck(state: *architecture.CpuState) void {
 /// Whether the debug_write stream sits at the start of a line — the last emitted
 /// byte was a newline (true at boot: nothing emitted yet). Guarded by the kernel
 /// lock in `systemDebugWrite`, like the stream it describes.
-var write_at_line_start: bool = true;
 
 /// debug_write(ptr, len): copy bytes from user memory into the kernel log.
 /// A bring-up diagnostic — real output goes through the VFS/console later.
@@ -1231,31 +1231,42 @@ var write_at_line_start: bool = true;
 fn systemDebugWrite(state: *architecture.CpuState) void {
     const ptr = architecture.systemCallArg(state, 0);
     const len = architecture.systemCallArg(state, 1);
+    const level_raw = architecture.systemCallArg(state, 2);
     if (len <= write_buffer.len and ptr < user_half_end and ptr + len <= user_half_end) {
         const source: [*]const u8 = @ptrFromInt(ptr);
+        // Levels above the enum range clamp to raw — old two-arg callers land
+        // there naturally (garbage in arg 2 stays harmless).
+        const level: abi.KlogLevel = if (level_raw <= @intFromEnum(abi.KlogLevel.raw))
+            @enumFromInt(level_raw)
+        else
+            .raw;
+        const t = scheduler.current();
         const flags = sync.enter();
         defer sync.leave(flags);
         @memcpy(write_buffer[0..len], source[0..len]); // keep the latest message
         write_len = len;
         write_from_user = architecture.fromUser(state);
         write_count += 1;
-        log.write(source[0..len]);
-        if (len != 0) write_at_line_start = source[len - 1] == '\n';
+        // The kernel stamps the sender's identity — attribution is structural,
+        // not a prefix convention the payload could forge (and it is stamped
+        // per line inside log.append).
+        log.append(t.id, t.name(), level, source[0..len]);
         architecture.setSystemCallResult(state, len);
     } else {
         fail(state);
     }
 }
 
-/// klog_read(offset, ptr, len) -> bytes copied: copy the kernel's in-memory
-/// diagnostic log (the RAM sink in log.zig) out to the user buffer at `ptr`,
-/// starting at `offset`. Returns the count copied — 0 once `offset` reaches the
-/// end — so a program reads the whole log by looping from 0 until it gets 0.
+/// klog_read(offset, ptr, len) -> bytes copied: copy tagged log-ring stream
+/// bytes beginning at stream offset `offset` out to the user buffer at `ptr`.
+/// Returns the count copied — 0 means caught up — and fails once `offset` has
+/// fallen behind the ring's tail (the records were overwritten) or lies past
+/// its head; the reader re-syncs via klog_status. A reader parses
+/// [KlogRecordHeader][name][message] frames out of the byte stream (abi.zig).
 ///
 /// The mirror of `debug_write`: the same overflow-safe user-half bounds check,
-/// but the copy runs kernel -> user. Written under the kernel lock so the source
-/// snapshot can't grow underneath the copy. A read-only diagnostic — it exposes
-/// only the log the kernel already broadcasts to serial, nothing else.
+/// but the copy runs kernel -> user, under the log lock (inside log.readAt) so
+/// the stream can't move underneath the copy. A read-only diagnostic.
 fn systemKlogRead(state: *architecture.CpuState) void {
     const offset = architecture.systemCallArg(state, 0);
     const ptr = architecture.systemCallArg(state, 1);
@@ -1263,21 +1274,31 @@ fn systemKlogRead(state: *architecture.CpuState) void {
     // Confine the whole destination span to the user (low) half. `len <=
     // user_half_end - ptr` bounds the length without an overflowing add.
     if (ptr < user_half_end and len <= user_half_end - ptr) {
-        const flags = sync.enter();
-        defer sync.leave(flags);
-        const snapshot = log.ramSnapshot();
-        var n: usize = 0;
-        if (offset < snapshot.len) {
-            n = @min(len, snapshot.len - offset);
         const dest: [*]u8 = @ptrFromInt(ptr);
-            @memcpy(dest[0..n], snapshot[offset..][0..n]);
-        }
+        const n = log.readAt(offset, dest[0..len]) orelse return fail(state);
         architecture.setSystemCallResult(state, n);
     } else {
         fail(state);
     }
 }
 
+/// klog_status(ptr) -> 0: copy a KlogStatus — the ring's live cursors plus the
+/// boot wall-clock anchor — out to the user buffer at `ptr`. How a log reader
+/// finds the oldest retained offset, detects lost records (sequence gaps), and
+/// names a per-boot log directory (boot_unix_seconds).
+fn systemKlogStatus(state: *architecture.CpuState) void {
+    const ptr = architecture.systemCallArg(state, 0);
+    const size = @sizeOf(abi.KlogStatus);
+    if (ptr < user_half_end and size <= user_half_end - ptr) {
+        var status = log.status();
+        const dest: [*]u8 = @ptrFromInt(ptr);
+        @memcpy(dest[0..size], std.mem.asBytes(&status)[0..size]);
+        architecture.setSystemCallResult(state, 0);
+    } else {
+        fail(state);
+    }
+}
+
 /// mmap(len, prot) -> base: grant `len` bytes (rounded up to whole pages) of
 /// fresh, zeroed, writable+NX memory in the caller's mmap arena, and return the
 /// base virtual address. `prot` is accepted but not yet honoured (grants are

system/kernel/wall-clock.zig

@@ -24,3 +24,9 @@ pub fn init() void {
 pub fn nowSeconds() u64 {
     return boot_unix_seconds + (architecture.nanos() -% boot_nanos) / 1_000_000_000;
 }
+
+/// The wall-clock time of boot itself (the RTC anchor) — what klog_status hands
+/// the logger service to name a per-boot log directory. Zero until `init` runs.
+pub fn bootSeconds() u64 {
+    return boot_unix_seconds;
+}

system/services/init/init.zig

@@ -192,15 +192,19 @@ fn flushKernelLog() void {
     var file = runtime.fs.open(log_path, .{ .create = true, .truncate = true }) orelse return; // no USB volume
     defer file.close();
     var chunk: [4096]u8 = undefined;
-    var offset: usize = 0;
+    // Interim (until the logger service in M-E): the stream is now framed
+    // records, so the file holds raw frames rather than plain text. Start at
+    // the ring's tail — offset 0 is gone once the ring has wrapped.
+    var offset: u64 = if (runtime.system.klogStatus()) |s| s.tail else 0;
+    const start = offset;
     while (true) {
-        const got = runtime.system.klogRead(offset, &chunk);
-        if (got == 0) break; // reached the end of the accumulated log
+        const got = runtime.system.klogRead(offset, &chunk) orelse break; // cursor lost
+        if (got == 0) break; // caught up
         if (file.writeAll(chunk[0..got]) == null) break; // storage went away
         offset += got;
     }
     var line: [96]u8 = undefined;
-    _ = runtime.system.write(std.fmt.bufPrint(&line, "/system/services/init: flushed log to {s} ({d} bytes)\n", .{ log_path, offset }) catch "");
+    _ = runtime.system.write(std.fmt.bufPrint(&line, "/system/services/init: flushed log to {s} ({d} bytes)\n", .{ log_path, offset - start }) catch "");
 }
 
 /// The stop sequence: persist the log while storage is still up, then terminate

system/services/log-flush/log-flush.zig

@@ -24,14 +24,18 @@ const log_path = "/mnt/usb/DANOS.LOG";
 /// the log is exhausted. Returns the number of bytes written.
 fn drainKernelLog(file: *fs.File) usize {
     var chunk: [4096]u8 = undefined;
-    var offset: usize = 0;
+    // Interim (until the logger service in M-E): the stream is now framed
+    // records, so the file holds raw frames rather than plain text. Start at
+    // the ring's tail — offset 0 is gone once the ring has wrapped.
+    var offset: u64 = if (runtime.system.klogStatus()) |s| s.tail else 0;
+    const start = offset;
     while (true) {
-        const got = runtime.system.klogRead(offset, &chunk);
-        if (got == 0) break; // reached the end of the accumulated log
+        const got = runtime.system.klogRead(offset, &chunk) orelse break; // cursor lost
+        if (got == 0) break; // caught up
         if (file.writeAll(chunk[0..got]) == null) break; // storage went away
         offset += got;
     }
-    return offset;
+    return offset - start;
 }
 
 pub fn main() void {