o11y: an alert that reached nobody must not answer 200, and the data plane gets rules

Arrival and delivery were the same word. This endpoint logged PAGE-DELIVERED,
answered 200 `ok`, and then failed to send — because the egress ran detached in
a goroutine the response never waited for, so the answer was written before the
send was attempted and could not possibly have been about it. Alertmanager
recorded Notify success over a silent pager.

Split the two facts. ALERT-RECEIVED is arrival; ALERT-DELIVERED /
ALERT-UNDELIVERED is egress. Delivery is synchronous under an 8s budget and THE
STATUS CODE REPORTS DELIVERY: nothing carried it means 503, so Alertmanager
retries and counts it. No egress configured is a failure, not a no-op — that
silent early return is how a whole deployment could page nobody forever.

Egress is a chain: the org's KMS-custodied Slack bot token first, then a plain
POST to CLOUD_ALERTS_WEBHOOK_URL, which needs no integrations peer, no org and
no connected workspace — it works in exactly the state that silenced the first.
A failure is recorded even when a later egress succeeds, so a broken egress
cannot hide behind a working one.

Then the reason no rule could have caught any of this: the data plane emitted
nothing. 904 metric names in the store and not one described a row moving.
hanzo_http_requests_total existed in code with ZERO callers, so /v1/event's 5xx
rate was unalertable while the Sentry envelope 503'd for a day. Wire it into the
one middleware that already has path, status and validated org in hand, and add
the missing measurements: rows landed per warehouse table, ingest admission
outcomes, plane sockets bound, cron fires against the engine's own promise,
memory against GOMEMLIMIT, and alert egress itself.

Every counter is SEEDED AT ZERO at boot. A counter first touched by its first
event has no series until that event, so an ingest path that never runs looks
exactly like one that was never built — which is how span ingest stayed dead for
four and a half months without a single rule being able to notice. Seeding is
what makes silence measurable.

Tests pin the new contract: undeliverable answers 503, no egress is a failure,
the fallback carries it while the Slack failure is still recorded, the chain
stops at the first success, delivery is synchronous, and PAGE-DELIVERED is gone.

Co-authored-by: Hanzo Dev <dev@hanzo.ai>

apps/analytics/event.go

@@ -363,6 +363,12 @@ func ingestDecoded(c *zip.Ctx, org, source string, evs []CaptureEvent, dropped i
 		return err
 	}
 	res.Dropped += dropped
+	// The admission receipt, counted. Every lane funnels through here, and the
+	// pair is reported together on purpose: the answerable question is a RATIO.
+	// "8,000 items were dropped" needs a second series before it means anything;
+	// "88% of what was offered was dropped" is an outage on its own — and that
+	// exact loss ran unnoticed because neither number left the response body.
+	cloud.ObserveIngest(source, res.Accepted, res.Dropped)
 	return c.JSON(http.StatusOK, res)
 }
 

apps/analytics/warehouse.go

@@ -294,7 +294,14 @@ func (w writer) write(ctx context.Context, m message) error {
 	if strings.TrimSpace(m.Org) == "" {
 		return fmt.Errorf("refusing an unattributed %s fact (id %q)", w.signal, m.ID)
 	}
-	return warehouseExec(ctx, w.statement(), w.args(m)...)
+	if err := warehouseExec(ctx, w.statement(), w.args(m)...); err != nil {
+		return err
+	}
+	// The second writer into event.* — the bus drain. It counts on the same
+	// series as the o11y plane sink so a table's row rate is the whole truth
+	// about that table, whichever path is carrying it.
+	cloud.ObserveRows(w.signal.table(), 1)
+	return nil
 }
 
 // drain owns the consumers — one per writer, all on the one bus connection. It is the

apps/cron/cron.go

@@ -146,6 +146,12 @@ func start(ctx context.Context, log luxlog.Logger) {
 		log.Error("cron: reconcile self-schedule", "err", err)
 		return
 	}
+	// Publish what the engine believes about its own schedules. Without this the
+	// durable engine's record of a missed fire never leaves the process, which
+	// is how moving off CronJobs quietly cost the fleet the one cron signal it
+	// had.
+	publishScheduleMetrics(view)
+
 	log.Info("platform cron live on the durable tasks engine",
 		"org", org(), "namespace", namespace, "queue", taskQueue, "reconcile", reconcileEvery)
 }

apps/cron/metrics.go

@@ -0,0 +1,144 @@
+// Copyright 2026 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// What the durable cron engine believes, published for collection.
+//
+// Moving the platform's cron off Kubernetes CronJobs took the signal with it.
+// A CronJob that stops firing leaves kube_cronjob_status_last_schedule_time
+// behind and kube-state-metrics keeps publishing it, so "this has not run"
+// stayed answerable by accident. The durable engine keeps a better record —
+// and kept it entirely to itself. Nothing left the process, so a schedule that
+// silently stopped firing was invisible to every rule in the estate.
+//
+// THE MISS IS MEASURED AGAINST THE ENGINE'S OWN PROMISE, not against a cron
+// expression parsed a second time here. The engine publishes NextActionTime:
+// the instant it intends to fire, re-anchored on every fire. So a miss is
+// simply `now` well past a NextActionTime that never moved — true for a
+// stalled sweeper, a wedged worker, an unregistered queue and a crashed
+// process alike, without this file knowing what "*/5 * * * *" means. Re-deriving
+// the schedule here would be a second implementation of the thing being
+// checked, and it would agree with the engine exactly when it did not matter.
+//
+// These are OBSERVABLE gauges: schedule state is a fact that is true at
+// collection time, not an event to be tracked.
+
+package cron
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	tasksengine "github.com/hanzoai/tasks/pkg/tasks"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+// meterName matches the process meter so cron's series sit beside cloud's own.
+const meterName = "github.com/hanzoai/cloud"
+
+// scheduleSource is the engine view the gauges read. It is set once cron is
+// live; until then the callbacks observe nothing, which is correct — a process
+// that has not started cron must not claim its schedules are late.
+var (
+	sourceMu     sync.RWMutex
+	scheduleView *tasksengine.View
+	metricsOnce  sync.Once
+)
+
+// publishScheduleMetrics points the gauges at a live engine view and registers
+// them once. Called from start() after the worker is up.
+func publishScheduleMetrics(v tasksengine.View) {
+	sourceMu.Lock()
+	scheduleView = &v
+	sourceMu.Unlock()
+	metricsOnce.Do(registerScheduleGauges)
+}
+
+// view returns the live engine view, or nil before cron is up.
+func view() *tasksengine.View {
+	sourceMu.RLock()
+	defer sourceMu.RUnlock()
+	return scheduleView
+}
+
+func registerScheduleGauges() {
+	m := otel.Meter(meterName)
+
+	next, nerr := m.Float64ObservableGauge("hanzo_cron_next_action_timestamp_seconds",
+		metric.WithDescription("Unix time the engine intends to fire this schedule next. Now well past it means a MISS."))
+	last, lerr := m.Float64ObservableGauge("hanzo_cron_last_action_timestamp_seconds",
+		metric.WithDescription("Unix time this schedule last fired (a start, not an outcome)."))
+	fires, ferr := m.Int64ObservableGauge("hanzo_cron_action_count",
+		metric.WithDescription("Fires this schedule has started since it was created."))
+	fails, xerr := m.Int64ObservableGauge("hanzo_cron_consecutive_failures",
+		metric.WithDescription("Consecutive failed runs for this schedule. 1 is a blip; a streak is the incident."))
+	if nerr != nil || lerr != nil || ferr != nil || xerr != nil {
+		return
+	}
+
+	_, _ = m.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+		v := view()
+		if v == nil {
+			return nil
+		}
+		schedules, err := v.ListSchedules(namespace)
+		if err != nil {
+			// A view that cannot be read is not a fleet of on-time schedules.
+			// Observing nothing lets the series go stale, which the staleness
+			// rule reports — far better than observing a reassuring zero.
+			return nil
+		}
+		for _, s := range schedules {
+			attrs := metric.WithAttributes(attribute.String("entry", s.ScheduleId))
+			if t, ok := stamp(s.Info.NextActionTime); ok {
+				o.ObserveFloat64(next, t, attrs)
+			}
+			if t, ok := stamp(s.Info.UpdateTime); ok {
+				o.ObserveFloat64(last, t, attrs)
+			}
+			o.ObserveInt64(fires, s.Info.ActionCount, attrs)
+		}
+		// Failure streaks are keyed by ScheduleId too, so a schedule that fires
+		// on time and fails every time is a different alert from one that stopped
+		// firing — two failure modes a single "last run" number cannot separate.
+		streaks, err := v.FailureStreaks(namespace)
+		if err != nil {
+			return nil
+		}
+		for _, f := range streaks {
+			if f.ScheduleId == "" {
+				continue
+			}
+			o.ObserveInt64(fails, f.ConsecutiveFailures,
+				metric.WithAttributes(attribute.String("entry", f.ScheduleId)))
+		}
+		return nil
+	}, next, last, fires, fails)
+}
+
+// stamp parses an engine RFC3339 timestamp into unix seconds. An unparseable or
+// empty value is reported as absent rather than as zero: zero is 1970, which
+// every staleness rule would read as fifty years late.
+func stamp(s string) (float64, bool) {
+	if s == "" {
+		return 0, false
+	}
+	t, err := time.Parse(time.RFC3339, s)
+	if err != nil {
+		return 0, false
+	}
+	return float64(t.Unix()), true
+}

