diff --git a/deploy/README.md b/deploy/README.md
index ad65dd7..3479441 100644
--- a/deploy/README.md
+++ b/deploy/README.md
@@ -79,8 +79,11 @@ Absolute throughput still needs the dedicated-node soak before public cutover.
 
 `grafana/jetstream-upstream.json` is the byte-exact dashboard from the pinned
 Jetstream V2 reference. `grafana/stream.json` is the deployable artifact,
-generated deterministically from that source with only the Go runtime row
-adapted to honest process metrics. Run `just dashboard-test`, then
+generated deterministically from that source. The Go runtime row is adapted to
+honest process metrics, and one explicit Stream-only row exposes scheduler
+tickets, unsettled stage gaps, and terminal failures during bootstrap and
+steady state; every upstream protocol/lifecycle panel remains unchanged. Run
+`just dashboard-test`, then
 `just process-metrics-contract`, then
 `just dashboard-contract <metrics-url-or-saved-scrape>`. These checks are
 network-independent when given a saved scrape or the loopback harness and
diff --git a/deploy/grafana/stream.json b/deploy/grafana/stream.json
index d93df34..3d8b861 100644
--- a/deploy/grafana/stream.json
+++ b/deploy/grafana/stream.json
@@ -42,7 +42,7 @@
       }
     ]
   },
-  "description": "Stream operational dashboard adapted from the checksum-pinned Jetstream V2 dashboard. Protocol and lifecycle rows are unchanged; only the Go runtime row uses Stream process metrics.",
+  "description": "Stream operational dashboard adapted from the checksum-pinned Jetstream V2 dashboard. Protocol and lifecycle rows are unchanged; the runtime row uses Stream process metrics, and a final Stream-only row exposes its live scheduler during bootstrap and steady state.",
   "editable": true,
   "fiscalYearStartMonth": 0,
   "graphTooltip": 1,
@@ -7210,6 +7210,317 @@
           "instant": false
         }
       ]
+    },
+    {
+      "id": 76,
+      "type": "row",
+      "title": "Stream pipeline diagnostics",
+      "collapsed": false,
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 170
+      },
+      "panels": [],
+      "description": "Live during bootstrap and steady state; gaps and terminal failures are never inferred from archive throughput."
+    },
+    {
+      "id": 77,
+      "type": "timeseries",
+      "title": "Pipeline tickets",
+      "description": "Cumulative scheduler boundaries. Emitted is the historical name for every settled ticket, including explicit drops and verifier rejections.",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${datasource}"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 0,
+        "y": 171
+      },
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "drawStyle": "line",
+            "lineInterpolation": "linear",
+            "lineWidth": 1,
+            "fillOpacity": 6,
+            "gradientMode": "none",
+            "showPoints": "never",
+            "pointSize": 4,
+            "spanNulls": true,
+            "insertNulls": false,
+            "barAlignment": 0,
+            "barWidthFactor": 0.6,
+            "axisPlacement": "auto",
+            "axisLabel": "",
+            "axisColorMode": "text",
+            "axisBorderShow": false,
+            "axisCenteredZero": false,
+            "scaleDistribution": {
+              "type": "linear"
+            },
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "stacking": {
+              "mode": "none",
+              "group": "A"
+            },
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              }
+            ]
+          },
+          "unit": "short",
+          "min": 0
+        },
+        "overrides": []
+      },
+      "options": {
+        "legend": {
+          "calcs": [],
+          "displayMode": "list",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "mode": "multi",
+          "sort": "desc"
+        }
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${datasource}"
+          },
+          "expr": "max(stream_pipeline_ticket{job=~\"$job\",instance=~\"$instance\"}) by (stage)",
+          "legendFormat": "{{stage}}",
+          "refId": "A",
+          "range": true,
+          "instant": false
+        }
+      ]
+    },
+    {
+      "id": 78,
+      "type": "timeseries",
+      "title": "Pipeline stage gaps",
+      "description": "Current unsettled work split between the unclaimed verifier queue and claimed work awaiting settlement. Their sum is submitted-to-emitted.",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${datasource}"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 8,
+        "y": 171
+      },
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "drawStyle": "line",
+            "lineInterpolation": "linear",
+            "lineWidth": 1,
+            "fillOpacity": 6,
+            "gradientMode": "none",
+            "showPoints": "never",
+            "pointSize": 4,
+            "spanNulls": true,
+            "insertNulls": false,
+            "barAlignment": 0,
+            "barWidthFactor": 0.6,
+            "axisPlacement": "auto",
+            "axisLabel": "",
+            "axisColorMode": "text",
+            "axisBorderShow": false,
+            "axisCenteredZero": false,
+            "scaleDistribution": {
+              "type": "linear"
+            },
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "stacking": {
+              "mode": "none",
+              "group": "A"
+            },
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              }
+            ]
+          },
+          "unit": "short",
+          "min": 0
+        },
+        "overrides": []
+      },
+      "options": {
+        "legend": {
+          "calcs": [],
+          "displayMode": "list",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "mode": "multi",
+          "sort": "desc"
+        }
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${datasource}"
+          },
+          "expr": "max(stream_pipeline_stage_gap{job=~\"$job\",instance=~\"$instance\"}) by (boundary)",
+          "legendFormat": "{{boundary}}",
+          "refId": "A",
+          "range": true,
+          "instant": false
+        }
+      ]
+    },
+    {
+      "id": 79,
+      "type": "timeseries",
+      "title": "Pipeline terminal failures",
+      "description": "Any non-zero value is deployment-significant. The gauge marks the current pipeline fatal; the counters retain its first failing stage.",
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${datasource}"
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 8,
+        "x": 16,
+        "y": 171
+      },
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "drawStyle": "line",
+            "lineInterpolation": "linear",
+            "lineWidth": 1,
+            "fillOpacity": 6,
+            "gradientMode": "none",
+            "showPoints": "never",
+            "pointSize": 4,
+            "spanNulls": true,
+            "insertNulls": false,
+            "barAlignment": 0,
+            "barWidthFactor": 0.6,
+            "axisPlacement": "auto",
+            "axisLabel": "",
+            "axisColorMode": "text",
+            "axisBorderShow": false,
+            "axisCenteredZero": false,
+            "scaleDistribution": {
+              "type": "linear"
+            },
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "stacking": {
+              "mode": "none",
+              "group": "A"
+            },
+            "thresholdsStyle": {
+              "mode": "line"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 1
+              }
+            ]
+          },
+          "unit": "short",
+          "min": 0
+        },
+        "overrides": []
+      },
+      "options": {
+        "legend": {
+          "calcs": [],
+          "displayMode": "list",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "mode": "multi",
+          "sort": "desc"
+        }
+      },
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${datasource}"
+          },
+          "expr": "max(stream_pipeline_fatal{job=~\"$job\",instance=~\"$instance\"})",
+          "legendFormat": "fatal",
+          "refId": "A",
+          "range": true,
+          "instant": false
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${datasource}"
+          },
+          "expr": "sum(stream_pipeline_failures_total{job=~\"$job\",instance=~\"$instance\"}) by (stage)",
+          "legendFormat": "{{stage}} failures",
+          "refId": "B",
+          "range": true,
+          "instant": false
+        }
+      ]
     }
   ],
   "refresh": "30s",
