diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md
index a6076dd..1ed30e5 100644
--- a/docs/semantic-parity.md
+++ b/docs/semantic-parity.md
@@ -21,7 +21,7 @@ this document or `bootstrap-semantic-parity.md` is open.
 | Timestamp import | Canonical dashboard and API include timestamp import lifecycle/rewrite metrics and behavior | **Closed at the implementation boundary.** The format preserves distinct `witnessed_at` and sentinel-zero `indexed_at` columns. Subscriber v1/v2 encoding applies upstream's `indexed_at != 0 ? indexed_at : witnessed_at` display rule while ranges and timestamp cursors remain anchored to immutable witness envelopes. The strict seekable RFC 4180 parser matches upstream's header, row rejection, byte-offset, CID, RFC3339, 64 KiB, and bounded-sampling contracts in a forward-only allocation-free pass. A separate Bloom-filtered RocksDB rule store bulk-loads sorted external SST chunks with CSV-order last-write-wins, reconstructs its resident collection gate on open, and implements specific-CID precedence with all-version fallback. Its central archive-lock hook stamps live, repair, bootstrap, and merge materializations before sequence assignment, and feeds the mutated rows to immediate delivery so hot and cold encodings agree. The sealed-JSS patch primitive preserves block topology/envelopes, clean frames, and the opaque bloom/collection footer while atomically changing only `indexed_at`; it is exercised against an upstream-produced segment. A concrete sealed-segment bloom catalog, bounded DID/FD bucketer, fsynced packed offsets, positioned revalidation, and collision-safe per-segment patch plan implement Phase B/C with specific-CID precedence and idempotent disk-to-disk coverage. A dedicated archive rewrite mutex serializes the real delete compactor with timestamp patching while leaving live append available. The unified metadata DB synchronously persists the current-job pointer, complete job record, and per-segment done set. The real runner activates rules before force-seal, durably crosses the bucketed handoff, checkpoints only after patch fsync/rename/dir-sync, retains partial cancellation progress, and resumes after full archive/rule/metadata reopen. The background manager canonicalizes symlinks, confines regular files, enforces one durable nonterminal job, and adopts it on startup. Bearer-gated import/status XRPC matches the pinned lexicons, fixed-width digest comparison, error names/statuses, steady-state admission, and current/by-id status behavior; a real loopback HTTP run exercised it against the local simulator. Every canonical import metric is produced at its real parse, route, patch, or durable-terminal boundary; cancellation remains a resumable pause and is not counted as terminal failure. |
 | Status/diagnostics | Repo attempts/error class/final PDS and per-host aggregates survive restart and are operator-queryable | **closed at the implementation boundary.** Durable v5 repo rows distinguish initial Backfill.Rev from latest Rev/UpdatedAt, retain resolved Handle/PDS and reserved record/byte fields, and atomically maintain `handle/<normalized>`. Merge source cursor commits refresh latest revisions without mutating backfill watermarks. Same-batch `host/<bucket>` totals, active/status counts, cumulative error classes, and five bounded recent samples survive restart; host moves and active flips are atomic with the source repo row. `/status?tab=hosts` reads those rows without a whole-network scan and matches upstream ordering and error presentation. Account queries preserve resolver-first handle semantics, the legacy `did`/`handle` aliases, and missing-identity hydration, then reconstruct records last-writer-wins from checksum-verified, DID-bloom-pruned sealed JSS plus rotation-safe active/pending and bootstrap-live snapshots. The resulting canonical MST is compared with the PDS commit root through real Sync 1.1 `getLatestCommit`/`getBlocks`; the page renders the same match fields and error state as upstream. `HEAD` performs lookup without verification; successful status responses are `no-store` and carry a generation timestamp. Expensive verification uses upstream's exact 4/source-IP/minute fixed-window limiter, 4,096-entry bound, stale pruning, oldest eviction, and explicit disable option. The offline receipt uses a real HTTP Stream route, real RocksDB row, real JSS archive, signed commit, and loopback PLC/PDS; four separate connections verify successfully, a fifth changing only its ephemeral port is blocked without another upstream request, HEAD remains side-effect-free, and disabling the limiter performs a new authoritative verification. |
 | Prometheus/Grafana | Every canonical dashboard query maps to a real, same-semantics producer; runtime-specific panels use honest Zig/process metrics | **closed.** The exact checksum-pinned upstream dashboard is vendored with an offline fail-closed scrape contract. Canonical build, live-firehose, archive, verifier, subscribe, backfill, timestamp-import, and compaction families all have same-semantics producers. Compaction measures pass success/failure/duration, early triggers, live tombstone size, folded reasons, worker examination/rewrite/clean outcomes, dropped-row reasons, bytes, manifest heals, durable watermark, and witnessed-time lag at the corresponding real boundaries. Real JSS tests cover bloom avoidance, bounded concurrent rewrites, malformed account failure, reason accounting, manifest-before-watermark ordering, and durable eviction. The ReleaseSafe production receipt reports `required=104 present=104 missing=0`; the runtime-only dashboard row remains an intentional checksum-pinned Zig/process adaptation with no synthetic `go_*` aliases. See `grafana-dashboard.md`. |
-| Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **partially closed.** `just differential-oracle` reuses the exact pinned upstream simulator, physical JSS reader, event-log normalizer/comparator, independent MST ground truth, invariant checker, fault scheduler, and public Go client through a fail-closed Go overlay. A real ReleaseSafe Stream child now proves bootstrap convergence; exact normalized durable equivalence for controlled create/update/delete/identity/account/sync/create_resync traffic; duplicate and regressed relay windows with account lifecycle events; the complete upstream frame-fault campaign (malformed/error/oversized/unknown/swallowed/partial-CAR); exact reconnect, sequence-gap, decode, unknown, missing-block, and repair metrics; all-kind anti-vacuity; mutation detection; clean SIGTERM drain; same-disk restart; and public plan/getSegment/getBlock/subscribe-v2 reconstruction. The complete upstream adversarial ingest matrix is also closed: live sibling-preserving invalid collection/rkey/field-width drops, divergent invalid-rev sync repair-then-whole-event-drop, signed non-TID verifier rejection plus follow-up repair, DID-scoped PLC failure without sibling wedge, and backfill CAR invalid UTF-8/path/width drops all run through real sockets, RocksDB, signed repos, and physical JSS with source/reason counter floors and filtered MST convergence. The upstream merge-cursor store-fault contract now also runs through a real process and RocksDB: the selected durable batch fails loudly, proves it fired, and a fault-free reopen converges from the unadvanced cursor. Still open: power-loss and segment-file fault schedules, compaction mutation coverage, and multi-seed determinism. |
+| Oracle | Event-log equivalence, final-state convergence, crash/power-loss, verifier repair, hostile input, and anti-vacuity receipts | **partially closed.** `just differential-oracle` reuses the exact pinned upstream simulator, physical JSS reader, event-log normalizer/comparator, independent MST ground truth, invariant checker, fault scheduler, and public Go client through a fail-closed Go overlay. A real ReleaseSafe Stream child now proves bootstrap convergence; exact normalized durable equivalence for controlled create/update/delete/identity/account/sync/create_resync traffic; duplicate and regressed relay windows with account lifecycle events; the complete upstream frame-fault campaign (malformed/error/oversized/unknown/swallowed/partial-CAR); exact reconnect, sequence-gap, decode, unknown, missing-block, and repair metrics; all-kind anti-vacuity; mutation detection; clean SIGTERM drain; same-disk restart; and public plan/getSegment/getBlock/subscribe-v2 reconstruction. The complete upstream adversarial ingest matrix is also closed: live sibling-preserving invalid collection/rkey/field-width drops, divergent invalid-rev sync repair-then-whole-event-drop, signed non-TID verifier rejection plus follow-up repair, DID-scoped PLC failure without sibling wedge, and backfill CAR invalid UTF-8/path/width drops all run through real sockets, RocksDB, signed repos, and physical JSS with source/reason counter floors and filtered MST convergence. The upstream merge-cursor store-fault contract runs through a real process and RocksDB: the selected durable batch fails loudly, proves it fired, and a fault-free reopen converges from the unadvanced cursor. Segment-file fault parity is also closed: every production JSS write/fsync/commit-rename crosses the same operation taxonomy, the real process proves ENOSPC/EIO/short-write failure and recovery, Patch and Rewrite exhaustively sweep every local ordinal against real files, and a compaction rename failure preserves both the source bytes and unadvanced watermark before a successful retry. The audit also added the missing active-segment parent-directory fsync and repaired temp-file cleanup. Still open: strict power-loss coverage, broader compaction mutation coverage, and multi-seed determinism. |
 
 ## Experiment admission
 