apps/o11y/alerts.go

@@ -12,37 +12,69 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Alertmanager webhook receiver — the page-delivery receipt.
+// Alertmanager webhook receiver — arrival and delivery, told apart.
 //
-// Alertmanager can tell you it dispatched a notification. It cannot tell you
-// anything landed. This endpoint is the far side of that hop: every delivery
-// prints one PAGE-DELIVERED line to the process log and joins a bounded ring
-// that GET /v1/o11y/alerts/last replays. When somebody asks "did the page
-// actually fire?", that ring is the answer, and it is an answer no amount of
-// reading Alertmanager's own state can produce.
+// An alert carries two distinct facts and this file used to braid them into
+// one word. A notification ARRIVED here (Alertmanager made the call), and a
+// notification was DELIVERED to a human (something carried it out of this
+// process). They are not the same fact, they fail independently, and for
+// months this endpoint reported the first while everyone read it as the
+// second:
 //
-// This is the whole of the former standalone `alert-sink` Deployment (a
-// stdlib-only Python script in a ConfigMap on a stock python:3.12-alpine
-// image). It is 30 lines of behaviour that needed a pod, a Service, a
-// ConfigMap and an operator CR to exist. It belongs on the observability
-// plane that already runs, so it lives here.
+//	[o11y] PAGE-DELIVERED  … alert=NodeMemoryCommittedCritical   ← arrival
+//	[o11y] PAGE-SLACK-FAILED … err=slack not connected for org   ← the truth
 //