diff --git a/docs/full-network-experiment-2026-07.md b/docs/full-network-experiment-2026-07.md
index fcd16d6..9d399ef 100644
--- a/docs/full-network-experiment-2026-07.md
+++ b/docs/full-network-experiment-2026-07.md
@@ -165,3 +165,34 @@ the intended concurrency multiplier: retained bytes scale with the sizes of up
 to 32 simultaneous complete repositories. A parity-default candidate therefore
 requires the upstream-sized 32 GiB memory class Jim reported; the 4 GiB
 colocated canary is not an admissible host for that build.
+
+## 2026-07-22 isolated full-network experiment
+
+The second experiment used an isolated CCX43 workload, an independent
+watchdog, an off-host four-mode public consumer, and a 2.5 TB XFS Cloud Volume.
+It was intentionally destroyed early, and the Hetzner project was verified
+empty afterward. The live-price cost ceiling at teardown was approximately
+$2.53.
+
+Before teardown, 26,099 repositories had crossed the durable completion
+boundary. The workload used approximately 73.9 GB of disk and 12.3 GB RSS,
+while block writes were approximately 36.9 MB/s. Those are partial-run sizing
+receipts, not a full-network completion estimate.
+
+The experiment exposed a correctness failure more important than throughput:
+the live scheduler stopped at 399,362 archived events while its WebSocket
+reader continued and advanced its reconnect cursor 459,475 upstream sequences
+beyond the last accepted frame. The process stayed alive and bootstrap kept
+crawling, so ordinary health and backfill counters concealed the dead live
+path. The exact original persistence error was not retained because the build
+had neither a terminal pipeline metric nor fail-loud capture supervision.
+
+Commit `b7cd5f0` closes the demonstrated loss mechanism. Zat acknowledges a raw
+frame only after Stream accepts it; Stream propagates a closed pipeline's
+original cause; bootstrap races live capture against the repository crawl; and
+a failure between frames can no longer degrade into `LiveCaptureEnded`.
+Real RocksDB fault injection, same-disk restart, the complete eight-crashpoint
+oracle, and the pinned upstream differential oracle all pass. The next
+experiment additionally requires pipeline tickets, stage gaps, and terminal
+failure signals to remain scrapeable throughout bootstrap, not only after
+steady-state construction.
diff --git a/docs/grafana-dashboard.md b/docs/grafana-dashboard.md
index bdc309e..e12b771 100644
--- a/docs/grafana-dashboard.md
+++ b/docs/grafana-dashboard.md
@@ -11,10 +11,15 @@ The exact reviewed bytes are vendored at
 `deploy/grafana/jetstream-upstream.json`; deployment does not depend on the
 network or a moving branch. Stream provisions
 `deploy/grafana/stream.json` (SHA-256
-`e27537b5a951bc1b5a6133aaef068b7b4b1eb2abff2e2f283722697cf6fe7017`). It is
+`e64377c02d50e9cbe0e2c1c78ede4a30aa61bdca751ac9783f63f9c08f3d3951`). It is
 deterministically generated by `tools/adapt_dashboard.py` from the pinned
 source. The generator refuses a changed source checksum, and the tests prove
-every non-runtime panel remains structurally identical to upstream.
+every upstream non-runtime panel remains structurally identical. Stream adds
+one explicit final diagnostics row for the scheduler which has no upstream Go
+equivalent: ticket boundaries, exact unsettled stage gaps, current fatal state,
+and first-failure counters partitioned by scheduler/emit/repair/cursor stage.
+These producers are active during historical bootstrap as well as steady
+state; the dashboard never infers scheduler health from archive throughput.
 
 `just dashboard-test` verifies both checksums, regeneration, and the contract
 parser. `just dashboard-contract <scrape-file-or-localhost-url>` extracts every