diff --git a/docs/upstream-harness.md b/docs/upstream-harness.md
index 1123d2b..16211a1 100644
--- a/docs/upstream-harness.md
+++ b/docs/upstream-harness.md
@@ -19,7 +19,7 @@ zig usage: point stream at `ws://localhost:7777/xrpc/com.atproto.sync.subscribeR
 
 ## oracle (what an equivalent zig harness would check)
 
-driver boots real server against simulator, walks lifecycle gated on durable-append acks (never sleeps). the simulator world doubles as independent ground truth. contracts: seq ordering, final-state vs world, event-log equivalence (catches lost intermediate events), compaction watermark rules, replay==archive, fold-convergence, anti-vacuity on injected faults (every scheduled fault must prove it fired). tiers incl. crash/restart (SIGKILL real children at crashpoints), deterministic store faults (a selected real RocksDB mutation fails, then the same disk state is reopened), and power-loss (strict-mem vfs dropping unsynced writes). detection power measured by mutation campaign.
+driver boots real server against simulator, walks lifecycle gated on durable-append acks (never sleeps). the simulator world doubles as independent ground truth. contracts: seq ordering, final-state vs world, event-log equivalence (catches lost intermediate events), compaction watermark rules, replay==archive, fold-convergence, anti-vacuity on injected faults (every scheduled fault must prove it fired). tiers incl. crash/restart (SIGKILL real children at crashpoints), deterministic store faults (a selected real RocksDB mutation fails, then the same disk state is reopened), segment faults (real JSS write/fsync/rename calls fail at a selected process-wide ordinal, with exhaustive Patch/Rewrite sweeps and fault-free reopen), and power-loss (strict-mem vfs dropping unsynced writes). detection power measured by mutation campaign.
 
 ## ingest / live consumer (upstream reference)
 
diff --git a/src/internal/archive.zig b/src/internal/archive.zig
index 0cac7be..534c9b6 100644
--- a/src/internal/archive.zig
+++ b/src/internal/archive.zig
@@ -18,6 +18,7 @@ const segment = @import("segment.zig");
 const writer_mod = @import("segment_writer.zig");
 const metrics = @import("metrics.zig");
 const manifest_mod = @import("manifest.zig");
+const segment_io = @import("segment_io.zig");
 
 const Io = std.Io;
 const Allocator = std.mem.Allocator;
@@ -175,8 +176,8 @@ pub const Archive = struct {
 
             var f = try self.dir.createFile(self.io, name, .{ .truncate = true });
             defer f.close(self.io);
-            try f.writeStreamingAll(self.io, w.bytes());
-            try f.sync(self.io);
+            try segment_io.writeStreamingAll(&f, self.io, w.bytes());
+            try segment_io.sync(&f, self.io);
             self.next_seq = @max(self.next_seq, max_seq + 1);
             log.info("recovery: sealed interrupted segment {s} ({d} bytes valid)", .{ name, valid_end });
             return;
@@ -185,14 +186,26 @@ pub const Archive = struct {
 
     fn openNextSegment(self: *Archive) !void {
         self.writer = try writer_mod.ActiveWriter.init(self.allocator);
+        errdefer self.writer.deinit();
         // the writer owns block granularity (it also sizes per-block blooms
         // from this at seal); we sync to disk whenever the buffer grows
         self.writer.max_events_per_block = self.max_events_per_block;
         var name_buf: [64]u8 = undefined;
         const name = formatSegmentName(&name_buf, self.seg_index);
         self.file = try self.dir.createFile(self.io, name, .{ .truncate = true });
-        try self.file.writeStreamingAll(self.io, self.writer.bytes());
-        try self.file.sync(self.io);
+        var file_open = true;
+        errdefer {
+            if (file_open) self.file.close(self.io);
+            self.dir.deleteFile(self.io, name) catch {};
+        }
+        try segment_io.writeStreamingAll(&self.file, self.io, self.writer.bytes());
+        try segment_io.sync(&self.file, self.io);
+        // The header is not durable if a power cut can still forget the new
+        // directory entry. Upstream syncs the parent after writer creation.
+        const dir_file: Io.File = .{ .handle = self.dir.handle, .flags = .{ .nonblocking = false } };
+        var sync_file = dir_file;
+        try segment_io.sync(&sync_file, self.io);
+        file_open = false;
         self.file_written = self.writer.bytes().len;
         self.synced_block_count = 0;
         if (self.stats) |stats| stats.ingest_active_segment_bytes.store(0, .monotonic);
@@ -312,8 +325,8 @@ pub const Archive = struct {
 
     fn syncToDisk(self: *Archive) !void {
         const all = self.writer.bytes();
-        try self.file.writePositionalAll(self.io, all[self.file_written..], self.file_written);
-        try self.file.sync(self.io);
+        try segment_io.writePositionalAll(&self.file, self.io, all[self.file_written..], self.file_written);
+        try segment_io.sync(&self.file, self.io);
         if (self.stats) |stats| {
             const flushed: u64 = self.writer.block_count - self.synced_block_count;
             if (flushed > 0) _ = stats.ingest_blocks_flushed_total.fetchAdd(flushed, .monotonic);
@@ -344,10 +357,10 @@ pub const Archive = struct {
         if (self.writer.block_count == 0) return; // nothing to seal
         try self.writer.seal();
         const all = self.writer.bytes();
-        try self.file.writePositionalAll(self.io, all[self.file_written..], self.file_written);
-        try self.file.sync(self.io);
-        try self.file.writePositionalAll(self.io, all[0..segment.header_size], 0);
-        try self.file.sync(self.io);
+        try segment_io.writePositionalAll(&self.file, self.io, all[self.file_written..], self.file_written);
+        try segment_io.sync(&self.file, self.io);
+        try segment_io.writePositionalAll(&self.file, self.io, all[0..segment.header_size], 0);
+        try segment_io.sync(&self.file, self.io);
         try self.manifest.refresh(self.dir, self.seg_index, false);
         self.file.close(self.io);
         self.writer.deinit();
@@ -380,10 +393,10 @@ pub const Archive = struct {
         } else {
             try self.writer.seal();
             const all = self.writer.bytes();
-            try self.file.writePositionalAll(self.io, all[self.file_written..], self.file_written);
-            try self.file.sync(self.io);
-            try self.file.writePositionalAll(self.io, all[0..segment.header_size], 0);
-            try self.file.sync(self.io);
+            try segment_io.writePositionalAll(&self.file, self.io, all[self.file_written..], self.file_written);
+            try segment_io.sync(&self.file, self.io);
+            try segment_io.writePositionalAll(&self.file, self.io, all[0..segment.header_size], 0);
+            try segment_io.sync(&self.file, self.io);
             try self.manifest.refresh(self.dir, self.seg_index, false);
             self.file.close(self.io);
             if (self.stats) |stats| {
@@ -438,6 +451,32 @@ test "segment name round-trip" {
     try testing.expectEqual(@as(?u64, null), parseSegmentIndex("seg_x.jss"));
 }
 
+test "active segment initialization faults remove the incomplete file" {
+    segment_io.disarm();
+    defer segment_io.disarm();
+    const cases = [_]struct { spec: []const u8, expected: anyerror }{
+        .{ .spec = "write:1:enospc", .expected = error.NoSpaceLeft },
+        .{ .spec = "sync:1:eio", .expected = error.InputOutput },
+        .{ .spec = "sync:2:eio", .expected = error.InputOutput },
+    };
+    for (cases) |case| {
+        var tmp = testing.tmpDir(.{});
+        defer tmp.cleanup();
+        var path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
+        const path_len = try tmp.dir.realPath(testing.io, &path_buf);
+        try testing.expect(segment_io.armSpec(case.spec));
+        try testing.expectError(case.expected, Archive.init(testing.allocator, testing.io, path_buf[0..path_len]));
+        segment_io.disarm();
+        var segments = try tmp.dir.openDir(testing.io, "segments", .{ .iterate = true });
+        defer segments.close(testing.io);
+        var it = segments.iterate();
+        try testing.expectEqual(@as(?Io.Dir.Entry, null), try it.next(testing.io));
+
+        var recovered = try Archive.init(testing.allocator, testing.io, path_buf[0..path_len]);
+        recovered.deinit();
+    }
+}
+
 test "archive: append, rotate, recover across restart" {
     var threaded: Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
diff --git a/src/internal/compact/pass.zig b/src/internal/compact/pass.zig
index a940852..535c74d 100644
--- a/src/internal/compact/pass.zig
+++ b/src/internal/compact/pass.zig
@@ -22,6 +22,7 @@ const meta_store = @import("../meta_store.zig");
 const metrics = @import("../metrics.zig");
 const rewrite_mod = @import("../segment_rewrite.zig");
 const segment = @import("../segment.zig");
+const segment_io = @import("../segment_io.zig");
 const tombstone = @import("../tombstone.zig");
 const watermark = @import("watermark.zig");
 
@@ -520,6 +521,52 @@ test "pass: tombstone cap splits the pass into chunks across segments" {
     try testing.expectEqual(@as(?u64, 4), wm.load(io));
 }
 
+test "pass: commit rename fault fails loud, preserves source, and retry converges" {
+    segment_io.disarm();
+    defer segment_io.disarm();
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+    const io = threaded.io();
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+    var path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
+    const n = try tmp.dir.realPath(io, &path_buf);
+    var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined;
+    const data_dir = try std.fmt.bufPrint(&data_dir_buf, "{s}/data", .{path_buf[0..n]});
+    {
+        var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir);
+        defer archive.deinit();
+        const post = "app.bsky.feed.post";
+        _ = try archive.append(row(.create, "did:plc:a", post, "k", "x"), 1);
+        _ = try archive.append(row(.delete, "did:plc:a", post, "k", ""), 2);
+        try archive.rotate();
+    }
+    var meta = try meta_store.Store.open(testing.allocator, data_dir);
+    defer meta.deinit();
+    var wm = watermark.Store.init(&meta);
+    defer wm.deinit(io);
+    var seg_path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
+    const seg_path = try std.fmt.bufPrint(&seg_path_buf, "{s}/segments", .{data_dir});
+    var seg_dir = try Io.Dir.cwd().openDir(io, seg_path, .{ .iterate = true });
+    defer seg_dir.close(io);
+    const original = try seg_dir.readFileAlloc(io, "seg_0000000000.jss", testing.allocator, .limited(1 << 20));
+    defer testing.allocator.free(original);
+
+    try testing.expect(segment_io.armSpec("rename:1:eio"));
+    try testing.expectError(error.InputOutput, run(testing.allocator, io, seg_dir, &wm, null, .{ .rewrite_workers = 1 }));
+    try testing.expectEqual(@as(?u64, null), wm.load(io));
+    try testing.expectError(error.FileNotFound, seg_dir.openFile(io, "seg_0000000000.jss.tmp", .{}));
+    const after_fault = try seg_dir.readFileAlloc(io, "seg_0000000000.jss", testing.allocator, .limited(1 << 20));
+    defer testing.allocator.free(after_fault);
+    try testing.expectEqualSlices(u8, original, after_fault);
+
+    segment_io.disarm();
+    const recovered = try run(testing.allocator, io, seg_dir, &wm, null, .{ .rewrite_workers = 1 });
+    try testing.expectEqual(@as(u64, 1), recovered.segments_rewritten);
+    try testing.expectEqual(@as(u64, 1), recovered.rows_dropped);
+    try testing.expectEqual(@as(?u64, 2), wm.load(io));
+}
+
 test "pass: malformed account marker records failure without advancing watermark" {
     var threaded: Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
diff --git a/src/internal/segment_io.zig b/src/internal/segment_io.zig
new file mode 100644
index 0000000..94f2e79
--- /dev/null
+++ b/src/internal/segment_io.zig
@@ -0,0 +1,88 @@
+//! Deterministic segment I/O fault seam for the offline restart oracle.
+//!
+//! Every production segment write, fsync, and commit rename passes through
+//! these wrappers. With no armed fault they delegate directly to std.Io. The
+//! oracle may arm one operation ordinal to return a real filesystem-shaped
+//! error before the syscall, matching upstream segment.IOFaultInjector while
+//! retaining Stream's real JSS implementation and on-disk files.
+
+const std = @import("std");
+
+const Io = std.Io;
+const log = std.log.scoped(.stream);
+
+pub const Op = enum { write, sync, rename };
+pub const Kind = enum { enospc, eio, shortwrite };
+
+const Fault = struct {
+    op: ?Op = null,
+    ordinal: u64 = 0,
+    kind: Kind = .eio,
+    seen: std.atomic.Value(u64) = .init(0),
+
+    fn check(self: *Fault, op: Op) !void {
+        if (self.op == null or self.op.? != op) return;
+        const seen = self.seen.fetchAdd(1, .monotonic) + 1;
+        if (seen != self.ordinal) return;
+        log.warn("segment fault: injected {s} on {s} at ordinal {d}", .{ @tagName(self.kind), @tagName(op), seen });
+        return switch (self.kind) {
+            .enospc => error.NoSpaceLeft,
+            .eio => error.InputOutput,
+            .shortwrite => error.ShortWrite,
+        };
+    }
+};
+
+var fault: Fault = .{};
+
+/// Parse `<write|sync|rename>:<positive ordinal>:<enospc|eio|shortwrite>`.
+/// Must be called once during startup before worker threads exist.
+pub fn armSpec(spec: []const u8) bool {
+    var parts = std.mem.splitScalar(u8, spec, ':');
+    const op = std.meta.stringToEnum(Op, parts.next() orelse return false) orelse return false;
+    const ordinal = std.fmt.parseInt(u64, parts.next() orelse return false, 10) catch return false;
+    const kind = std.meta.stringToEnum(Kind, parts.next() orelse return false) orelse return false;
+    if (ordinal == 0 or parts.next() != null) return false;
+    if (op != .write and (kind == .enospc or kind == .shortwrite)) return false;
+    fault = .{ .op = op, .ordinal = ordinal, .kind = kind };
+    return true;
+}
+
+pub fn disarm() void {
+    fault = .{};
+}
+
+pub fn writeStreamingAll(file: *Io.File, io: Io, bytes: []const u8) !void {
+    try fault.check(.write);
+    try file.writeStreamingAll(io, bytes);
+}
+
+pub fn writePositionalAll(file: *Io.File, io: Io, bytes: []const u8, offset: u64) !void {
+    try fault.check(.write);
+    try file.writePositionalAll(io, bytes, offset);
+}
+
+pub fn sync(file: *Io.File, io: Io) !void {
+    try fault.check(.sync);
+    try file.sync(io);
+}
+
+pub fn rename(dir: Io.Dir, old: []const u8, new_dir: Io.Dir, new: []const u8, io: Io) !void {
+    try fault.check(.rename);
+    try dir.rename(old, new_dir, new, io);
+}
+
+test "fault spec is strict and fires only at the selected operation ordinal" {
+    const testing = std.testing;
+    disarm();
+    defer disarm();
+    try testing.expect(!armSpec(""));
+    try testing.expect(!armSpec("write:0:enospc"));
+    try testing.expect(!armSpec("sync:1:shortwrite"));
+    try testing.expect(!armSpec("rename:1:eio:extra"));
+    try testing.expect(armSpec("write:2:enospc"));
+    try fault.check(.sync);
+    try fault.check(.write);
+    try testing.expectError(error.NoSpaceLeft, fault.check(.write));
+    try fault.check(.write);
+}
diff --git a/src/internal/segment_patch.zig b/src/internal/segment_patch.zig
index ec714fc..2b4a287 100644
--- a/src/internal/segment_patch.zig
+++ b/src/internal/segment_patch.zig
@@ -11,6 +11,7 @@ const std = @import("std");
 const gloom = @import("gloom.zig");
 const segment = @import("segment.zig");
 const writer = @import("segment_writer.zig");
+const segment_io = @import("segment_io.zig");
 const zstd = @import("zstd.zig");
 
 const Allocator = std.mem.Allocator;
@@ -222,16 +223,19 @@ pub fn patchFile(
     const tmp_name = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{name});
     dir.deleteFile(io, tmp_name) catch |err| if (err != error.FileNotFound) return err;
     var file = try dir.createFile(io, tmp_name, .{ .truncate = true });
+    var file_open = true;
     errdefer {
-        file.close(io);
+        if (file_open) file.close(io);
         dir.deleteFile(io, tmp_name) catch {};
     }
-    try file.writeStreamingAll(io, new_bytes);
-    try file.sync(io);
+    try segment_io.writeStreamingAll(&file, io, new_bytes);
+    try segment_io.sync(&file, io);
     file.close(io);
-    try dir.rename(tmp_name, dir, name, io);
+    file_open = false;
+    try segment_io.rename(dir, tmp_name, dir, name, io);
     const dir_file: Io.File = .{ .handle = dir.handle, .flags = .{ .nonblocking = false } };
-    try dir_file.sync(io);
+    var sync_file = dir_file;
+    try segment_io.sync(&sync_file, io);
     return result;
 }
 
@@ -311,6 +315,53 @@ test "patch changes only indexed_at and preserves clean frames plus footer tail"
     }
 }
 
+test "patch file exhaustively propagates segment I/O faults without silent corruption" {
+    segment_io.disarm();
+    defer segment_io.disarm();
+    const old = try fixture(testing.allocator);
+    defer testing.allocator.free(old);
+
+    // Learn the deterministic committed output from the real implementation.
+    var reference_tmp = testing.tmpDir(.{});
+    defer reference_tmp.cleanup();
+    {
+        var file = try reference_tmp.dir.createFile(testing.io, "seg.jss", .{});
+        defer file.close(testing.io);
+        try file.writeStreamingAll(testing.io, old);
+    }
+    _ = try patchFile(testing.allocator, testing.io, reference_tmp.dir, "seg.jss", &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp);
+    const committed = try reference_tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20));
+    defer testing.allocator.free(committed);
+    try testing.expect(!std.mem.eql(u8, old, committed));
+
+    const cases = [_]struct { spec: []const u8, committed_after_error: bool }{
+        .{ .spec = "write:1:enospc", .committed_after_error = false },
+        .{ .spec = "write:1:shortwrite", .committed_after_error = false },
+        .{ .spec = "sync:1:eio", .committed_after_error = false },
+        .{ .spec = "rename:1:eio", .committed_after_error = false },
+        .{ .spec = "sync:2:eio", .committed_after_error = true },
+    };
+    for (cases) |case| {
+        var tmp = testing.tmpDir(.{});
+        defer tmp.cleanup();
+        {
+            var file = try tmp.dir.createFile(testing.io, "seg.jss", .{});
+            defer file.close(testing.io);
+            try file.writeStreamingAll(testing.io, old);
+        }
+        try testing.expect(segment_io.armSpec(case.spec));
+        try testing.expectError(switch (case.spec[0]) {
+            'w' => if (std.mem.endsWith(u8, case.spec, "enospc")) error.NoSpaceLeft else error.ShortWrite,
+            else => error.InputOutput,
+        }, patchFile(testing.allocator, testing.io, tmp.dir, "seg.jss", &.{}, Stamp{ .did = "did:plc:a", .value = 99 }, stamp));
+        segment_io.disarm();
+        try testing.expectError(error.FileNotFound, tmp.dir.openFile(testing.io, "seg.jss.tmp", .{}));
+        const got = try tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20));
+        defer testing.allocator.free(got);
+        try testing.expectEqualSlices(u8, if (case.committed_after_error) committed else old, got);
+    }
+}
+
 test "patch is idempotent and candidate bloom skip never calls mutate" {
     const old = try fixture(testing.allocator);
     defer testing.allocator.free(old);
diff --git a/src/internal/segment_rewrite.zig b/src/internal/segment_rewrite.zig
index fec0d9f..66a7fee 100644
--- a/src/internal/segment_rewrite.zig
+++ b/src/internal/segment_rewrite.zig
@@ -19,6 +19,7 @@ const footer_mod = @import("segment_footer.zig");
 const gloom = @import("gloom.zig");
 const segment = @import("segment.zig");
 const writer_mod = @import("segment_writer.zig");
+const segment_io = @import("segment_io.zig");
 const zstd = @import("zstd.zig");
 
 const Io = std.Io;
@@ -158,15 +159,21 @@ pub fn rewriteFile(
     var tmp_buf: [128]u8 = undefined;
     const tmp_name = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{name});
     var f = try dir.createFile(io, tmp_name, .{ .truncate = true });
-    errdefer f.close(io);
-    try f.writeStreamingAll(io, new_bytes);
-    try f.sync(io);
+    var file_open = true;
+    errdefer {
+        if (file_open) f.close(io);
+        dir.deleteFile(io, tmp_name) catch {};
+    }
+    try segment_io.writeStreamingAll(&f, io, new_bytes);
+    try segment_io.sync(&f, io);
     f.close(io);
-    try dir.rename(tmp_name, dir, name, io);
+    file_open = false;
+    try segment_io.rename(dir, tmp_name, dir, name, io);
     // make the rename durable: fsync the directory fd (std.Io.Dir has no
     // sync; the Threaded backend's Dir handle is a posix fd)
     const dir_file: Io.File = .{ .handle = dir.handle, .flags = .{ .nonblocking = false } };
-    try dir_file.sync(io);
+    var sync_file = dir_file;
+    try segment_io.sync(&sync_file, io);
     return result;
 }
 
@@ -275,6 +282,52 @@ test "rewrite: clean segment returns null, file layout preserved on drops" {
     try testing.expectEqual(@as(usize, 15), total);
 }
 
+test "rewrite file exhaustively propagates segment I/O faults without silent corruption" {
+    segment_io.disarm();
+    defer segment_io.disarm();
+    const old = try buildSealed(testing.allocator);
+    defer testing.allocator.free(old);
+
+    var reference_tmp = testing.tmpDir(.{});
+    defer reference_tmp.cleanup();
+    {
+        var file = try reference_tmp.dir.createFile(testing.io, "seg.jss", .{});
+        defer file.close(testing.io);
+        try file.writeStreamingAll(testing.io, old);
+    }
+    _ = try rewriteFile(testing.allocator, testing.io, reference_tmp.dir, "seg.jss", .{}, DropBelow{ .cutoff = 12 }, keepAbove);
+    const committed = try reference_tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20));
+    defer testing.allocator.free(committed);
+    try testing.expect(!std.mem.eql(u8, old, committed));
+
+    const cases = [_]struct { spec: []const u8, committed_after_error: bool }{
+        .{ .spec = "write:1:enospc", .committed_after_error = false },
+        .{ .spec = "write:1:shortwrite", .committed_after_error = false },
+        .{ .spec = "sync:1:eio", .committed_after_error = false },
+        .{ .spec = "rename:1:eio", .committed_after_error = false },
+        .{ .spec = "sync:2:eio", .committed_after_error = true },
+    };
+    for (cases) |case| {
+        var tmp = testing.tmpDir(.{});
+        defer tmp.cleanup();
+        {
+            var file = try tmp.dir.createFile(testing.io, "seg.jss", .{});
+            defer file.close(testing.io);
+            try file.writeStreamingAll(testing.io, old);
+        }
+        try testing.expect(segment_io.armSpec(case.spec));
+        try testing.expectError(switch (case.spec[0]) {
+            'w' => if (std.mem.endsWith(u8, case.spec, "enospc")) error.NoSpaceLeft else error.ShortWrite,
+            else => error.InputOutput,
+        }, rewriteFile(testing.allocator, testing.io, tmp.dir, "seg.jss", .{}, DropBelow{ .cutoff = 12 }, keepAbove));
+        segment_io.disarm();
+        try testing.expectError(error.FileNotFound, tmp.dir.openFile(testing.io, "seg.jss.tmp", .{}));
+        const got = try tmp.dir.readFileAlloc(testing.io, "seg.jss", testing.allocator, .limited(1 << 20));
+        defer testing.allocator.free(got);
+        try testing.expectEqualSlices(u8, if (case.committed_after_error) committed else old, got);
+    }
+}
+
 test "rewrite: unrelated candidate DIDs skip the real segment decode" {
     const old = try buildSealed(testing.allocator);
     defer testing.allocator.free(old);
diff --git a/src/main.zig b/src/main.zig
index 7f5bef1..65e23e7 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -19,6 +19,7 @@ const meta_store = @import("internal/meta_store.zig");
 const pipeline_mod = @import("internal/pipeline.zig");
 const repair_mod = @import("internal/repair.zig");
 const server_mod = @import("internal/server.zig");
+const segment_io = @import("internal/segment_io.zig");
 const steady_mod = @import("internal/compact/steady.zig");
 const tail_mod = @import("internal/tail.zig");
 const timestamp_rules = @import("internal/timestamp/rules.zig");
@@ -124,6 +125,7 @@ pub fn main(init: std.process.Init.Minimal) !void {
     var repo_action_rate_limits = true;
     var store_fault_prefix: ?[]const u8 = null;
     var store_fault_ordinal: u64 = 1;
+    var segment_fault_spec: ?[]const u8 = null;
 
     var arg_it = init.args.iterate();
     _ = arg_it.next(); // program name
@@ -169,6 +171,9 @@ pub fn main(init: std.process.Init.Minimal) !void {
             store_fault_prefix = arg["--store-fault-prefix=".len..];
         } else if (std.mem.startsWith(u8, arg, "--store-fault-ordinal=")) {
             store_fault_ordinal = try std.fmt.parseInt(u64, arg["--store-fault-ordinal=".len..], 10);
+        } else if (std.mem.startsWith(u8, arg, "--segment-fault=")) {
+            // debug/oracle only: fail one real segment operation
+            segment_fault_spec = arg["--segment-fault=".len..];
         } else if (std.mem.eql(u8, arg, "--backfill")) {
             do_backfill = true;
         } else if (std.mem.eql(u8, arg, "--no-verify")) {
@@ -188,6 +193,12 @@ pub fn main(init: std.process.Init.Minimal) !void {
             return error.BadArgs;
         }
     }
+    if (segment_fault_spec) |spec| {
+        if (!segment_io.armSpec(spec)) {
+            log.err("invalid segment fault; expected op:positive-ordinal:kind", .{});
+            return error.BadArgs;
+        }
+    }
     const backfill_repos: []const []const u8 = if (backfill_repos_raw) |raw| try parseBackfillRepos(allocator, raw) else &.{};
     defer if (backfill_repos.len > 0) allocator.free(backfill_repos);
     if (backfill_repos.len > 0 and backfill_max_repos > 0) return error.ConflictingBackfillSelection;
@@ -211,7 +222,10 @@ pub fn main(init: std.process.Init.Minimal) !void {
     };
     defer hub.deinit();
 
-    var archive = try archive_mod.Archive.initWithStats(allocator, io, data_dir, &stats);
+    var archive = archive_mod.Archive.initWithStats(allocator, io, data_dir, &stats) catch |err| {
+        logPersistenceFailure(data_dir, err);
+        return err;
+    };
     var archive_closed = false;
     defer {
         if (!archive_closed) archive.close();
@@ -330,7 +344,7 @@ pub fn main(init: std.process.Init.Minimal) !void {
     const mid_lifecycle = if (stored_phase) |ph| ph != .steady_state else false;
     if (mid_lifecycle or do_backfill or backfill_max_repos > 0 or backfill_repos.len > 0) {
         hub.serving.store(false, .release);
-        try lifecycle.run(allocator, io, .{
+        lifecycle.run(allocator, io, .{
             .data_dir = data_dir,
             .upstream = upstream,
             .relay_http = relay_http,
@@ -340,7 +354,10 @@ pub fn main(init: std.process.Init.Minimal) !void {
             .workers = backfill_workers,
             .max_inflight_bytes = backfill_max_inflight_bytes,
             .compaction_enabled = compaction_interval_s != 0,
-        }, &archive, &meta, &cursor_store, if (verifier) |*v| v else null, &stats);
+        }, &archive, &meta, &cursor_store, if (verifier) |*v| v else null, &stats) catch |err| {
+            logPersistenceFailure(data_dir, err);
+            return err;
+        };
         hub.serving.store(true, .release);
     }
     if (cursor == null) {
@@ -464,6 +481,12 @@ pub fn main(init: std.process.Init.Minimal) !void {
     archive_closed = true;
 }
 
+fn logPersistenceFailure(data_dir: []const u8, err: anyerror) void {
+    if (err == error.NoSpaceLeft) {
+        log.err("fatal persistence error: disk full under {s}; free space, then restart stream", .{data_dir});
+    }
+}
+
 fn runConsumer(consumer: *ingest.Consumer, done: *std.atomic.Value(bool)) !void {
     defer done.store(true, .release);
     try consumer.run();
diff --git a/tests/oracle.py b/tests/oracle.py
index 09325db..ec2e65a 100644
--- a/tests/oracle.py
+++ b/tests/oracle.py
@@ -283,6 +283,52 @@ def run_store_fault_case():
         log.close()
 
 
+def run_segment_fault_cases():
+    """Real JSS write/fsync failures must fail loud and recover cleanly."""
+    cases = [
+        ("write:1:enospc", "NoSpaceLeft", True),
+        ("sync:2:eio", "InputOutput", False),
+        ("write:3:shortwrite", "ShortWrite", False),
+    ]
+    for spec, error_name, wants_disk_message in cases:
+        shutil.rmtree(DATA, ignore_errors=True)
+        label = spec.replace(":", "-")
+        log_path = pathlib.Path(f"/tmp/oracle-segment-fault-{label}.log")
+        log = open(log_path, "w")
+        p = subprocess.Popen(ARGS + [f"--segment-fault={spec}"], stdout=log, stderr=log)
+        rc = p.wait(timeout=300)
+        assert rc != 0, f"{spec}: armed segment-fault run completed instead of failing loud"
+        log.flush()
+        logged = log_path.read_text()
+        op, ordinal, kind = spec.split(":")
+        marker = f"segment fault: injected {kind} on {op} at ordinal {ordinal}"
+        assert marker in logged, f"{spec}: segment fault never fired (anti-vacuity)"
+        assert error_name in logged, f"{spec}: injected segment failure did not reach process boundary"
+        if wants_disk_message:
+            assert "fatal persistence error: disk full" in logged
+            assert str(DATA) in logged
+            assert "restart stream" in logged
+
+        p = subprocess.Popen(ARGS, stdout=log, stderr=log)
+        try:
+            wait_serving(p)
+            st, body = http_status("/xrpc/network.bsky.jetstream.listSegments")
+            assert st == 200, st
+            segs = json.loads(body)["segments"]
+            replay_count = min(50, sum(s["eventCount"] for s in segs))
+            assert replay_count > 0, segs
+            assert asyncio.run(replay_invariants(replay_count)) == replay_count
+            assert not (DATA / "backfill").exists(), f"{spec}: recovery did not finish cleanup"
+            assert not list((DATA / "segments").glob("*.tmp")), f"{spec}: stale segment tmp survived"
+        finally:
+            p.send_signal(signal.SIGTERM)
+            try:
+                p.wait(timeout=15)
+            except subprocess.TimeoutExpired:
+                p.kill()
+            log.close()
+
+
 def run_backfill_metrics_receipt():
     """A clean production run must exercise, not merely register, backfill metrics."""
     shutil.rmtree(DATA, ignore_errors=True)
@@ -344,6 +390,10 @@ def main():
     run_store_fault_case()
     print(f"ok store fault recovery ({time.time() - t0:.1f}s)")
 
+    t0 = time.time()
+    run_segment_fault_cases()
+    print(f"ok segment fault recovery ({time.time() - t0:.1f}s)")
+
     for point in CRASHPOINTS:
         t0 = time.time()
         run_case(point)