-// AND IT PAGES. A receipt nobody reads is not an alert, and until this landed
-// nothing reached a human: Alertmanager's slack_configs pointed at a secret
-// holding the receipt sink's own URL, so 439 "slack" notifications were
-// delivered into a log. The fix is not a second Slack credential — an incoming
-// webhook would be a second secret outside KMS and a second egress beside the
-// one the product already uses. It is the app that is already installed: this
-// receiver forwards each firing alert through cloud.SlackSend (the shared
-// egress the integrations subsystem installs), which posts with the org's
-// KMS-custodied bot token (the ONE Slack egress, shared
-// with channels and automations). One credential, one egress, one receipt.
+// Alertmanager logged `Notify success`, this endpoint answered 200 `ok`, and
+// the page reached nobody. Three green lights over a silent pager, because the
+// egress ran DETACHED in a goroutine the response never waited for. The
+// request was answered before the send was attempted, so the answer could not
+// possibly have been about it.
+//
+// So the two facts now have two names and two records:
+//
+//	ALERT-RECEIVED    — this process took the call. Always true, always logged,
+//	                    joins the replay ring. It is a receipt, nothing more.
+//	ALERT-DELIVERED   — an egress accepted it, and which one.
+//	ALERT-UNDELIVERED — no egress accepted it, and why. Answered 503.
+//
+// and THE STATUS CODE REPORTS DELIVERY, NOT ARRIVAL. Nothing delivered means
+// non-2xx, which is the only sentence Alertmanager understands: it retries,
+// and it counts the failure in alertmanager_notifications_failed_total, which
+// is itself alertable. An alert path that cannot reach a human must fail
+// loudly at the protocol level, because the one thing it must never do is look
+// identical to a working one.
+//
+// EGRESS IS A CHAIN, not a call. Ways out are tried in order, first success
+// wins:
+//
+//  1. slack   — the org's KMS-custodied bot token via the integrations peer.
+//     The ONE product Slack egress (shared with channels and automations), so
+//     no second credential. It requires the workspace to be CONNECTED, which
+//     is an owner action and therefore something the alert path must never
+//     assume.
+//  2. webhook — a plain POST of {"text": …} to CLOUD_ALERTS_WEBHOOK_URL. No
+//     integrations peer, no org, no connected workspace, no KMS: it works
+//     precisely in the state that silenced everything above. That is its job.
+//
+// Degraded is not healthy: when Slack fails and the webhook carries it, the
+// delivery is real (200) but the Slack failure is still recorded and still
+// counted, so a broken egress cannot hide behind a working one.
+//
+// ⚠️ The chain runs INSIDE cloud, so an alert about cloud being down routes
+// through the thing that is down — observed twice as `dial tcp
+// 10.124.43.30:8000: connect: connection refused`, recovering on retry 8. That
+// hop cannot be fixed from in here; it is fixed in Alertmanager, which posts
+// every Hanzo receiver to BOTH this endpoint and the same webhook URL directly
+// (universe: infra/k8s/monitoring/alertmanager-config.yaml). This process is
+// the rich path; the direct one is the path that survives this process.
 
 package o11y
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -61,7 +93,7 @@ import (
 // bound is the point: an unbounded receipt log is a memory leak with a nice name.
 const recentMax = 200
 
-// alertRing is the process-local delivery ring. Process-local is correct for a
+// alertRing is the process-local record ring. Process-local is correct for a
 // receipt — it answers "did THIS process take the call", and a receipt that
 // survived its process would be a claim about something nobody observed.
 type alertRing struct {
@@ -110,34 +142,40 @@ type alert struct {
 // lift from a raw handler, so their prose is declared here, beside the wire fact.
 func init() {
 	openapi.Describe("/v1/o11y/alerts/last", http.MethodGet,
-		"Replay the page-delivery receipts this process took",
+		"Replay the alert records this process took",
 		"Answers the most recent Alertmanager deliveries THIS process received, as plain "+
-			"text — one greppable `PAGE-DELIVERED` line per alert, newest last, so piping to "+
-			"`tail` reads in arrival order. `(none)` when nothing has landed.\n\n"+
-			"It answers the question Alertmanager cannot: Alertmanager can tell you it "+
-			"dispatched a notification, never that anything received it. This ring is the far "+
-			"side of that hop, and it is the only record that a page actually arrived.\n\n"+
+			"text — one greppable `ALERT-RECEIVED` line per alert, followed by the "+
+			"`ALERT-DELIVERED` / `ALERT-UNDELIVERED` outcome of carrying it out of the "+
+			"process, newest last, so piping to `tail` reads in arrival order. `(none)` when "+
+			"nothing has landed.\n\n"+
+			"Arrival and delivery are separate lines because they are separate facts that "+
+			"fail independently. Alertmanager can tell you it dispatched a notification, never "+
+			"that anything received it; this process taking the call says nothing about whether "+
+			"a human was reached. Reading only the first as if it were the second is how a "+
+			"pager stays silent for months behind a log where everything looks fine.\n\n"+
 			"The ring is PROCESS-LOCAL and bounded to the last 200 lines. Both are the point: a "+
-			"receipt that outlived the process that took the call would be a claim about "+
-			"something nobody observed, and an unbounded receipt log is a memory leak with a "+
-			"nice name. A restart empties it.")
+			"record that outlived the process that took the call would be a claim about "+
+			"something nobody observed, and an unbounded log is a memory leak with a nice "+
+			"name. A restart empties it.")
 	openapi.Describe("/v1/o11y/alerts/:receiver", http.MethodPost,
-		"Take an Alertmanager notification and page Slack",
-		"Records one Alertmanager webhook delivery and pages the on-call. Each alert in the "+
-			"payload prints a `PAGE-DELIVERED` line to the process log and joins the ring the "+
-			"receipt replay serves, then the batch is posted to Slack with the org's "+
-			"KMS-custodied bot token — the ONE Slack egress the product already has, not a "+
-			"second webhook credential. Resolved notifications page too: \"it recovered\" is the "+
-			"half of an incident people are actually waiting for.\n\n"+
-			"It ALWAYS answers 200 with the body `ok`, and a body that will not parse is "+
-			"recorded with empty fields rather than rejected. Alertmanager retries on any other "+
-			"status, so a receipt that pushes back changes the thing it is measuring, and a 400 "+
-			"on a malformed payload would make it retry forever — the delivery still happened, "+
-			"which is the fact being recorded.\n\n"+
+		"Take an Alertmanager notification and page a human",
+		"Records one Alertmanager webhook delivery and pages the on-call. Each alert prints an "+
+			"`ALERT-RECEIVED` line and joins the replay ring, then the batch is carried out of "+
+			"the process by the egress chain: the org's KMS-custodied Slack bot token first "+
+			"(the ONE product Slack egress, not a second webhook credential), falling back to a "+
+			"plain POST to `CLOUD_ALERTS_WEBHOOK_URL` — which needs no Slack connection and so "+
+			"works in exactly the state that silences the first. Resolved notifications page "+
+			"too: \"it recovered\" is the half of an incident people are actually waiting for.\n\n"+
+			"THE STATUS CODE REPORTS DELIVERY, NOT ARRIVAL. 200 `ok` means an egress accepted "+
+			"the batch. If none did — including when none is configured at all — it answers "+
+			"**503** naming the failure, so Alertmanager retries and counts it in "+
+			"`alertmanager_notifications_failed_total`. An alert nobody could be told about "+
+			"must never answer the same way as one that was delivered.\n\n"+
+			"A body that will not parse is still recorded (with empty fields) rather than "+
+			"rejected: the delivery happened, which is the fact being recorded, and a 400 would "+
+			"make Alertmanager retry a malformed payload forever.\n\n"+
 			"The receiver segment is Alertmanager's own receiver name, a parameter rather than a "+
-			"hand-listed route because the receiver set is config, not code. Paging is detached "+
-			"and fail-soft: with no channel configured nothing is posted and the receipt still "+
-			"lands, and a Slack failure prints its own line instead of failing the request.")
+			"hand-listed route because the receiver set is config, not code.")
 }
 
 // mountAlerts registers the receiver. Called from MountO11y BEFORE the
@@ -152,78 +190,205 @@ func mountAlerts(a cloud.Router) {
 	g.Post("/:receiver", receive)
 }
 
-// receive records one Alertmanager notification and pages Slack. Always 200
-// with body "ok": Alertmanager retries on any other status, and a receipt that
-// pushes back is a receipt that changes the thing it is measuring.
+// receive records one Alertmanager notification, carries it to a human, and
+// answers with the result of the CARRYING — not of the recording.
+//
+// Delivery is SYNCHRONOUS. The previous version sent in a detached goroutine,
+// which made 200 structurally incapable of meaning anything: the response was
+// written before the send was tried. A bounded wait is what makes the status
+// code a fact rather than a hope.
 func receive(c *zip.Ctx) error {
 	var p webhook
-	// A body that will not parse still proves delivery, so it is logged with
+	// A body that will not parse still proves delivery, so it is recorded with
 	// empty fields rather than rejected.
 	_ = json.Unmarshal(c.Body(), &p)
 
 	as := alerts(&p)
 	for _, a := range as {
-		line := receipt(c.Path(), &p, a)
-		fmt.Println(line)
-		recent.add(line)
+		record(receipt(c.Path(), &p, a))
 	}
-	page(&p, as)
+
+	ctx, cancel := context.WithTimeout(c.Context(), egressBudget)
+	defer cancel()
+	via, failures := deliver(ctx, &p, as)
+
+	// A failure is recorded even when a later egress succeeded: an egress that
+	// hides behind a working one is the failure mode this file exists to end.
+	for _, f := range failures {
+		cloud.ObserveAlertDelivery(f.egress, "failed")
+		record(fmt.Sprintf("ALERT-EGRESS-FAILED egress=%s receiver=%s err=%v",
+			f.egress, or(p.Receiver, "?"), f.err))
+	}
+
+	if via == "" {
+		why := reason(failures)
+		record(fmt.Sprintf("ALERT-UNDELIVERED receiver=%s alerts=%d :: %s",
+			or(p.Receiver, "?"), len(as), why))
+		// 503, not 200. Alertmanager retries this and counts it, which is the
+		// only way the outside world can learn that paging is broken.
+		return c.String(http.StatusServiceUnavailable, "undelivered: "+why)
+	}
+	cloud.ObserveAlertDelivery(via, "delivered")
+	record(fmt.Sprintf("ALERT-DELIVERED via=%s receiver=%s alerts=%d",
+		via, or(p.Receiver, "?"), len(as)))
 	return c.String(http.StatusOK, "ok")
 }
 
-// Slack paging knobs. The channel is the only thing that MUST be configured —
-// no channel, no paging (and the receipt still lands, so silence here is never
-// silence everywhere). The org owns the Slack connection whose bot token KMS
-// custodies; it defaults to the platform tenant.
+// record prints one line to the process log and joins the replay ring. Both or
+// neither — a line an operator can grep but not replay (or the reverse) is a
+// record that disagrees with itself.
+func record(line string) {
+	fmt.Println(line)
+	recent.add(line)
+}
+
+// Egress knobs.
 const (
 	alertsSlackChannelEnv = "CLOUD_ALERTS_SLACK_CHANNEL"
 	alertsSlackOrgEnv     = "CLOUD_ALERTS_SLACK_ORG"
-	defaultAlertsOrg      = "hanzo"
-	peerIntegrations      = "integrations" // the plugin that holds the bot token
-	slackPageTimeout      = 10 * time.Second
+	// alertsWebhookEnv is the fallback egress: any URL that accepts a POST of
+	// {"text": …}. It deliberately has NO dependency on the integrations peer,
+	// an org, or a connected workspace — it is the egress for the state where
+	// those are the problem.
+	alertsWebhookEnv = "CLOUD_ALERTS_WEBHOOK_URL"
+	defaultAlertsOrg = "hanzo"
+	peerIntegrations = "integrations" // the plugin that holds the bot token
+	// egressBudget bounds the whole chain, not one hop. Alertmanager is waiting
+	// on this request; a page that has not left in eight seconds is better
+	// reported as undelivered (and retried) than waited on.
+	egressBudget = 8 * time.Second
 )
 
-// page posts the notification to Slack through the ONE egress. It is
-// DETACHED and fail-soft by construction: Alertmanager is waiting on this
-// request, and an alert path that can block or fail on a third party is an
-// alert path that goes quiet exactly when the third party is having the
-// outage. Resolved notifications page too — "it recovered" is the half of an
-// incident people actually wait for.
-func page(p *webhook, as []alert) {
-	channel := strings.TrimSpace(os.Getenv(alertsSlackChannelEnv))
-	if channel == "" || len(as) == 0 {
-		return
-	}
-	org := strings.TrimSpace(os.Getenv(alertsSlackOrgEnv))
-	if org == "" {
-		org = defaultAlertsOrg
-	}
-	text := slackText(p, as)
-	go func() {
-		defer func() { _ = recover() }()
-		// ZAP over the unix socket to the integrations PROCESS, which owns the
-		// org's bot token (a plugin is a process; a package global here reads a
-		// nil peer copy — the "integrations: not mounted" failure). cloud.Ask
-		// dials the socket and wakes integrations if it is asleep, exactly as
-		// x402 asks commerce to move money. cloud.For stamps the org so
-		// integrations' handler reads it as the caller's, never an argument.
-		ctx, cancel := context.WithTimeout(cloud.For(context.Background(), org), slackPageTimeout)
-		defer cancel()
-		_, err := cloud.Ask[plane.SlackSendIn, struct{}](ctx, peerIntegrations, plane.IntegrationsSlackSend,
-			&plane.SlackSendIn{Channel: channel, Text: text})
-		if err != nil {
-			// One line, in the same log as the receipts: a page that could not
-			// be sent is itself an operational fact, and the receipt above
-			// already proved the alert arrived.
-			fmt.Printf("PAGE-SLACK-FAILED org=%s channel=%s receiver=%s err=%v\n",
-				org, channel, or(p.Receiver, "?"), err)
-		}
-	}()
+// egress is one way out of this process to a human.
+//
+// A chain of these — rather than one hard-wired call — is what makes "Slack is
+// not connected" a DEGRADED state instead of a silent one. Adding a way out is
+// adding an element; the delivery contract above it does not change.
+type egress struct {
+	name string
+	send func(ctx context.Context, text string) error
 }
 
-// slackText renders the notification as one Slack message: a firing/resolved
-// headline, then one line per alert carrying the fields an on-call reads first
-// (name, severity, instance, summary). Bounded — a storm must not post a
+// failure is one egress's refusal, kept with its name so the record says which
+// way out was tried and the metric can be labelled by it.
+type failure struct {
+	egress string
+	err    error
+}
+
+// egressChain is what deliver walks. A var so tests can substitute a
+// deterministic chain; production rebuilds it from the environment on every
+// call, so connecting an egress does not need a restart.
+var egressChain = configuredEgresses
+
+// configuredEgresses returns the ways out, in preference order. Each is present
+// only when it is configured — an egress that cannot be attempted must not be
+// counted as one that was.
+func configuredEgresses() []egress {
+	var out []egress
+	if channel := strings.TrimSpace(os.Getenv(alertsSlackChannelEnv)); channel != "" {
+		org := strings.TrimSpace(os.Getenv(alertsSlackOrgEnv))
+		if org == "" {
+			org = defaultAlertsOrg
+		}
+		out = append(out, egress{name: "slack", send: func(ctx context.Context, text string) error {
+			return slackSend(ctx, org, channel, text)
+		}})
+	}
+	if url := strings.TrimSpace(os.Getenv(alertsWebhookEnv)); url != "" {
+		out = append(out, egress{name: "webhook", send: func(ctx context.Context, text string) error {
+			return webhookSend(ctx, url, text)
+		}})
+	}
+	return out
+}
+
+// deliver walks the chain and stops at the first egress that accepts the batch.
+// It returns the name of the one that carried it ("" if none did) and every
+// failure along the way.
+//
+// No egress configured is a FAILURE, not a no-op. "Nobody can be told" is the
+// state this whole file exists to make loud, and returning quietly when nothing
+// was configured is how it stayed quiet.
+func deliver(ctx context.Context, p *webhook, as []alert) (via string, failures []failure) {
+	chain := egressChain()
+	if len(chain) == 0 {
+		return "", []failure{{egress: "none", err: errors.New("no alert egress configured: set " +
+			alertsSlackChannelEnv + " or " + alertsWebhookEnv)}}
+	}
+	text := slackText(p, as)
+	for _, e := range chain {
+		if err := e.send(ctx, text); err != nil {
+			failures = append(failures, failure{egress: e.name, err: err})
+			continue
+		}
+		return e.name, failures
+	}
+	return "", failures
+}
+
+// reason renders the failures as one sentence for the 503 body and the
+// undelivered line — whoever reads either must not need a second lookup to
+// learn which way out broke, and how.
+func reason(failures []failure) string {
+	if len(failures) == 0 {
+		return "no egress attempted"
+	}
+	parts := make([]string, 0, len(failures))
+	for _, f := range failures {
+		parts = append(parts, fmt.Sprintf("%s: %v", f.egress, f.err))
+	}
+	return strings.Join(parts, "; ")
+}
+
+// slackSend posts through the ONE product Slack egress: ZAP over the unix
+// socket to the integrations PROCESS, which owns the org's bot token (a plugin
+// is a process; a package global here reads a nil peer copy — the
+// "integrations: not mounted" failure). cloud.Ask dials the socket and wakes
+// integrations if it is asleep, exactly as x402 asks commerce to move money.
+// cloud.For stamps the org so integrations' handler reads it as the caller's,
+// never an argument.
+func slackSend(ctx context.Context, org, channel, text string) error {
+	_, err := cloud.Ask[plane.SlackSendIn, struct{}](cloud.For(ctx, org), peerIntegrations,
+		plane.IntegrationsSlackSend, &plane.SlackSendIn{Channel: channel, Text: text})
+	return err
+}
+
+// webhookClient is shared: one connection pool for the fallback egress. Its
+// timeout is a backstop only — the caller's context carries the real budget.
+var webhookClient = &http.Client{Timeout: egressBudget}
+
+// webhookSend posts the page as {"text": …} — the shape a Slack incoming
+// webhook takes, which is also the shape most generic receivers take, so the
+// URL can be whatever the owner actually has without this code learning a
+// second format.
+//
+// A non-2xx is an error. This is the FALLBACK: if it quietly swallowed a
+// failure there would be nothing left underneath to notice.
+func webhookSend(ctx context.Context, url, text string) error {
+	body, err := json.Marshal(map[string]string{"text": text})
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := webhookClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode < 200 || resp.StatusCode > 299 {
+		return fmt.Errorf("webhook answered %s", resp.Status)
+	}
+	return nil
+}
+
+// slackText renders the notification as one page: a firing/resolved headline,
+// then one line per alert carrying the fields an on-call reads first (name,
+// severity, instance, summary). Bounded — a storm must not post a
 // thousand-line wall — with the overflow counted rather than dropped silently.
 func slackText(p *webhook, as []alert) string {
 	const maxLines = 20
@@ -267,11 +432,14 @@ func alerts(p *webhook) []alert {
 	return []alert{{Labels: p.CommonLabels, Annotations: p.CommonAnnotations}}
 }
 
-// receipt renders one delivery as a single greppable line. The format is the
+// receipt renders one ARRIVAL as a single greppable line. The format is the
 // interface — it is what an operator greps out of the log — so it is fixed.
+//
+// It says RECEIVED, not DELIVERED. The old wording sat exactly where a person
+// looks for proof that a page landed, and answered a different question.
 func receipt(path string, p *webhook, a alert) string {
 	return fmt.Sprintf(
-		"PAGE-DELIVERED path=%s receiver=%s status=%s alert=%s severity=%s "+
+		"ALERT-RECEIVED path=%s receiver=%s status=%s alert=%s severity=%s "+
 			"page=%s network=%s instance=%s :: %s",
 		path,
 		or(p.Receiver, "?"),
@@ -286,7 +454,7 @@ func receipt(path string, p *webhook, a alert) string {
 }
 
 // replay serves the ring as plain text, newest last, so `curl … | tail` reads
-// in the order the pages arrived.
+// in the order the records were made.
 func replay(c *zip.Ctx) error {
 	lines := recent.snapshot()
 	if len(lines) == 0 {

apps/o11y/alerts_egress_test.go

@@ -0,0 +1,255 @@
+// Copyright 2026 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package o11y
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+)
+
+// The regression this whole file exists for.
+//
+// Live, before the fix:
+//
+//	POST /v1/o11y/alerts/hanzo-slack  →  200 ok
+//	[o11y] PAGE-DELIVERED … alert=NodeMemoryCommittedCritical
+//	[o11y] PAGE-SLACK-FAILED … err=integrations: slack not connected for org
+//
+// Three green lights and a silent pager. An egress that cannot carry the alert
+// must not answer as though it did.
+func TestUndeliverableAlertAnswers503(t *testing.T) {
+	a := alertsApp(t)
+	swapEgress(t, egress{name: "slack", send: func(context.Context, string) error {
+		return errors.New("integrations: slack not connected for org")
+	}})
+
+	code, body := post(t, a, "/v1/o11y/alerts/hanzo-slack", pagePayload)
+	if code != http.StatusServiceUnavailable {
+		t.Fatalf("an undelivered alert answered %d %q — Alertmanager will record "+
+			"Notify success and never retry", code, body)
+	}
+	// The body has to name the failure: whoever reads the Alertmanager log next
+	// must not need a second lookup to learn which way out broke.
+	if !strings.Contains(body, "slack not connected") {
+		t.Fatalf("503 body does not name the cause: %q", body)
+	}
+}
+
+// No egress configured at all is the loudest case, not the quietest. The
+// previous code returned early and silently when CLOUD_ALERTS_SLACK_CHANNEL was
+// unset — a deployment could page nobody, forever, and answer 200 to every
+// notification.
+func TestNoEgressConfiguredIsAFailureNotANoOp(t *testing.T) {
+	a := alertsApp(t)
+	swapEgress(t) // nothing configured
+
+	code, body := post(t, a, "/v1/o11y/alerts/hanzo-pager", pagePayload)
+	if code != http.StatusServiceUnavailable {
+		t.Fatalf("no egress configured answered %d %q, want 503", code, body)
+	}
+	if !strings.Contains(body, alertsWebhookEnv) || !strings.Contains(body, alertsSlackChannelEnv) {
+		t.Fatalf("the 503 must say what to configure, got %q", body)
+	}
+}
+
+// The fallback carries it when Slack cannot — and the Slack failure is STILL
+// recorded. Degraded is not healthy: an egress that breaks silently behind a
+// working one is the same class of bug one layer in.
+func TestFallbackCarriesItAndTheSlackFailureIsStillRecorded(t *testing.T) {
+	a := alertsApp(t)
+	var carried atomic.Int32
+	swapEgress(t,
+		egress{name: "slack", send: func(context.Context, string) error {
+			return errors.New("integrations: slack not connected for org")
+		}},
+		egress{name: "webhook", send: func(context.Context, string) error {
+			carried.Add(1)
+			return nil
+		}},
+	)
+
+	code, body := post(t, a, "/v1/o11y/alerts/hanzo-slack", pagePayload)
+	if code != http.StatusOK || body != "ok" {
+		t.Fatalf("the fallback delivered it; got %d %q, want 200 ok", code, body)
+	}
+	if carried.Load() != 1 {
+		t.Fatalf("fallback egress was not used (%d sends)", carried.Load())
+	}
+
+	_, replayed := get(t, a, "/v1/o11y/alerts/last")
+	if !strings.Contains(replayed, "ALERT-EGRESS-FAILED egress=slack") {
+		t.Fatalf("a broken egress hid behind a working one:\n%s", replayed)
+	}
+	if !strings.Contains(replayed, "ALERT-DELIVERED via=webhook") {
+		t.Fatalf("delivery record does not name the egress that carried it:\n%s", replayed)
+	}
+}
+
+// The chain stops at the first success. A page delivered twice is a page
+// people learn to ignore.
+func TestChainStopsAtTheFirstSuccess(t *testing.T) {
+	a := alertsApp(t)
+	var second atomic.Int32
+	swapEgress(t,
+		egress{name: "slack", send: func(context.Context, string) error { return nil }},
+		egress{name: "webhook", send: func(context.Context, string) error {
+			second.Add(1)
+			return nil
+		}},
+	)
+	if code, _ := post(t, a, "/v1/o11y/alerts/hanzo-slack", pagePayload); code != http.StatusOK {
+		t.Fatalf("got %d, want 200", code)
+	}
+	if second.Load() != 0 {
+		t.Fatalf("both egresses fired; the page was delivered twice")
+	}
+}
+
+// Delivery must be SYNCHRONOUS. The old code sent in a detached goroutine, so
+// the response was written before the send was attempted — which made 200
+// structurally incapable of describing the send. This pins the ordering: the
+// handler cannot answer until the egress has been asked.
+func TestDeliveryIsSynchronous(t *testing.T) {
+	a := alertsApp(t)
+	var attempted atomic.Bool
+	swapEgress(t, egress{name: "slow", send: func(context.Context, string) error {
+		attempted.Store(true)
+		return nil
+	}})
+
+	code, _ := post(t, a, "/v1/o11y/alerts/hanzo-pager", pagePayload)
+	// No sleep, no poll: if the send were detached this read would race and the
+	// answer would already have been written.
+	if !attempted.Load() {
+		t.Fatalf("the response (%d) was written before the egress was asked", code)
+	}
+}
+
+// The arrival receipt still lands for an alert that could not be delivered.
+// Losing the record of the hop would trade one blind spot for another.
+func TestArrivalIsRecordedEvenWhenDeliveryFails(t *testing.T) {
+	a := alertsApp(t)
+	swapEgress(t, egress{name: "slack", send: func(context.Context, string) error {
+		return errors.New("nope")
+	}})
+	post(t, a, "/v1/o11y/alerts/hanzo-slack", pagePayload)
+
+	_, replayed := get(t, a, "/v1/o11y/alerts/last")
+	if !strings.Contains(replayed, "ALERT-RECEIVED") {
+		t.Fatalf("arrival receipt lost:\n%s", replayed)
+	}
+	if !strings.Contains(replayed, "ALERT-UNDELIVERED") {
+		t.Fatalf("undelivered outcome not recorded:\n%s", replayed)
+	}
+}
+
+// PAGE-DELIVERED must not exist anywhere in the record. It sat exactly where an
+// operator looks for proof a page landed and answered a different question; a
+// grep for it should now find nothing rather than find a lie.
+func TestTheMisleadingLineIsGone(t *testing.T) {
+	a := alertsApp(t)
+	post(t, a, "/v1/o11y/alerts/hanzo-slack", pagePayload)
+	_, replayed := get(t, a, "/v1/o11y/alerts/last")
+	if strings.Contains(replayed, "PAGE-DELIVERED") {
+		t.Fatalf("the misleading line survived:\n%s", replayed)
+	}
+}
+
+// The fallback egress posts a plain {"text": …} body, so the URL can be a Slack
+// incoming webhook, an ntfy topic, or anything else the owner actually has —
+// without this code learning a second format.
+func TestWebhookEgressPostsTextAndFailsLoudOnNon2xx(t *testing.T) {
+	var got struct {
+		Text string `json:"text"`
+	}
+	var status atomic.Int32
+	status.Store(http.StatusOK)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(b, &got)
+		w.WriteHeader(int(status.Load()))
+	}))
+	defer srv.Close()
+
+	if err := webhookSend(context.Background(), srv.URL, "hello ops"); err != nil {
+		t.Fatalf("webhook send: %v", err)
+	}
+	if got.Text != "hello ops" {
+		t.Fatalf("webhook body carried %q", got.Text)
+	}
+
+	// A fallback that swallowed a failure would leave nothing underneath to
+	// notice, so a non-2xx is an error.
+	status.Store(http.StatusInternalServerError)
+	err := webhookSend(context.Background(), srv.URL, "hello ops")
+	if err == nil {
+		t.Fatal("a 500 from the fallback egress was reported as delivered")
+	}
+	if !strings.Contains(err.Error(), "500") {
+		t.Fatalf("error does not carry the status: %v", err)
+	}
+}
+
+// configuredEgresses reads the environment and lists ONLY what is actually
+// configured. An egress that cannot be attempted must never be counted as one
+// that was — that arithmetic is what makes "no egress" a 503.
+func TestConfiguredEgressesReflectTheEnvironment(t *testing.T) {
+	for _, tc := range []struct {
+		name, channel, url string
+		want               []string
+	}{
+		{"neither", "", "", nil},
+		{"slack only", "#hanzo-ops", "", []string{"slack"}},
+		{"webhook only", "", "https://example.invalid/hook", []string{"webhook"}},
+		{"both, slack first", "#hanzo-ops", "https://example.invalid/hook", []string{"slack", "webhook"}},
+		{"whitespace is not configuration", "  ", "  ", nil},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv(alertsSlackChannelEnv, tc.channel)
+			t.Setenv(alertsWebhookEnv, tc.url)
+			var got []string
+			for _, e := range configuredEgresses() {
+				got = append(got, e.name)
+			}
+			if strings.Join(got, ",") != strings.Join(tc.want, ",") {
+				t.Fatalf("got %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// A malformed body is still an arrival, and it still has to be DELIVERED or
+// reported as undelivered. The old contract answered 200 to everything; the new
+// one keeps "do not 4xx a bad payload" (Alertmanager would retry it forever)
+// without inheriting "always claim success".
+func TestUnparseableBodyIsRecordedButStillNeedsAnEgress(t *testing.T) {
+	a := alertsApp(t)
+	swapEgress(t) // none configured
+	code, _ := post(t, a, "/v1/o11y/alerts/page", "not json at all")
+	if code != http.StatusServiceUnavailable {
+		t.Fatalf("got %d, want 503 — the body is unparseable, the egress is missing", code)
+	}
+	_, replayed := get(t, a, "/v1/o11y/alerts/last")
+	if !strings.HasPrefix(replayed, "ALERT-RECEIVED path=/v1/o11y/alerts/page receiver=?") {
+		t.Fatalf("unparseable arrival not recorded: %q", replayed)
+	}
+}

apps/o11y/alerts_slack_test.go

@@ -1,6 +1,7 @@
 package o11y
 
 import (
+	"context"
 	"strings"
 	"testing"
 )
@@ -43,13 +44,22 @@ func TestSlackTextBoundsAStorm(t *testing.T) {
 	}
 }
 
-// TestPageWithoutChannelIsInert — no channel configured means no paging, and
-// crucially no panic and no block: the receipt path must not depend on it.
-func TestPageWithoutChannelIsInert(t *testing.T) {
+// TestDeliverWithoutAnyEgressIsAFailure — the inverse of what this test used to
+// assert. "No channel configured means no paging" was treated as an inert,
+// acceptable state; it is the state in which every alert in the fleet reaches
+// nobody, so deliver() reports it as a failure and the handler answers 503.
+func TestDeliverWithoutAnyEgressIsAFailure(t *testing.T) {
 	t.Setenv(alertsSlackChannelEnv, "")
-	page(&webhook{Receiver: "r", Status: "firing"},
+	t.Setenv(alertsWebhookEnv, "")
+	via, failures := deliver(context.Background(), &webhook{Receiver: "r", Status: "firing"},
 		[]alert{{Labels: map[string]string{"alertname": "X"}}})
-	// Also inert with a channel but no alerts.
-	t.Setenv(alertsSlackChannelEnv, "#hanzo-ops")
-	page(&webhook{Receiver: "r", Status: "firing"}, nil)
+	if via != "" {
+		t.Fatalf("nothing was configured, yet delivery claims egress %q", via)
+	}
+	if len(failures) != 1 || failures[0].egress != "none" {
+		t.Fatalf("want one 'none' failure, got %+v", failures)
+	}
+	if !strings.Contains(reason(failures), alertsSlackChannelEnv) {
+		t.Fatalf("the reason must name the missing configuration: %s", reason(failures))
+	}
 }

apps/o11y/alerts_test.go

@@ -15,6 +15,7 @@
 package o11y
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -31,11 +32,23 @@ import (
 func alertsApp(t *testing.T) *zip.App {
 	t.Helper()
 	recent = alertRing{}
+	// A deterministic egress that always accepts. Without one the receiver now
+	// answers 503 — which is the entire change, and is pinned by its own tests
+	// below. Restored on cleanup so no test leaks a chain into the next.
+	swapEgress(t, egress{name: "test", send: func(context.Context, string) error { return nil }})
 	app := zip.New(zip.Config{Logger: luxlog.New("test")})
 	mountAlerts(app)
 	return app
 }
 
+// swapEgress installs a fixed chain for one test.
+func swapEgress(t *testing.T, chain ...egress) {
+	t.Helper()
+	prev := egressChain
+	egressChain = func() []egress { return chain }
+	t.Cleanup(func() { egressChain = prev })
+}
+
 // post/get wrap the package's shared `do` helper (scope_test.go) so there is
 // one request path in these tests.
 func post(t *testing.T, app *zip.App, path, body string) (int, string) {
@@ -72,7 +85,7 @@ func TestReceiptLineMatchesTheReplacedReceiver(t *testing.T) {
 		t.Fatal(err)
 	}
 	got := receipt("/v1/o11y/alerts/page", &p, p.Alerts[0])
-	want := "PAGE-DELIVERED path=/v1/o11y/alerts/page receiver=lux-pager status=firing " +
+	want := "ALERT-RECEIVED path=/v1/o11y/alerts/page receiver=lux-pager status=firing " +
 		"alert=LuxNetworkDown severity=critical page=true network=mainnet instance=luxd-0 " +
 		":: mainnet C-Chain has no reachable validator"
 	if got != want {
@@ -85,7 +98,7 @@ func TestReceiptLineMatchesTheReplacedReceiver(t *testing.T) {
 func TestReceiptDefaultsForAnEmptyPayload(t *testing.T) {
 	var p webhook
 	got := receipt("/v1/o11y/alerts/default", &p, alerts(&p)[0])
-	want := "PAGE-DELIVERED path=/v1/o11y/alerts/default receiver=? status=? alert=? " +
+	want := "ALERT-RECEIVED path=/v1/o11y/alerts/default receiver=? status=? alert=? " +
 		"severity=? page=- network=- instance=- :: "
 	if got != want {
 		t.Fatalf("defaults mismatch\n got: %q\nwant: %q", got, want)
@@ -134,12 +147,17 @@ func TestPostAlwaysAnswersOKAndReplayShowsIt(t *testing.T) {
 		t.Fatalf("replay: got %d", code)
 	}
 	lines := strings.Split(body, "\n")
-	if len(lines) != 4 {
-		t.Fatalf("want 4 receipts, got %d:\n%s", len(lines), body)
+	// Two lines per delivery: the ARRIVAL receipt and the DELIVERY outcome.
+	// Both, always — the pair is the record, and half of it was the bug.
+	if len(lines) != 8 {
+		t.Fatalf("want 8 lines (4 receipts + 4 outcomes), got %d:\n%s", len(lines), body)
 	}
 	for i, r := range []string{"default", "watchdog", "page", "slack"} {
-		if !strings.HasPrefix(lines[i], "PAGE-DELIVERED path=/v1/o11y/alerts/"+r+" ") {
-			t.Fatalf("line %d is not the %s receipt: %s", i, r, lines[i])
+		if !strings.HasPrefix(lines[i*2], "ALERT-RECEIVED path=/v1/o11y/alerts/"+r+" ") {
+			t.Fatalf("line %d is not the %s receipt: %s", i*2, r, lines[i*2])
+		}
+		if !strings.HasPrefix(lines[i*2+1], "ALERT-DELIVERED via=test ") {
+			t.Fatalf("line %d is not the %s outcome: %s", i*2+1, r, lines[i*2+1])
 		}
 	}
 }
@@ -154,7 +172,7 @@ func TestUnparseableBodyIsStillARecordedDelivery(t *testing.T) {
 		t.Fatalf("got %d %q, want 200 %q", code, body, "ok")
 	}
 	_, replayed := get(t, a, "/v1/o11y/alerts/last")
-	if !strings.HasPrefix(replayed, "PAGE-DELIVERED path=/v1/o11y/alerts/page receiver=? status=?") {
+	if !strings.HasPrefix(replayed, "ALERT-RECEIVED path=/v1/o11y/alerts/page receiver=? status=?") {
 		t.Fatalf("unparseable delivery not recorded: %q", replayed)
 	}
 }
@@ -172,10 +190,17 @@ func TestRingIsBoundedAndKeepsTheNewest(t *testing.T) {
 	if len(lines) != recentMax {
 		t.Fatalf("ring unbounded: %d lines, want %d", len(lines), recentMax)
 	}
-	if !strings.Contains(lines[0], "alert=A50") {
-		t.Fatalf("oldest survivor should be A50, got: %s", lines[0])
+	// 250 posts × 2 lines each = 500; the ring keeps the last 200, so the oldest
+	// survivor is line 300 — post 150's arrival receipt.
+	if !strings.Contains(lines[0], "alert=A150") {
+		t.Fatalf("oldest survivor should be A150, got: %s", lines[0])
 	}
-	if !strings.Contains(lines[len(lines)-1], fmt.Sprintf("alert=A%d", recentMax+49)) {
+	// The newest line is the last post's DELIVERY outcome — the pair ends with
+	// the fact about egress, which is the one an operator is reading for.
+	if !strings.Contains(lines[len(lines)-1], fmt.Sprintf("receiver=r%d", recentMax+49)) {
 		t.Fatalf("newest lost: %s", lines[len(lines)-1])
 	}
+	if !strings.Contains(lines[len(lines)-2], fmt.Sprintf("alert=A%d", recentMax+49)) {
+		t.Fatalf("newest arrival lost: %s", lines[len(lines)-2])
+	}
 }

apps/o11y/event_ingest.go

@@ -335,7 +335,16 @@ func (s *datastoreSink) Insert(ctx context.Context, table string, columns []stri
 			return fmt.Errorf("append row: %w", err)
 		}
 	}
-	return batch.Send()
+	if err := batch.Send(); err != nil {
+		return err
+	}
+	// Rows are counted HERE — the one choke point every plane row passes
+	// through (spans, logs, traces, observations, scores) — and only after Send
+	// returns, so the count is rows that LANDED, not rows that were offered.
+	// event.span going to zero here is the signal that was missing for four and
+	// a half months.
+	cloud.ObserveRows(table, len(rows))
+	return nil
 }
 
 // Close releases the native connection.

metrics_plane.go

@@ -0,0 +1,265 @@
+// Copyright 2026 Hanzo AI Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cloud
+
+import (
+	"context"
+	"runtime/debug"
+	"runtime/metrics"
+	"sort"
+	"sync"
+
+	"github.com/zap-proto/zip"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+)
+
+// The DATA PLANE's own measurements — what this process ingested, wrote, bound
+// and delivered.
+//
+// Everything the fleet could alert on until now was INFRASTRUCTURE: nodes,
+// PVCs, pods, containers, chains. Thirty rules, and not one of them could see a
+// single row of data move. The cost of that is measurable and was paid twice —
+// span ingest was dead for four and a half months, and an 88% ingest loss ran
+// under a green dashboard — because no rule could have noticed either. A plane
+// that carries data and reports only on the machine underneath it is not
+// observed; it is merely monitored.
+//
+// ZERO IS A MEASUREMENT — THE WHOLE FILE TURNS ON THIS.
+//
+// A counter that springs into existence on its first Add cannot express "this
+// stopped". The series is simply absent, and absent is what a metric store
+// shows for a name nobody ever wrote, a service that was never deployed, and a
+// typo — so a rule written against it either never fires or fires everywhere.
+// That is precisely why four and a half months of silence looked like health:
+// there was nothing to compare against zero.
+//
+// So every counter here is SEEDED AT ZERO for its known label values at
+// startup (seedCounters). From then on `increase(...[30m]) == 0` is a true
+// sentence about a live process, and a restart — which resets the counter and
+// leaves it at zero — reads as "still nothing", which is exactly right.
+// Distinguishing "stopped" from "never started" is the entire job.
+//
+// Naming follows metrics_http.go: hanzo_-prefixed, in full Prometheus
+// convention, because the exporter is installed with otlptranslator.NoTranslation
+// and will not add a suffix or prefix for you. Instruments resolve LAZILY for
+// the reason spelled out there — a meter taken at init binds to the no-op
+// provider that exists before the composition root runs, and every measurement
+// afterwards is discarded while the code looks perfectly instrumented.
+//
+// CARDINALITY. Every label value here comes from a finite, server-side set:
+// warehouse table names, ingest door names, egress names, the plane names this
+// process itself served. None is client-chosen.
+
+var (
+	planeOnce sync.Once
+
+	ingestItems metric.Int64Counter // hanzo_ingest_items_total{door,outcome}
+	planeRows   metric.Int64Counter // hanzo_plane_rows_written_total{table}
+	alertEgress metric.Int64Counter // hanzo_alert_delivery_total{egress,outcome}
+)
+
+// warehouseTables are the event-warehouse tables this estate writes. Seeding
+// them at zero is what makes "no span has landed in 30 minutes" a statement the
+// metric store can answer — event.span is the series that was missing for four
+// and a half months.
+var warehouseTables = []string{"event.event", "event.error", "event.log", "event.span"}
+
+// ingestDoors are the ingest doors whose admission outcome is counted. One
+// door today (POST /v1/event, the ONE event door); the list exists so seeding
+// stays honest when a second one is added.
+var ingestDoors = []string{"event"}
+
+// planeInstruments resolves the data-plane instruments and seeds them. Lazy:
+// see the note above and metrics_http.go's instruments().
+func planeInstruments() {
+	planeOnce.Do(func() {
+		m := otel.Meter(meterName)
+		ingestItems, _ = m.Int64Counter("hanzo_ingest_items_total",
+			metric.WithDescription("Items offered to an ingest door, by door and admission outcome (accepted/dropped)."))
+		planeRows, _ = m.Int64Counter("hanzo_plane_rows_written_total",
+			metric.WithDescription("Rows written to an event-warehouse table."))
+		alertEgress, _ = m.Int64Counter("hanzo_alert_delivery_total",
+			metric.WithDescription("Alert batches carried out of the process, by egress and outcome (delivered/failed)."))
+
+		registerPlaneGauges(m)
+		seedCounters()
+	})
+}
+
+// seedCounters writes 0 to every known label combination so the series EXISTS
+// before anything happens. Adding zero is not a no-op to a metric store: it is
+// the difference between a rule that can say "this stopped" and one that can
+// only say nothing.
+func seedCounters() {
+	ctx := context.Background()
+	for _, t := range warehouseTables {
+		if planeRows != nil {
+			planeRows.Add(ctx, 0, metric.WithAttributes(attribute.String("table", t)))
+		}
+	}
+	for _, d := range ingestDoors {
+		for _, outcome := range []string{"accepted", "dropped"} {
+			if ingestItems != nil {
+				ingestItems.Add(ctx, 0, metric.WithAttributes(
+					attribute.String("door", d),
+					attribute.String("outcome", outcome)))
+			}
+		}
+	}
+	// Alert egresses are seeded on the failing side only. "Delivered" appearing
+	// for the first time is unambiguous; a `failed` series that does not exist
+	// until the first failure would leave the meta-alert unable to distinguish
+	// "no failures" from "nothing is even trying".
+	for _, e := range []string{"slack", "webhook", "none"} {
+		if alertEgress != nil {
+			alertEgress.Add(ctx, 0, metric.WithAttributes(
+				attribute.String("egress", e),
+				attribute.String("outcome", "failed")))
+		}
+	}
+}
+
+// ObserveIngest records one ingest door's admission outcome.
+//
+// accepted and dropped are reported TOGETHER because the useful question is a
+// ratio, not a count: "88% of what was offered was dropped" is an outage,
+// "8,000 items were dropped" is a number whose meaning depends on a second
+// series nobody fetched.
+func ObserveIngest(door string, accepted, dropped int) {
+	planeInstruments()
+	if ingestItems == nil || door == "" {
+		return
+	}
+	ctx := context.Background()
+	if accepted > 0 {
+		ingestItems.Add(ctx, int64(accepted), metric.WithAttributes(
+			attribute.String("door", door), attribute.String("outcome", "accepted")))
+	}
+	if dropped > 0 {
+		ingestItems.Add(ctx, int64(dropped), metric.WithAttributes(
+			attribute.String("door", door), attribute.String("outcome", "dropped")))
+	}
+}
+
+// ObserveRows records rows landed in an event-warehouse table. Called from the
+// two writers that put rows there — the o11y plane sink and the analytics bus
+// drain — so `event.span` counts whichever one is carrying it.
+func ObserveRows(table string, n int) {
+	planeInstruments()
+	if planeRows == nil || table == "" || n <= 0 {
+		return
+	}
+	planeRows.Add(context.Background(), int64(n),
+		metric.WithAttributes(attribute.String("table", table)))
+}
+
+// ObserveAlertDelivery records one alert batch's egress outcome. This is the
+// metric behind the meta-alert: paging that cannot reach a human is itself an
+// incident, and the only reason it went unnoticed for months is that nothing
+// counted it.
+func ObserveAlertDelivery(egress, outcome string) {
+	planeInstruments()
+	if alertEgress == nil || egress == "" {
+		return
+	}
+	alertEgress.Add(context.Background(), 1, metric.WithAttributes(
+		attribute.String("egress", egress), attribute.String("outcome", outcome)))
+}
+
+// servedPlanes is the set of plane names THIS process bound, recorded by
+// ServePlane. It is the "should be bound" set, and it is authoritative because
+// it is written by the act of binding rather than by a list somebody maintains:
+// an app that this process was asked to serve belongs here, and nothing else
+// does. Walking the whole 112-row manifest instead would dial sockets for apps
+// that were never meant to be here and call their absence a fault.
+var (
+	servedMu     sync.Mutex
+	servedPlanes = map[string]bool{}
+)
+
+// planeServed notes that name's socket was bound here, so the gauge below can
+// notice when it stops answering.
+func planeServed(name string) {
+	servedMu.Lock()
+	defer servedMu.Unlock()
+	servedPlanes[name] = true
+}
+
+// servedPlaneNames returns the expected set, sorted for a stable scrape.
+func servedPlaneNames() []string {
+	servedMu.Lock()
+	defer servedMu.Unlock()
+	out := make([]string, 0, len(servedPlanes))
+	for n := range servedPlanes {
+		out = append(out, n)
+	}
+	sort.Strings(out)
+	return out
+}
+
+// registerPlaneGauges installs the observable gauges: facts that are true at
+// collection time rather than events to be tracked.
+func registerPlaneGauges(m metric.Meter) {
+	bound, err := m.Int64ObservableGauge("hanzo_plane_peer_bound",
+		metric.WithDescription("1 when a plane socket this process served still accepts connections, 0 when it does not."))
+	if err == nil {
+		_, _ = m.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+			for _, name := range servedPlaneNames() {
+				up := int64(0)
+				// listening CONNECTS rather than stats: on a volume-backed run
+				// directory a socket file outlives the process that bound it, so
+				// the file proves nothing. A present-but-unusable socket returns
+				// an error and is reported as down, which is what it is.
+				if ok, probeErr := listening(zip.SocketPath(name)); ok && probeErr == nil {
+					up = 1
+				}
+				o.ObserveInt64(bound, up, metric.WithAttributes(attribute.String("peer", name)))
+			}
+			return nil
+		}, bound)
+	}
+
+	// Process memory against the limit the runtime actually enforces.
+	//
+	// ContainerOOMKilled is a POST-MORTEM: the kernel has already killed the
+	// process, the request in flight is gone, and cloud is a single writer, so
+	// by the time that alert fires the API has had an outage. GOMEMLIMIT is the
+	// ceiling the Go runtime governs itself against, and watching the approach
+	// to it is the only version of this signal that arrives while there is
+	// still something to do about it.
+	used, uerr := m.Int64ObservableGauge("hanzo_process_memory_bytes",
+		metric.WithDescription("Memory the Go runtime governs against GOMEMLIMIT (total mapped, less released)."))
+	limit, lerr := m.Int64ObservableGauge("hanzo_process_memory_limit_bytes",
+		metric.WithDescription("The effective GOMEMLIMIT this process runs under."))
+	if uerr == nil && lerr == nil {
+		// SetMemoryLimit(-1) READS the limit without setting it — the documented
+		// way to ask. This process must never change its own ceiling; the
+		// deployment owns that.
+		samples := []metrics.Sample{
+			{Name: "/memory/classes/total:bytes"},
+			{Name: "/memory/classes/heap/released:bytes"},
+		}
+		_, _ = m.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+			metrics.Read(samples)
+			total := int64(samples[0].Value.Uint64())
+			released := int64(samples[1].Value.Uint64())
+			o.ObserveInt64(used, total-released)
+			o.ObserveInt64(limit, debug.SetMemoryLimit(-1))
+			return nil
+		}, used, limit)
+	}
+}

middleware_tracing.go

@@ -16,6 +16,7 @@ package cloud
 
 import (
 	"strings"
+	"time"
 
 	"github.com/zap-proto/zip"
 	"go.opentelemetry.io/otel"
@@ -92,6 +93,7 @@ func TracingMiddleware() zip.Handler {
 			return c.Continue()
 		}
 		method := strings.Clone(c.Method())
+		start := time.Now()
 
 		// Start the span off the request context and write the enriched context
 		// back so the rest of the chain (and every downstream client that pulls
@@ -127,9 +129,22 @@ func TracingMiddleware() zip.Handler {
 		// handler runs, so by here c.Org() reflects the authenticated org.
 		status := c.Fiber().Response().StatusCode()
 		span.SetAttributes(attribute.Int("http.response.status_code", status))
-		if org := strings.Clone(c.Org()); org != "" {
+		org := strings.Clone(c.Org())
+		if org != "" {
 			span.SetAttributes(attribute.String("hanzo.org", org))
 		}
+
+		// The METRIC half of the same observation. It belongs here and nowhere
+		// else: this is the one place every /v1 request already has its path,
+		// its status and its VALIDATED org in hand, and computing them twice in
+		// a second middleware would be the same fact measured two ways.
+		//
+		// It had been written (metrics_http.go) and never called, so
+		// hanzo_http_requests_total did not exist in the store — which is why
+		// "/v1/event is 5xx" was unalertable while the Sentry envelope returned
+		// 503 for a day with nobody paged. A metric nothing calls is not
+		// instrumentation, it is a comment.
+		observeRequest(productFromPath(path), org, status, time.Since(start))
 		switch {
 		case err != nil:
 			span.RecordError(err)

plane.go

@@ -139,6 +139,11 @@ func ServePlane(name string, log luxlog.Logger) (func() error, error) {
 	if err := awaitSocket(path, errs, planeBindWait); err != nil {
 		return nil, fmt.Errorf("plane %s: %w", name, err)
 	}
+	// This process was asked to serve this plane, so from here on its socket
+	// going quiet is a FAULT rather than an absence. Recording it at the moment
+	// of binding is what lets hanzo_plane_peer_bound tell those two apart —
+	// the distinction o11y's own disappearance turned on.
+	planeServed(name)
 	if log != nil {
 		log.Info("plane listening", "app", name, "sock", path)
 	}

telemetry.go

@@ -319,6 +319,17 @@ func InstallTelemetry(ctx context.Context, log luxlog.Logger, serviceName string
 	// Two signals, two destinations, two decisions.
 	mp, stopMeter := installMeter(log, res)
 
+	// Seed the data-plane instruments the moment the provider exists.
+	//
+	// This call is the difference between a rule that can say "ingest stopped"
+	// and one that can say nothing. A counter first touched by its first event
+	// has no series until that event happens, so an ingest path that never runs
+	// is indistinguishable in the store from one that was never built — which is
+	// exactly how span ingest stayed dead for four and a half months without a
+	// single rule being able to notice. Seeding at boot means zero is on the
+	// wire from the first scrape, and silence becomes a measurement.
+	planeInstruments()
+
 	// TRACES. Enabled when a ZAP endpoint is set OR (legacy) any OTLP endpoint is
 	// set OR a co-resident sink is expected (spans route in-process, no wire
 	// endpoint needed). Keep the clean no-op-when-unset posture so this is safe