@@ -75,6 +80,10 @@ The following dashboard families already have non-placeholder producers:
 - `process_voluntary_context_switches_total`
 - `process_involuntary_context_switches_total`
 - `stream_process_threads`
+- `stream_pipeline_ticket`
+- `stream_pipeline_stage_gap`
+- `stream_pipeline_fatal`
+- `stream_pipeline_failures_total`
 - `jetstream_build_info`
 - `jetstream_livestream_events_received_total`
 - `jetstream_livestream_decode_errors_total`
diff --git a/src/internal/metrics.zig b/src/internal/metrics.zig
index c62a891..12e566d 100644
--- a/src/internal/metrics.zig
+++ b/src/internal/metrics.zig
@@ -43,6 +43,7 @@ pub const OrchestratorState = enum {
 };
 pub const CompactionPassResult = enum { ok, error_result };
 pub const CompactionReason = enum { record, sync, account };
+pub const PipelineFailureStage = enum { scheduler, emit, repair, cursor };
 
 const http_series_count = @typeInfo(HttpHandler).@"enum".fields.len *
     @typeInfo(HttpMethod).@"enum".fields.len * @typeInfo(HttpCode).@"enum".fields.len;
@@ -119,6 +120,15 @@ pub const Stats = struct {
     livestream_stream_error_other_total: Counter = .init(0),
     livestream_stream_error_missing_total: Counter = .init(0),
     livestream_last_seen_upstream_event_timestamp_seconds: std.atomic.Value(i64) = .init(0),
+    pipeline_submitted: Counter = .init(0),
+    pipeline_claimed: Counter = .init(0),
+    /// Tickets settled by drop-oldest before a verifier worker claimed them.
+    pipeline_unclaimed_dropped: Counter = .init(0),
+    /// Historical `emitted` name: every settled ticket, including drops and
+    /// verifier rejections. This is the scheduler's durable drain boundary.
+    pipeline_emitted: Counter = .init(0),
+    pipeline_fatal: Counter = .init(0),
+    pipeline_failures: std.enums.EnumArray(PipelineFailureStage, Counter) = .initFill(.init(0)),
     compaction_watermark: Counter = .init(0),
     compaction_watermark_lag_us: Counter = .init(0),
     compaction_passes: std.enums.EnumArray(CompactionPassResult, Counter) = .initFill(.init(0)),
@@ -463,6 +473,39 @@ pub fn format(
         tail.tip,
     }) catch {};
 
+    const pipeline_submitted = stats.pipeline_submitted.load(.monotonic);
+    const pipeline_claimed = stats.pipeline_claimed.load(.monotonic);
+    const pipeline_unclaimed_dropped = stats.pipeline_unclaimed_dropped.load(.monotonic);
+    const pipeline_emitted = stats.pipeline_emitted.load(.monotonic);
+    const pipeline_claimed_settled = pipeline_emitted -| pipeline_unclaimed_dropped;
+    w.print(
+        \\# TYPE stream_pipeline_ticket gauge
+        \\stream_pipeline_ticket{{stage="submitted"}} {d}
+        \\stream_pipeline_ticket{{stage="claimed"}} {d}
+        \\stream_pipeline_ticket{{stage="emitted"}} {d}
+        \\# TYPE stream_pipeline_stage_gap gauge
+        \\stream_pipeline_stage_gap{{boundary="submitted_to_claimed"}} {d}
+        \\stream_pipeline_stage_gap{{boundary="claimed_to_emitted"}} {d}
+        \\stream_pipeline_stage_gap{{boundary="submitted_to_emitted"}} {d}
+        \\# TYPE stream_pipeline_fatal gauge
+        \\stream_pipeline_fatal {d}
+        \\# TYPE stream_pipeline_failures_total counter
+        \\
+    , .{
+        pipeline_submitted,
+        pipeline_claimed,
+        pipeline_emitted,
+        pipeline_submitted -| pipeline_claimed -| pipeline_unclaimed_dropped,
+        pipeline_claimed -| pipeline_claimed_settled,
+        pipeline_submitted -| pipeline_emitted,
+        stats.pipeline_fatal.load(.monotonic),
+    }) catch {};
+    inline for (comptime std.enums.values(PipelineFailureStage)) |stage| {
+        w.print("stream_pipeline_failures_total{{stage=\"{s}\"}} {d}\n", .{
+            @tagName(stage), stats.pipeline_failures.getPtrConst(stage).load(.monotonic),
+        }) catch {};
+    }
+
     w.print(
         \\# TYPE process_start_time_seconds gauge
         \\process_start_time_seconds {d}
@@ -1392,6 +1435,11 @@ test "format renders counters" {
     _ = stats.orchestrator_merge_did_lookups_total.fetchAdd(20, .monotonic);
     _ = stats.orchestrator_merge_repo_revs_updated_total.fetchAdd(21, .monotonic);
     _ = stats.orchestrator_merge_dids_discovered_post_bootstrap_total.fetchAdd(22, .monotonic);
+    stats.pipeline_submitted.store(20, .monotonic);
+    stats.pipeline_claimed.store(17, .monotonic);
+    stats.pipeline_emitted.store(11, .monotonic);
+    stats.pipeline_fatal.store(1, .monotonic);
+    _ = stats.pipeline_failures.getPtr(.cursor).fetchAdd(2, .monotonic);
 
     var buf: [64 * 1024]u8 = undefined;
     const out = format(&buf, &stats, 2, .{
@@ -1463,6 +1511,12 @@ test "format renders counters" {
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_did_lookups_total 20") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_repo_revs_updated_total 21") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "jetstream_orchestrator_merge_dids_discovered_post_bootstrap_total 22") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_ticket{stage=\"submitted\"} 20") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_stage_gap{boundary=\"submitted_to_claimed\"} 3") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_stage_gap{boundary=\"claimed_to_emitted\"} 6") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_stage_gap{boundary=\"submitted_to_emitted\"} 9") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_fatal 1") != null);
+    try std.testing.expect(std.mem.indexOf(u8, out, "stream_pipeline_failures_total{stage=\"cursor\"} 2") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_repos_total{result=\"complete\"} 11") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_workers_active 4") != null);
     try std.testing.expect(std.mem.indexOf(u8, out, "stream_backfill_queue_depth 23") != null);
diff --git a/src/internal/pipeline.zig b/src/internal/pipeline.zig
index 9c9abb4..210504a 100644
--- a/src/internal/pipeline.zig
+++ b/src/internal/pipeline.zig
@@ -92,6 +92,13 @@ pub const Pipeline = struct {
     futures: [worker_count + 1]?Io.Future(void) = @splat(null),
 
     pub fn start(self: *Pipeline) !void {
+        if (self.stats) |stats| {
+            stats.pipeline_submitted.store(0, .monotonic);
+            stats.pipeline_claimed.store(0, .monotonic);
+            stats.pipeline_unclaimed_dropped.store(0, .monotonic);
+            stats.pipeline_emitted.store(0, .monotonic);
+            stats.pipeline_fatal.store(0, .monotonic);
+        }
         var spawned: usize = 0;
         errdefer {
             self.signalStop();
@@ -189,7 +196,12 @@ pub const Pipeline = struct {
     pub fn fatalError(self: *Pipeline) ?anyerror {
         self.mutex.lockUncancelable(self.io);
         defer self.mutex.unlock(self.io);
-        return self.fatal_error orelse if (self.repair) |coordinator| coordinator.fatalError() else null;
+        if (self.fatal_error) |err| return err;
+        if (self.repair) |coordinator| if (coordinator.fatalError()) |err| {
+            self.recordFatalLocked(.repair, err);
+            return err;
+        };
+        return null;
     }
 
     /// called by the websocket reader. Decode happens on this dispatch path,
@@ -269,10 +281,12 @@ pub const Pipeline = struct {
             self.active_chains += 1;
             self.work_cond.signal(self.io);
         }
+        if (self.stats) |stats| stats.pipeline_submitted.store(self.next_submit, .monotonic);
     }
 
     fn dropJobLocked(self: *Pipeline, job: *Job) void {
         if (job.seq) |seq| _ = self.inflight.remove(seq);
+        if (self.stats) |stats| _ = stats.pipeline_unclaimed_dropped.fetchAdd(1, .monotonic);
         self.markProcessedLocked(job.ticket);
         if (self.stats) |stats| _ = stats.verify_queue_dropped_total.fetchAdd(1, .monotonic);
         log.warn("verify queue overflow dropped oldest event: key={s} ticket={d}", .{ job.key, job.ticket });
@@ -301,6 +315,7 @@ pub const Pipeline = struct {
             while (true) {
                 self.mutex.lockUncancelable(self.io);
                 self.claimed += 1;
+                if (self.stats) |stats| stats.pipeline_claimed.store(self.claimed, .monotonic);
                 self.mutex.unlock(self.io);
 
                 var verdict: verify.Result = .{};
@@ -346,7 +361,7 @@ pub const Pipeline = struct {
                 self.completed.append(self.allocator, job) catch |err| {
                     self.mutex.unlock(self.io);
                     owned_job = null;
-                    self.failJob(job, err);
+                    self.failJob(job, .scheduler, err);
                     return;
                 };
                 owned_job = null;
@@ -420,7 +435,7 @@ pub const Pipeline = struct {
                         self.finishJob(job);
                     } else {
                         batch.append(self.allocator, .{ .job = job }) catch |err| {
-                            self.failJob(job, err);
+                            self.failJob(job, .scheduler, err);
                             return;
                         };
                         self.mutex.lockUncancelable(self.io);
@@ -468,7 +483,7 @@ pub const Pipeline = struct {
                     if (job.event) |event| self.emit(self.emit_ctx, event, job.verdict, trace.context()) catch |err| {
                         trace.fail(err);
                         release_from = i + 1;
-                        self.failJob(job, err);
+                        self.failJob(job, .emit, err);
                         return;
                     };
                     if (job.seq) |seq| max_seq = @max(max_seq, seq);
@@ -481,7 +496,7 @@ pub const Pipeline = struct {
                         release_from = i + 1;
                         completion.deinit();
                         self.finishBufferedRepair();
-                        self.failPipeline(error.MissingRepairEmitter);
+                        self.failPipeline(.repair, error.MissingRepairEmitter);
                         return;
                     };
                     callback(self.emit_ctx, completion, trace.context()) catch |err| {
@@ -489,7 +504,7 @@ pub const Pipeline = struct {
                         release_from = i + 1;
                         completion.deinit();
                         self.finishBufferedRepair();
-                        self.failPipeline(err);
+                        self.failPipeline(.repair, err);
                         return;
                     };
                     completion.deinit();
@@ -502,7 +517,7 @@ pub const Pipeline = struct {
                 if (watermark > self.cursor_watermark) {
                     if (self.emit_cursor) |callback| callback(self.emit_ctx, watermark) catch |err| {
                         trace.fail(err);
-                        self.failPipeline(err);
+                        self.failPipeline(.cursor, err);
                         return;
                     };
                     self.cursor_watermark = watermark;
@@ -538,12 +553,13 @@ pub const Pipeline = struct {
 
     fn markProcessedLocked(self: *Pipeline, ticket: u64) void {
         self.processed += 1;
+        if (self.stats) |stats| stats.pipeline_emitted.store(self.processed, .monotonic);
         if (ticket == self.processed_prefix) {
             self.processed_prefix += 1;
             while (self.processed_sparse.remove(self.processed_prefix)) self.processed_prefix += 1;
         } else if (ticket > self.processed_prefix) {
             self.processed_sparse.put(self.allocator, ticket, {}) catch {
-                if (self.fatal_error == null) self.fatal_error = error.OutOfMemory;
+                self.recordFatalLocked(.scheduler, error.OutOfMemory);
                 self.alive = false;
             };
         }
@@ -584,7 +600,7 @@ pub const Pipeline = struct {
             batch.append(self.allocator, .{ .repair = completion }) catch |err| {
                 completion.deinit();
                 self.finishBufferedRepair();
-                self.failPipeline(err);
+                self.failPipeline(.scheduler, err);
                 return false;
             };
             if (deadline.* == null)
@@ -623,14 +639,14 @@ pub const Pipeline = struct {
         self.wake.set(self.io);
     }
 
-    fn failJob(self: *Pipeline, job: *Job, err: anyerror) void {
+    fn failJob(self: *Pipeline, job: *Job, stage: metrics.PipelineFailureStage, err: anyerror) void {
         self.freeJob(job);
-        self.failPipeline(err);
+        self.failPipeline(stage, err);
     }
 
-    fn failPipeline(self: *Pipeline, err: anyerror) void {
+    fn failPipeline(self: *Pipeline, stage: metrics.PipelineFailureStage, err: anyerror) void {
         self.mutex.lockUncancelable(self.io);
-        if (self.fatal_error == null) self.fatal_error = err;
+        self.recordFatalLocked(stage, err);
         self.alive = false;
         self.work_cond.broadcast(self.io);
         self.space_cond.broadcast(self.io);
@@ -638,6 +654,17 @@ pub const Pipeline = struct {
         self.mutex.unlock(self.io);
     }
 
+    /// Caller holds mutex. Count exactly the first terminal cause; secondary
+    /// shutdown errors must not obscure the stage which stopped ingestion.
+    fn recordFatalLocked(self: *Pipeline, stage: metrics.PipelineFailureStage, err: anyerror) void {
+        if (self.fatal_error != null) return;
+        self.fatal_error = err;
+        if (self.stats) |stats| {
+            stats.pipeline_fatal.store(1, .monotonic);
+            _ = stats.pipeline_failures.getPtr(stage).fetchAdd(1, .monotonic);
+        }
+    }
+
     fn compactReadyLocked(self: *Pipeline) void {
         if (self.ready_read == self.ready.items.len) {
             self.ready.clearRetainingCapacity();
@@ -1155,10 +1182,12 @@ test "pipeline exposes emit failure without advancing emitted ticket" {
         }
     };
     var context: u8 = 0;
+    var stats: metrics.Stats = .{};
     var p: Pipeline = .{
         .allocator = testing.allocator,
         .io = threaded.io(),
         .verifier = null,
+        .stats = &stats,
         .emit_ctx = &context,
         .emit = Failing.emit,
     };
@@ -1177,9 +1206,52 @@ test "pipeline exposes emit failure without advancing emitted ticket" {
     const gauges = p.gauges();
     try testing.expectEqual(@as(u64, 1), gauges.submitted);
     try testing.expectEqual(@as(u64, 0), gauges.emitted);
+    try testing.expectEqual(@as(u64, 1), stats.pipeline_submitted.load(.monotonic));
+    try testing.expectEqual(@as(u64, 1), stats.pipeline_claimed.load(.monotonic));
+    try testing.expectEqual(@as(u64, 0), stats.pipeline_emitted.load(.monotonic));
+    try testing.expectEqual(@as(u64, 1), stats.pipeline_fatal.load(.monotonic));
+    try testing.expectEqual(@as(u64, 1), stats.pipeline_failures.getPtrConst(.emit).load(.monotonic));
     try testing.expectError(error.Closed, p.submit(frame));
 }
 
+test "pipeline classifies cursor persistence failure" {
+    var threaded: Io.Threaded = .init(testing.allocator, .{});
+    defer threaded.deinit();
+
+    const FailingCursor = struct {
+        fn emit(_: *anyopaque, _: zat.FirehoseEvent, _: verify.Result, _: observability.Context) !void {}
+        fn cursor(_: *anyopaque, _: i64) !void {
+            return error.CursorPersistenceFailed;
+        }
+    };
+    var context: u8 = 0;
+    var stats: metrics.Stats = .{};
+    var p: Pipeline = .{
+        .allocator = testing.allocator,
+        .io = threaded.io(),
+        .verifier = null,
+        .stats = &stats,
+        .emit_ctx = &context,
+        .emit = FailingCursor.emit,
+        .emit_cursor = FailingCursor.cursor,
+    };
+    try p.start();
+    defer p.stop();
+    const frame = try controlFrame(testing.allocator, .{ .identity = .{
+        .seq = 42,
+        .did = "did:plc:cursor-failure",
+        .time = "2026-07-16T00:00:00Z",
+    } });
+    defer testing.allocator.free(frame);
+    try p.submit(frame);
+
+    while (p.fatalError() == null) std.Thread.yield() catch {};
+    try testing.expectEqual(error.CursorPersistenceFailed, p.fatalError().?);
+    try testing.expectEqual(@as(u64, 1), stats.pipeline_fatal.load(.monotonic));
+    try testing.expectEqual(@as(u64, 1), stats.pipeline_failures.getPtrConst(.cursor).load(.monotonic));
+    try testing.expectEqual(@as(u64, 0), stats.pipeline_failures.getPtrConst(.emit).load(.monotonic));
+}
+
 test "malformed frame advances ordered verifier registration" {
     var threaded: Io.Threaded = .init(testing.allocator, .{});
     defer threaded.deinit();
diff --git a/src/internal/server.zig b/src/internal/server.zig
index f535fff..3465483 100644
--- a/src/internal/server.zig
+++ b/src/internal/server.zig
@@ -102,8 +102,6 @@ pub const Hub = struct {
     status_identity_resolution: bool = true,
     repo_action_rate_limits: bool = true,
     repo_action_limiter: repo_action_limiter.Limiter = .{},
-    pipeline_gauges: ?*const fn (*anyopaque) [3]u64 = null,
-    pipeline_ctx: ?*anyopaque = null,
     active: std.atomic.Value(u32) = .init(0),
     bootstrap_enabled: bool = false,
     compaction_enabled: bool = false,
@@ -1058,16 +1056,7 @@ fn respondMetrics(conn: *websocket.Conn, hub: *Hub, observation: *HttpObservatio
         .base = g.base,
         .tip = g.tip,
     }, now_s, process_metrics.collect(hub.io), if (hub.archive) |archive| disk_space.freeBytes(archive.dir) else null);
-    var extra_buf: [256]u8 = undefined;
-    var extra: []const u8 = "";
-    if (hub.pipeline_gauges) |pg| {
-        const v = pg(hub.pipeline_ctx.?);
-        extra = std.fmt.bufPrint(&extra_buf, "# TYPE stream_pipeline_ticket gauge\n" ++
-            "stream_pipeline_ticket{{stage=\"submitted\"}} {d}\n" ++
-            "stream_pipeline_ticket{{stage=\"claimed\"}} {d}\n" ++
-            "stream_pipeline_ticket{{stage=\"emitted\"}} {d}\n", .{ v[0], v[1], v[2] }) catch "";
-    }
-    respondPartsMaybeHead(conn, "200 OK", "text/plain; version=0.0.4", out, extra, head);
+    respondPartsMaybeHead(conn, "200 OK", "text/plain; version=0.0.4", out, "", head);
 }
 
 /// adapter the xrpc api writes through (json / raw+etag / xrpc error)
diff --git a/src/main.zig b/src/main.zig
index 4100ee4..e258d31 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -1259,8 +1259,6 @@ pub fn main(init: std.process.Init.Minimal) !void {
     try pipe.start();
     defer pipe.stop();
     consumer.pipeline = &pipe;
-    hub.pipeline_ctx = &pipe;
-    hub.pipeline_gauges = &pipelineGauges;
     defer import_manager.deinit();
     import_manager_err_owned = false;
     import_ready.store(true, .release);
@@ -1345,12 +1343,6 @@ fn runConsumer(consumer: *ingest.Consumer, done: *std.atomic.Value(bool)) !void
     try consumer.run();
 }
 
-fn pipelineGauges(ctx: *anyopaque) [3]u64 {
-    const pipe: *pipeline_mod.Pipeline = @ptrCast(@alignCast(ctx));
-    const g = pipe.gauges();
-    return .{ g.submitted, g.claimed, g.emitted };
-}
-
 fn runWsServer(server: *websocket.Server(server_mod.Handler), listener: *Io.net.Server, context: *server_mod.ListenerContext) void {
     server.runIo(listener, context);
 }
diff --git a/tests/oracle.py b/tests/oracle.py
index 4a2aa2a..9174e0d 100644
--- a/tests/oracle.py
+++ b/tests/oracle.py
@@ -427,6 +427,64 @@ def run_backfill_metrics_receipt():
         shutil.rmtree(DATA, ignore_errors=True)
 
 
+def run_bootstrap_pipeline_metrics_receipt():
+    """Pipeline scheduler metrics must exist and move before serving ungates."""
+    shutil.rmtree(DATA, ignore_errors=True)
+    log = open("/tmp/oracle-bootstrap-pipeline-metrics.log", "w")
+    p = subprocess.Popen(ARGS + [
+        "--max-backfill-repos=100",
+        "--backfill-workers=1",
+    ], stdout=log, stderr=log)
+    try:
+        deadline = time.time() + 60
+        receipt = None
+        while time.time() < deadline:
+            if p.poll() is not None:
+                raise AssertionError(f"stream exited before bootstrap metrics receipt: rc={p.returncode}")
+            try:
+                samples = metrics_snapshot()
+            except AssertionError:
+                time.sleep(0.02)
+                continue
+            submitted = metric(samples, "stream_pipeline_ticket", stage="submitted")
+            if metric(samples, "jetstream_orchestrator_phase") == 1 and submitted > 0:
+                receipt = samples
+                break
+            time.sleep(0.02)
+        assert receipt is not None, "never observed an active bootstrap pipeline"
+
+        submitted = metric(receipt, "stream_pipeline_ticket", stage="submitted")
+        claimed = metric(receipt, "stream_pipeline_ticket", stage="claimed")
+        emitted = metric(receipt, "stream_pipeline_ticket", stage="emitted")
+        assert submitted >= claimed
+        assert submitted >= emitted
+        queued_gap = metric(
+            receipt, "stream_pipeline_stage_gap", boundary="submitted_to_claimed"
+        )
+        writer_gap = metric(
+            receipt, "stream_pipeline_stage_gap", boundary="claimed_to_emitted"
+        )
+        total_gap = metric(
+            receipt, "stream_pipeline_stage_gap", boundary="submitted_to_emitted"
+        )
+        assert queued_gap >= 0 and writer_gap >= 0
+        assert total_gap == submitted - emitted
+        assert queued_gap + writer_gap == total_gap
+        assert metric(receipt, "stream_pipeline_fatal") == 0
+        for stage in ("scheduler", "emit", "repair", "cursor"):
+            key = ("stream_pipeline_failures_total", (("stage", stage),))
+            assert key in receipt, f"missing registered pipeline failure stage {stage}"
+            assert metric(receipt, "stream_pipeline_failures_total", stage=stage) == 0
+    finally:
+        p.send_signal(signal.SIGTERM)
+        try:
+            p.wait(timeout=15)
+        except subprocess.TimeoutExpired:
+            p.kill()
+        log.close()
+        shutil.rmtree(DATA, ignore_errors=True)
+
+
 def main():
     try:
         with urllib.request.urlopen(SIM + "/xrpc/com.atproto.sync.listRepos?limit=1", timeout=5) as r:
@@ -434,6 +492,10 @@ def main():
     except OSError:
         sys.exit("simulator not reachable on :7777 — run `just simulator` first")
 
+    t0 = time.time()
+    run_bootstrap_pipeline_metrics_receipt()
+    print(f"ok bootstrap pipeline metrics ({time.time() - t0:.1f}s)")
+
     t0 = time.time()
     run_backfill_metrics_receipt()
     print(f"ok backfill metrics ({time.time() - t0:.1f}s)")
diff --git a/tests/test_dashboard_contract.py b/tests/test_dashboard_contract.py
index 46af914..229f55b 100644
--- a/tests/test_dashboard_contract.py
+++ b/tests/test_dashboard_contract.py
@@ -34,10 +34,30 @@ class DashboardContractTest(unittest.TestCase):
         upstream = json.loads(check_dashboard.UPSTREAM_DASHBOARD.read_bytes())
         stream = json.loads(check_dashboard.DEFAULT_DASHBOARD.read_bytes())
         runtime_ids = {69, 70, 71, 72, 73, 74, 75}
+        stream_only_ids = {76, 77, 78, 79}
         upstream_panels = {panel["id"]: panel for panel in upstream["panels"] if panel["id"] not in runtime_ids}
-        stream_panels = {panel["id"]: panel for panel in stream["panels"] if panel["id"] not in runtime_ids}
+        stream_panels = {
+            panel["id"]: panel
+            for panel in stream["panels"]
+            if panel["id"] not in runtime_ids | stream_only_ids
+        }
         self.assertEqual(upstream_panels, stream_panels)
 
+    def test_stream_pipeline_diagnostics_are_explicit_additions(self):
+        stream = json.loads(check_dashboard.DEFAULT_DASHBOARD.read_bytes())
+        panels = {panel["id"]: panel for panel in stream["panels"]}
+        self.assertEqual("Stream pipeline diagnostics", panels[76]["title"])
+        self.assertEqual("Pipeline tickets", panels[77]["title"])
+        self.assertEqual("Pipeline stage gaps", panels[78]["title"])
+        self.assertEqual("Pipeline terminal failures", panels[79]["title"])
+        required = check_dashboard.dashboard_families(check_dashboard.DEFAULT_DASHBOARD.read_bytes())
+        self.assertTrue({
+            "stream_pipeline_ticket",
+            "stream_pipeline_stage_gap",
+            "stream_pipeline_fatal",
+            "stream_pipeline_failures_total",
+        }.issubset(required))
+
     def test_histogram_type_expands_queryable_children(self):
         present = check_dashboard.scrape_families(
             "# TYPE request_seconds histogram\nrequest_seconds_bucket{le=\"1\"} 2\n"
diff --git a/tools/adapt_dashboard.py b/tools/adapt_dashboard.py
index c12beff..fd22c07 100644
--- a/tools/adapt_dashboard.py
+++ b/tools/adapt_dashboard.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Deterministically adapt only Jetstream's Go runtime row for Stream."""
+"""Deterministically adapt Jetstream's dashboard for the Stream runtime."""
 
 from __future__ import annotations
 
@@ -22,6 +22,20 @@ def target(template: dict, expr: str, legend: str, ref_id: str) -> dict:
     return value
 
 
+def pipeline_panel(template: dict, panel_id: int, title: str, description: str, x: int) -> dict:
+    panel = copy.deepcopy(template)
+    panel.update(
+        id=panel_id,
+        title=title,
+        description=description,
+        gridPos={"h": 8, "w": 8, "x": x, "y": 171},
+    )
+    panel["fieldConfig"]["defaults"]["unit"] = "short"
+    panel["fieldConfig"]["defaults"]["min"] = 0
+    panel["fieldConfig"]["overrides"] = []
+    return panel
+
+
 def adapt(raw: bytes) -> bytes:
     digest = hashlib.sha256(raw).hexdigest()
     if digest != SOURCE_SHA256:
@@ -88,12 +102,88 @@ def adapt(raw: bytes) -> bytes:
     ]
     panels[74]["fieldConfig"]["defaults"]["unit"] = "ops"
 
+    # Upstream has no scheduler between its websocket reader and archive
+    # writer. Stream does, so retain every upstream protocol/lifecycle panel
+    # byte-for-byte and append a clearly Stream-specific diagnostics row.
+    diagnostics_row = copy.deepcopy(panels[75])
+    diagnostics_row.update(
+        id=76,
+        title="Stream pipeline diagnostics",
+        description="Live during bootstrap and steady state; gaps and terminal failures are never inferred from archive throughput.",
+        gridPos={"h": 1, "w": 24, "x": 0, "y": 170},
+    )
+
+    tickets = pipeline_panel(
+        panels[69],
+        77,
+        "Pipeline tickets",
+        "Cumulative scheduler boundaries. Emitted is the historical name for every settled ticket, including explicit drops and verifier rejections.",
+        0,
+    )
+    tickets["targets"] = [
+        target(
+            panels[69]["targets"][0],
+            f'max(stream_pipeline_ticket{selector}) by (stage)',
+            "{{stage}}",
+            "A",
+        )
+    ]
+
+    gaps = pipeline_panel(
+        panels[69],
+        78,
+        "Pipeline stage gaps",
+        "Current unsettled work split between the unclaimed verifier queue and claimed work awaiting settlement. Their sum is submitted-to-emitted.",
+        8,
+    )
+    gaps["targets"] = [
+        target(
+            panels[69]["targets"][0],
+            f'max(stream_pipeline_stage_gap{selector}) by (boundary)',
+            "{{boundary}}",
+            "A",
+        )
+    ]
+
+    failures = pipeline_panel(
+        panels[69],
+        79,
+        "Pipeline terminal failures",
+        "Any non-zero value is deployment-significant. The gauge marks the current pipeline fatal; the counters retain its first failing stage.",
+        16,
+    )
+    failures["targets"] = [
+        target(
+            panels[69]["targets"][0],
+            f'max(stream_pipeline_fatal{selector})',
+            "fatal",
+            "A",
+        ),
+        target(
+            panels[69]["targets"][0],
+            f'sum(stream_pipeline_failures_total{selector}) by (stage)',
+            "{{stage}} failures",
+            "B",
+        ),
+    ]
+    failures["fieldConfig"]["defaults"]["thresholds"] = {
+        "mode": "absolute",
+        "steps": [
+            {"color": "green", "value": None},
+            {"color": "red", "value": 1},
+        ],
+    }
+    failures["fieldConfig"]["defaults"]["custom"]["thresholdsStyle"] = {"mode": "line"}
+
+    dashboard["panels"].extend([diagnostics_row, tickets, gaps, failures])
+
     dashboard["title"] = "Stream / Jetstream V2"
     dashboard["uid"] = "stream-jetstream-v2"
     dashboard["tags"] = ["stream", "jetstream", "atproto"]
     dashboard["description"] = (
         "Stream operational dashboard adapted from the checksum-pinned Jetstream V2 dashboard. "
-        "Protocol and lifecycle rows are unchanged; only the Go runtime row uses Stream process metrics."
+        "Protocol and lifecycle rows are unchanged; the runtime row uses Stream process metrics, "
+        "and a final Stream-only row exposes its live scheduler during bootstrap and steady state."
     )
     return (json.dumps(dashboard, indent=2, ensure_ascii=False) + "\n").encode()
 
diff --git a/tools/check_dashboard.py b/tools/check_dashboard.py
index d045c1d..b4488a4 100755
--- a/tools/check_dashboard.py
+++ b/tools/check_dashboard.py
@@ -17,7 +17,7 @@ UPSTREAM_DASHBOARD = ROOT / "deploy/grafana/jetstream-upstream.json"
 DEFAULT_DASHBOARD = ROOT / "deploy/grafana/stream.json"
 EXPECTED_SHA256 = {
     UPSTREAM_DASHBOARD.resolve(): "ef6033c182d8c3f9d88af7d91e61126724efbe72e6b940984c9de12e9530cd91",
-    DEFAULT_DASHBOARD.resolve(): "e27537b5a951bc1b5a6133aaef068b7b4b1eb2abff2e2f283722697cf6fe7017",
+    DEFAULT_DASHBOARD.resolve(): "e64377c02d50e9cbe0e2c1c78ede4a30aa61bdca751ac9783f63f9c08f3d3951",
 }
 METRIC_RE = re.compile(r"\b(?:jetstream|go|process|stream)_[a-zA-Z_:][a-zA-Z0-9_:]*")
 TYPE_RE = re.compile(r"^# TYPE ([a-zA-Z_:][a-zA-Z0-9_:]*) (\w+)$")