Khi nào thì thôi cố

export const internalDeliveryOutcomeRequestSchema = z.strictObject({
  delivery_id: z.string().uuid(),
  attempt: z.number().int().positive(),
  status: z.number().int().optional(),
  error: z.string().max(2000).optional(),
  latency_ms: z.number().int().nonnegative(),
});

@@ -161,6 +161,66 @@ export function deliverySubjectFor(environmentId: string): string {
   return `${DELIVERY_SUBJECT_PREFIX}.${environmentId}`;
 }
 
+// ---------------------------------------------------------------------------
+// Analytical events (chapter 3.6, constitution III).
+//
+// The THIRD grammar in this file, and it is here for the reason the other two
+// are: a consumer that assembles its own subject filter receives nothing the day
+// the grammar changes — no error, no warning, just an empty stream position. Part
+// 4's ingester is that consumer, and it does not exist yet, which is exactly when
+// a shared definition is cheapest to establish.
+//
+// `analytics.{domain}.{action}.{environment_id}` extends chapter 3.4's
+// `events.{domain}.{action}.{env}` rather than inventing a second convention.
+//
+// The stream is SEPARATE from `EVENTS` and `DELIVERIES`, and that is a decision
+// rather than tidiness (research R4). `EVENTS` carries tenant domain events whose
+// consumers must not miss one; `DELIVERIES` carries work. Attempt records are
+// neither: high volume, deliberately lossy (research R5), and a consumer that
+// wants them does not want every message event alongside.
+// ---------------------------------------------------------------------------
+
+export const ANALYTICS_STREAM = "ANALYTICS";
+export const ANALYTICS_SUBJECT_PREFIX = "analytics";
+export const ALL_ANALYTICS_SUBJECT = `${ANALYTICS_SUBJECT_PREFIX}.>`;
+
+/** A UUID, checked rather than trusted.
+ *
+ * The environment id becomes a SUBJECT TOKEN, and NATS subjects are
+ * dot-delimited: a value containing a dot silently creates a deeper subject than
+ * intended, and one containing `*` or `>` creates a wildcard. Neither fails at
+ * publish time. Both would put one tenant's attempt records where another
+ * tenant's filter can reach them, which is constitution I stated as a parsing
+ * problem. Every caller in this platform holds a uuid already, so refusing
+ * anything else costs nothing. */
+const UUID =
+  /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
+
+export function analyticsSubjectFor(
+  domain: string,
+  action: string,
+  environmentId: string,
+): string {
+  if (!domain) throw new Error("a domain is required");
+  if (!action) throw new Error("an action is required");
+  if (!UUID.test(environmentId)) {
+    throw new Error("an environment id must be a uuid");
+  }
+  return [ANALYTICS_SUBJECT_PREFIX, domain, action, environmentId].join(".");
+}
+
+/** The one action this chapter publishes. Named so a consumer filters on a
+ * constant rather than on a string it typed. */
+export const WEBHOOK_ATTEMPT_ACTION = { domain: "webhook", action: "attempt" };
+
+export function webhookAttemptSubject(environmentId: string): string {
+  return analyticsSubjectFor(
+    WEBHOOK_ATTEMPT_ACTION.domain,
+    WEBHOOK_ATTEMPT_ACTION.action,
+    environmentId,
+  );
+}
+
 // ---------------------------------------------------------------------------
 // The dispatch contract (chapter 3.5, constitution IV).
 //

import { describe, expect, it } from "vitest";
 
import {
  ALL_ANALYTICS_SUBJECT,
  ALL_EVENTS_SUBJECT,
  analyticsSubjectFor,
  subjectFor,
  webhookAttemptSubject,
} from "./internal.js";
 
// The subject grammar is ADR-02's, and a consumer that filters on a subject it
// assembled itself receives nothing the day the grammar drifts — silently. So
// the grammar is held here rather than trusted: the shape of a subject, the one
// abbreviation, the failure that must be loud, and the guarantee that whatever
// comes out is still reachable from the wildcard every consumer subscribes to.
 
/** `events.>` in NATS terms: everything below the prefix, at any depth. */
function matchesWildcard(subject: string, wildcard: string): boolean {
  const prefix = wildcard.replace(/>$/, "");
  return subject.startsWith(prefix) && subject.length > prefix.length;
}
 
describe("subjectFor builds ADR-02's `events.{domain}.{action}.{env}`", () => {
  it("puts the environment last", () => {
    expect(subjectFor("message.created", "env_123")).toBe(
      "events.msg.created.env_123",
    );
  });
 
  it("abbreviates `message` to `msg`, as ADR-02's own example does", () => {
    const subject = subjectFor("message.created", "env_123");
    expect(subject.split(".")[1]).toBe("msg");
  });
 
  it("passes a domain with no abbreviation through unchanged", () => {
    expect(subjectFor("channel.created", "env_123")).toBe(
      "events.channel.created.env_123",
    );
  });
 
  it("throws on a missing part instead of producing `events..created.`", () => {
    expect(() => subjectFor("", "env_123")).toThrow(/event type is required/);
    expect(() => subjectFor("message.created", "")).toThrow(
      /environment id is required/,
    );
  });
 
  it("produces subjects the consumer's wildcard matches", () => {
    for (const type of ["message.created", "channel.created"]) {
      expect(matchesWildcard(subjectFor(type, "env_123"), ALL_EVENTS_SUBJECT)).toBe(
        true,
      );
    }
  });
});
 
// The analytics grammar (chapter 3.6). Held for the same reason as the one above
// — Part 4's ingester will filter on it and does not exist yet — plus one this
// grammar has and that one does not: the environment id becomes a SUBJECT TOKEN,
// and a subject token is parsed by the broker rather than escaped by it.
describe("analyticsSubjectFor builds `analytics.{domain}.{action}.{env}`", () => {
  const ENV = "9f3c1e7a-0b2d-4c8e-9a1f-6d5b4c3a2e10";
 
  it("puts the environment last, as the events grammar does", () => {
    expect(analyticsSubjectFor("webhook", "attempt", ENV)).toBe(
      `analytics.webhook.attempt.${ENV}`,
    );
  });
 
  it("names the one action this chapter publishes", () => {
    expect(webhookAttemptSubject(ENV)).toBe(analyticsSubjectFor("webhook", "attempt", ENV));
  });
 
  it("produces subjects the ingester's wildcard matches", () => {
    expect(matchesWildcard(webhookAttemptSubject(ENV), ALL_ANALYTICS_SUBJECT)).toBe(
      true,
    );
  });
 
  it("does not collide with the events stream's wildcard", () => {
    // Two streams, two prefixes. A subject reachable from both would mean the
    // EVENTS consumers start receiving attempt records, which is the coupling
    // research R4 separated the streams to avoid.
    expect(matchesWildcard(webhookAttemptSubject(ENV), ALL_EVENTS_SUBJECT)).toBe(
      false,
    );
  });
 
  it("refuses an environment id that is not a uuid", () => {
    // THIS IS THE TENANT-ISOLATION CASE, not input tidiness. A subject is
    // dot-delimited and NATS reads `*` and `>` as wildcards, so a value carrying
    // either would publish one tenant's attempt records where another tenant's
    // filter can reach them — and nothing would fail at publish time.
    for (const bad of [
      "",
      "env_123",
      "not-a-uuid",
      // The dangerous three: a dot creates a deeper subject than intended, and
      // the two wildcards create a subscription rather than a destination.
      "9f3c1e7a-0b2d-4c8e-9a1f-6d5b4c3a2e10.extra",
      "*",
      ">",
    ]) {
      expect(() => analyticsSubjectFor("webhook", "attempt", bad)).toThrow(
        /environment id must be a uuid/,
      );
    }
  });
 
  it("throws on a missing domain or action rather than producing `analytics..`", () => {
    expect(() => analyticsSubjectFor("", "attempt", ENV)).toThrow(/domain is required/);
    expect(() => analyticsSubjectFor("webhook", "", ENV)).toThrow(/action is required/);
  });
});

@@ -7,6 +7,8 @@ import {
   type NatsConnection,
 } from "nats";
 
+import { ALL_ANALYTICS_SUBJECT, ANALYTICS_STREAM } from "@relay/protocol";
+
 import type { Publisher, PublishedMessage } from "./publisher";
 
 // The one adapter that knows what a broker is (chapter 3.3, ADR-02).
@@ -41,6 +43,13 @@ const SECOND_NS = 1_000_000_000;
  * weekend. */
 const MAX_AGE_NS = 7 * 24 * 60 * 60 * SECOND_NS;
 
+/** The analytics stream's own bound (chapter 3.6). The same seven days as
+ * `EVENTS`, for a different reason: there it absorbs a consumer outage without
+ * losing a tenant's events, here it absorbs an ingester outage without letting
+ * the stream become a database. Named separately so the two can diverge when
+ * Part 4 has an opinion, rather than one changing the other by accident. */
+const ANALYTICS_MAX_AGE_NS = 7 * 24 * 60 * 60 * SECOND_NS;
+
 /** An unbounded stream is a full disk with extra steps. The bound turns that
  * into a number an operator can watch — and with `discard: old`, hitting it
  * loses the OLDEST events rather than refusing new publishes. Refusing
@@ -94,6 +103,50 @@ export async function ensureStream(nc: NatsConnection): Promise<void> {
   await jsm.streams.update(STREAM, { ...existing.config, ...mutable });
 }
 
+/** The ANALYTICS stream (chapter 3.6, constitution III).
+ *
+ * A THIRD stream rather than a third use of `EVENTS`, and the reasons are all
+ * about the difference between an operational event and an analytical one
+ * (research R4): different volume, different retention, and a Part 4 ingester
+ * that should be able to consume attempt records without also consuming every
+ * message event in the platform.
+ *
+ * Reuses the `ensure` parameter chapter 3.5 added rather than introducing a
+ * second mechanism — the publisher already knows how to bring one stream into
+ * existence, and "which stream" has been a parameter since a delivery published
+ * to `deliveries.*` came back 503.
+ *
+ * SEVEN DAYS, `discard: old`, and no acknowledgement anywhere: nothing consumes
+ * this stream in this chapter. Seven days is long enough for an ingester to be
+ * down over a long weekend and short enough that the stream does not quietly
+ * become the analytical database it is supposed to feed. At the bound the oldest
+ * analytics is the least interesting, which is the one case where dropping the
+ * old data is the right answer — the opposite choice on `EVENTS` would lose a
+ * tenant's events, and there `discard: old` is a bound on a liability instead. */
+export async function ensureAnalyticsStream(nc: NatsConnection): Promise<void> {
+  const jsm = await nc.jetstreamManager();
+  const mutable = {
+    subjects: [ALL_ANALYTICS_SUBJECT],
+    max_age: ANALYTICS_MAX_AGE_NS,
+    max_bytes: MAX_BYTES,
+    discard: DiscardPolicy.Old,
+    num_replicas: replicaCount(),
+  };
+  const existing = await jsm.streams.info(ANALYTICS_STREAM).catch(() => null);
+  if (existing === null) {
+    await jsm.streams.add({
+      name: ANALYTICS_STREAM,
+      retention: RetentionPolicy.Limits,
+      storage: StorageType.File,
+      ...mutable,
+    });
+    return;
+  }
+  // Retention and storage are immutable on an existing stream — chapter 3.4
+  // measured that (its research R1), and the lesson transfers unchanged.
+  await jsm.streams.update(ANALYTICS_STREAM, { ...existing.config, ...mutable });
+}
+
 export function createJetStreamPublisher({
   url = process.env.RELAY_NATS_URL ?? DEFAULT_NATS_URL,
   ensure = ensureStream,

import { webhookAttemptSubject } from "@relay/protocol";
import type { Logger } from "@relay/service-kit";
 
import type { Publisher } from "../outbox/publisher";
 
// The attempt record, on its way to the analytical path (chapter 3.6, FR-001,
// constitution III).
//
// This file is small and its whole subject is one decision, so the decision is
// written here rather than in the chapter alone: THE PUBLISH IS ALLOWED TO FAIL,
// and failing must cost the delivery nothing.
//
// Constitution III: "Analytical events are emitted asynchronously — never
// synchronously on the request path. Failure or backlog of the analytical
// pipeline MUST NOT affect message delivery, API availability, or webhook
// dispatch." FR-WHK-06 says every delivery attempt shall be recorded. Those two
// cannot both be maximised, and research R5 chose which one wins:
//
//   * record every attempt without loss → the record shares a transaction with
//     the outcome (3.3's outbox), and then a stalled analytics consumer backs up
//     an operational table;
//   * guarantee independence → the publish happens after the commit, outside it,
//     and a crash in that gap loses the record.
//
// Independence wins, because the two costs are not equal. A lost attempt record
// is a gap in a dashboard. A blocked outcome transaction is a customer's webhooks
// stopping because a metering pipeline is unwell — which constitution III names
// as a design failure in as many words.
//
// So "every attempt" is APPROXIMATE, and the chapter says so in the paragraph
// that introduces the feature rather than in a footnote.
 
/** The publisher that reaches the ANALYTICS stream, as a DI token.
 *
 * A SECOND publisher rather than a second use of the first. Each
 * `createJetStreamPublisher` ensures exactly one stream, and chapter 3.5 already
 * learned what happens when a publisher is pointed at a stream it did not create:
 * every publish comes back 503, which is a confusing way to discover that
 * JetStream does not create streams on demand. Declared here, beside the function
 * that uses it, so the module wiring reads as configuration rather than as
 * knowledge. */
export const ANALYTICS_PUBLISHER = Symbol("ANALYTICS_PUBLISHER");
 
/** What one attempt looked like. Assembled by the caller from the outcome it has
 * just recorded, because the api is the only party holding all of it: the
 * dispatcher has the status and the latency but not the environment, and nothing
 * but the api knows what the outcome WAS. */
export interface AttemptRecord {
  deliveryId: string;
  endpointId: string;
  environmentId: string;
  eventId: string;
  attempt: number;
  /** Absent when nothing answered. A timeout has no status, and inventing one —
   * 0, 599 — would make every dashboard built on this lie in the same direction. */
  status?: number;
  /** Present when there was no status. Already capped at 2000 characters by the
   * seam's schema, so nothing is truncated here. */
  error?: string;
  latencyMs: number;
  outcome: "delivered" | "rescheduled" | "dead_lettered";
  attemptedAt: Date;
}
 
/** The wire shape, in `contracts/attempts.md`. Snake case because it leaves the
 * platform: a consumer written in another language reads this, and Part 4's
 * ingester is the first of them. */
interface AttemptEvent {
  delivery_id: string;
  endpoint_id: string;
  environment_id: string;
  event_id: string;
  attempt: number;
  attempted_at: string;
  status?: number;
  error?: string;
  latency_ms: number;
  outcome: string;
}
 
/** IDENTIFIERS, STATUSES AND DURATIONS ONLY (FR-004, NFR-SEC-06).
 *
 * The record is built by naming every field rather than by spreading the input,
 * and that is the point. A spread would carry whatever a future caller happened
 * to put on the record — a payload, a decrypted secret, a header — onto a stream
 * with seven-day retention that nothing in this platform reads yet. An allow-list
 * fails closed when somebody adds a field; a spread fails open. */
function shape(record: AttemptRecord): AttemptEvent {
  return {
    delivery_id: record.deliveryId,
    endpoint_id: record.endpointId,
    environment_id: record.environmentId,
    event_id: record.eventId,
    attempt: record.attempt,
    attempted_at: record.attemptedAt.toISOString(),
    // `exactOptionalPropertyTypes` is on, so these are spread in rather than
    // assigned: an explicit `undefined` is not the same as an absent key, and the
    // difference is the whole meaning of "nothing answered".
    ...(record.status !== undefined ? { status: record.status } : {}),
    ...(record.error !== undefined ? { error: record.error } : {}),
    latency_ms: record.latencyMs,
    outcome: record.outcome,
  };
}
 
/** Publish one attempt record. Never throws.
 *
 * Call this AFTER the outcome transaction has committed, never inside it. Inside,
 * a broker that is slow makes an operational transaction slow and holds a row
 * lock while it waits, which is the coupling this whole design exists to avoid —
 * and the sabotage battery moves this call inside the transaction to prove the
 * suite would notice.
 *
 * The failure is swallowed HERE rather than at each call site, so no caller can
 * accidentally turn an analytics outage into a delivery outage by forgetting a
 * `catch`. A caller that wanted to know would have to ask, and none does. */
export async function publishAttempt(
  publisher: Publisher,
  logger: Logger,
  record: AttemptRecord,
): Promise<void> {
  try {
    await publisher.publish({
      subject: webhookAttemptSubject(record.environmentId),
      // The broker's deduplication key, and it is `{delivery}:{attempt}` for the
      // reason chapter 3.5 learned the hard way: the delivery id alone is stable
      // across all seven attempts, and using it collapsed every retry into the
      // first attempt's message. That bug cost the platform its entire retry
      // schedule and was found by a walk rather than by a test. Here the same
      // mistake would silently discard attempts 2 through 7 from every dashboard.
      id: `${record.deliveryId}:${record.attempt}`,
      payload: shape(record),
    });
  } catch (error) {
    // One line, no payload, no secret — and no rethrow. The delivery already
    // happened and the outcome is already committed; there is nothing this
    // failure can usefully undo, and plenty it could break by trying.
    logger.log("error", "analytics.attempt_publish_failed", {
      delivery_id: record.deliveryId,
      attempt: record.attempt,
      error: String(error),
    });
  }
}

import { describe, expect, it } from "vitest";
 
import { createLogger, type Logger } from "@relay/service-kit";
 
import type { PublishedMessage, Publisher } from "../outbox/publisher";
import { publishAttempt, type AttemptRecord } from "./analytics";
 
// The attempt record's SHAPE and its SILENCE (chapter 3.6).
//
// Two claims are worth a unit test, and both are about what the payload is not.
// FR-004 says an attempt record carries identifiers, statuses and durations and
// nothing else; contract invariant 4 says a publish failure changes nothing for
// the caller. Neither needs a broker to check, and a test that needed one would be
// a test nobody runs while writing the code.
 
/** Captures what was handed to the broker, without being one. */
function recorder(): Publisher & { sent: PublishedMessage[] } {
  const sent: PublishedMessage[] = [];
  return {
    sent,
    async publish(message) {
      sent.push(message);
    },
    async close() {},
  };
}
 
function exploding(error = new Error("nats: no responders")): Publisher {
  return {
    async publish() {
      throw error;
    },
    async close() {},
  };
}
 
/** The sink receives one JSON STRING per line, so it is parsed back rather than
 * inspected as an object — which also proves the fields survive serialisation,
 * the only form anybody reads a log line in. */
function captured(): { logger: Logger; lines: Record<string, unknown>[] } {
  const lines: Record<string, unknown>[] = [];
  const logger = createLogger("analytics-test", (line) => {
    lines.push(JSON.parse(String(line)) as Record<string, unknown>);
  });
  return { logger, lines };
}
 
const ENV = "9f3c1e7a-0b2d-4c8e-9a1f-6d5b4c3a2e10";
 
const base: AttemptRecord = {
  deliveryId: "1c2d3e4f-0000-4000-8000-000000000001",
  endpointId: "1c2d3e4f-0000-4000-8000-000000000002",
  environmentId: ENV,
  eventId: "1c2d3e4f-0000-4000-8000-000000000003",
  attempt: 3,
  status: 503,
  latencyMs: 214,
  outcome: "rescheduled",
  attemptedAt: new Date("2026-08-18T09:14:22.481Z"),
};
 
const silent = createLogger("analytics-test", () => {});
 
describe("publishAttempt shapes the event contracts/attempts.md describes", () => {
  it("carries the four identifiers, the attempt, the status, the latency and the outcome", async () => {
    const publisher = recorder();
    await publishAttempt(publisher, silent, base);
 
    expect(publisher.sent).toHaveLength(1);
    expect(publisher.sent[0]!.payload).toEqual({
      delivery_id: base.deliveryId,
      endpoint_id: base.endpointId,
      environment_id: ENV,
      event_id: base.eventId,
      attempt: 3,
      attempted_at: "2026-08-18T09:14:22.481Z",
      status: 503,
      latency_ms: 214,
      outcome: "rescheduled",
    });
  });
 
  it("puts the environment on the subject as well as in the payload", async () => {
    // Invariant 2. The subject is how a future consumer filters by tenant, and a
    // subject disagreeing with its payload would route one environment's records
    // under another's filter.
    const publisher = recorder();
    await publishAttempt(publisher, silent, base);
 
    const { subject, payload } = publisher.sent[0]!;
    expect(subject).toBe(`analytics.webhook.attempt.${ENV}`);
    expect(subject.endsWith((payload as { environment_id: string }).environment_id)).toBe(
      true,
    );
  });
 
  it("deduplicates on the delivery AND the attempt, not the delivery alone", async () => {
    // Chapter 3.5's bug, in the one place it could recur. The delivery id is
    // stable across all seven attempts, so a `msgID` of the delivery alone would
    // let the broker collapse attempts 2 through 7 into the first — and every
    // dashboard built on this stream would show one attempt per failing delivery
    // for ever, with no error anywhere.
    const publisher = recorder();
    await publishAttempt(publisher, silent, { ...base, attempt: 1 });
    await publishAttempt(publisher, silent, { ...base, attempt: 2 });
 
    const ids = publisher.sent.map((m) => m.id);
    expect(ids).toEqual([`${base.deliveryId}:1`, `${base.deliveryId}:2`]);
    expect(new Set(ids).size).toBe(2);
  });
 
  it("omits `status` entirely when nothing answered, rather than sending a zero", async () => {
    // A timeout has no status. Sending 0 or 599 would be a number a dashboard
    // could average, and the average would be a fiction.
    const publisher = recorder();
    // Built by omission rather than by setting `status: undefined`, because
    // `exactOptionalPropertyTypes` makes those two different things and only one of
    // them is what a timeout looks like.
    const timedOut: AttemptRecord = { ...base };
    delete timedOut.status;
    await publishAttempt(publisher, silent, {
      ...timedOut,
      error: "timeout after 10000ms",
      latencyMs: 10_000,
    });
 
    const payload = publisher.sent[0]!.payload as Record<string, unknown>;
    expect("status" in payload).toBe(false);
    expect(payload["error"]).toBe("timeout after 10000ms");
    // The latency of a timeout IS the timeout, and it is still reported: "how
    // long did we wait" is a real question even when "what did they say" has no
    // answer.
    expect(payload["latency_ms"]).toBe(10_000);
  });
 
  it("omits `error` when there was a status", async () => {
    const publisher = recorder();
    await publishAttempt(publisher, silent, base);
    expect("error" in (publisher.sent[0]!.payload as object)).toBe(false);
  });
 
  it("carries no payload, secret, signature or header — whatever it is handed", async () => {
    // FR-004 and SC-006, and the reason this asserts on EXTRA fields rather than
    // on the ones it expects: the risk is not that a field goes missing, it is
    // that one is added. `shape` names every key it copies, so a caller that
    // decorates the record with a secret cannot leak it onto a stream with
    // seven-day retention. A spread would have.
    const publisher = recorder();
    await publishAttempt(publisher, silent, {
      ...base,
      // Everything a careless future caller might attach. None of it is on the
      // interface, hence the cast — a compile error is the FIRST line of defence
      // and this test is the second.
      ...({
        payload: { text: "B2, north ramp" },
        secret: "whsec_do_not_publish_this",
        signature: "v1=deadbeef",
        headers: { authorization: "Bearer rk_live_xxx" },
        url: "https://customer.example/hook",
      } as unknown as AttemptRecord),
    });
 
    const payload = publisher.sent[0]!.payload as Record<string, unknown>;
    expect(Object.keys(payload).sort()).toEqual([
      "attempt",
      "attempted_at",
      "delivery_id",
      "endpoint_id",
      "environment_id",
      "event_id",
      "latency_ms",
      "outcome",
      "status",
    ]);
    const serialised = JSON.stringify(publisher.sent[0]);
    for (const forbidden of [
      "B2, north ramp",
      "whsec_do_not_publish_this",
      "v1=deadbeef",
      "Bearer",
      "customer.example",
    ]) {
      expect(serialised).not.toContain(forbidden);
    }
  });
});
 
describe("publishAttempt swallows its own failure (contract invariant 4)", () => {
  it("does not throw when the broker refuses", async () => {
    // THE ONE THAT MATTERS. The caller has already committed an outcome, and the
    // dispatcher is waiting for an answer. If this threw, an analytics outage
    // would become a 500 on the dispatch seam, the dispatcher would not
    // acknowledge, and the delivery would be posted to the customer again — a
    // duplicate webhook caused by a metering pipeline. Constitution III names
    // that inversion as a design failure.
    const { logger, lines } = captured();
    await expect(
      publishAttempt(exploding(), logger, base),
    ).resolves.toBeUndefined();
 
    expect(lines).toHaveLength(1);
    expect(lines[0]!["msg"]).toBe("analytics.attempt_publish_failed");
    expect(lines[0]!["delivery_id"]).toBe(base.deliveryId);
    expect(lines[0]!["attempt"]).toBe(3);
  });
 
  it("logs the failure without the payload it was carrying", async () => {
    const { logger, lines } = captured();
    await publishAttempt(exploding(), logger, {
      ...base,
      ...({ payload: { text: "B2, north ramp" } } as unknown as AttemptRecord),
    });
 
    // A log line about a failed publish is the most tempting place in the
    // codebase to dump the thing that failed to publish.
    expect(JSON.stringify(lines)).not.toContain("B2, north ramp");
  });
});

@@ -17,6 +17,8 @@ import {
   type InternalExpandResponse,
 } from "@relay/protocol";
 
+import type { Logger } from "@relay/service-kit";
+
 import { Accepts, CredentialGuard } from "../auth/credential.guard";
 import type { Db } from "../db/client";
 import {
@@ -27,6 +29,9 @@ import {
   replayDeadLetter,
 } from "../db/repository";
 import { ZodValidationPipe } from "../messages/zod-validation.pipe";
+import { LOGGER } from "../logger";
+import { ANALYTICS_PUBLISHER, publishAttempt } from "../webhooks/analytics";
+import type { Publisher } from "../outbox/publisher";
 
 // The dispatcher's only road to state (chapter 3.5, constitution IV).
 //
@@ -45,7 +50,11 @@ import { ZodValidationPipe } from "../messages/zod-validation.pipe";
 @UseGuards(CredentialGuard)
 @Accepts("platform")
 export class DispatchController {
-  constructor(@Inject("DB") private readonly db: Db) {}
+  constructor(
+    @Inject("DB") private readonly db: Db,
+    @Inject(ANALYTICS_PUBLISHER) private readonly analytics: Publisher,
+    @Inject(LOGGER) private readonly logger: Logger,
+  ) {}
 
   /** One event becomes one delivery per matching endpoint — claimed, so it
    * happens exactly once however often the broker redelivers (research R2). */
@@ -100,7 +109,43 @@ export class DispatchController {
         attempt: body.attempt,
         ...(body.status !== undefined ? { status: body.status } : {}),
         ...(body.error !== undefined ? { error: body.error } : {}),
+        latencyMs: body.latency_ms,
       });
+
+      // THE ATTEMPT RECORD, and everything about this call's POSITION is the
+      // decision (chapter 3.6, research R5, constitution III).
+      //
+      // AFTER the transaction, not inside it: `recordAttemptOutcome` has already
+      // returned, so its row locks are released and its work is durable. A publish
+      // inside would hold a lock on the delivery while waiting on a broker, and a
+      // slow analytics path would become a slow delivery path — the coupling
+      // constitution III exists to forbid. `publishAttempt` cannot throw, so this
+      // line cannot change the answer below it either.
+      //
+      // BEFORE the response rather than after it, because there is no "after":
+      // returning ends the request. The cost is that the dispatcher waits for one
+      // publish, which is why that publish has no retry and no timeout of its own
+      // beyond the client's.
+      //
+      // Only when something was RECORDED. A repeat report — the dispatcher
+      // crashed between reporting and acknowledging — changed no row, and
+      // publishing for it would put an attempt on the stream that never happened
+      // (contract invariant 1).
+      if (result.recorded) {
+        await publishAttempt(this.analytics, this.logger, {
+          deliveryId: body.delivery_id,
+          endpointId: result.endpointId,
+          environmentId: result.environmentId,
+          eventId: result.eventId,
+          attempt: result.attempt,
+          ...(body.status !== undefined ? { status: body.status } : {}),
+          ...(body.error !== undefined ? { error: body.error } : {}),
+          latencyMs: body.latency_ms,
+          outcome: result.outcome,
+          attemptedAt: new Date(),
+        });
+      }
+
       return {
         outcome: result.outcome,
         ...(result.nextAttemptAt

@@ -3,6 +3,13 @@ import { Module, Scope } from "@nestjs/common";
 import { MessagesModule } from "../messages/messages.module";
 import { AuthModule } from "../auth/auth.module";
 import { createDb, createPool, type Db } from "../db/client";
+import { LOGGER, apiLogger } from "../logger";
+import {
+  createJetStreamPublisher,
+  ensureAnalyticsStream,
+} from "../outbox/jetstream.publisher";
+import type { Publisher } from "../outbox/publisher";
+import { ANALYTICS_PUBLISHER } from "../webhooks/analytics";
 import { BackfillController } from "./backfill.controller";
 import { InternalController } from "./internal.controller";
 import { DispatchController } from "./dispatch.controller";
@@ -33,6 +40,29 @@ import { SessionController } from "./session.controller";
       useFactory: (): Db => createDb(createPool()),
       scope: Scope.DEFAULT,
     },
+    // Chapter 3.6: the attempt record's way onto the analytical path. Its own
+    // publisher, ensuring its own stream — see ANALYTICS_PUBLISHER's note.
+    //
+    // The connection is LAZY, as every broker client in this workspace is, and
+    // here that property is load-bearing rather than tidy: the api must accept
+    // outcome reports with the broker unreachable. If this connected eagerly, a
+    // dead broker would take the dispatch seam down with it, which is the exact
+    // inversion constitution III forbids.
+    {
+      provide: ANALYTICS_PUBLISHER,
+      useFactory: (): Publisher =>
+        createJetStreamPublisher({ ensure: ensureAnalyticsStream }),
+      scope: Scope.DEFAULT,
+    },
+    // `AppModule` provides this too, but a provider is visible to the module that
+    // declares it and to nothing it imports — so the controllers here would have
+    // nothing to inject. Same factory, so the service name in a log line stays
+    // `api` and the log stream does not sprout a second identity for one line.
+    {
+      provide: LOGGER,
+      useFactory: apiLogger,
+      scope: Scope.DEFAULT,
+    },
   ],
 })
 export class InternalModule {}

attempt 1  at +0h00m00s   (within the 1h window)
attempt 2  at +0h00m01s   (within the 1h window)
attempt 3  at +0h00m06s   (within the 1h window)
attempt 4  at +0h00m36s   (within the 1h window)
attempt 5  at +0h05m36s   (within the 1h window)
attempt 6  at +0h35m36s   (within the 1h window)
attempt 7  at +2h35m36s

@@ -401,10 +401,29 @@ export const webhookEndpoints = pgTable(
     // accepting either is correct throughout (contracts/webhooks.md §Rotation).
     secretPreviousCiphertext: text("secret_previous_ciphertext"),
     secretRotatedAt: timestamp("secret_rotated_at", { withTimezone: true }),
-    // An owner can pause an endpoint. What disables one AUTOMATICALLY after
-    // continuous failure is FR-WHK-07's, in the follow-on chapter — the column
-    // exists now so that chapter adds a rule rather than a migration.
+    // An owner can pause an endpoint. Chapter 3.6 is the follow-on chapter 3.5
+    // named here, and the prediction held: automatic disablement added a rule and
+    // four columns, and did not have to change this one.
     enabled: boolean("enabled").notNull().default(true),
+    // THE FAILURE RUN (chapter 3.6, FR-006). The current unbroken sequence of
+    // failures, and nothing more — history is the attempt event stream, not this.
+    //
+    // Two columns rather than a table, because a run is one row per endpoint BY
+    // DEFINITION: it is the *current* run, and there is only ever one. A table
+    // would need a "which row is current" rule, and that rule is the bug
+    // (research R2). Both null when the endpoint is healthy, and any delivered
+    // outcome sets them back to null.
+    failureRunStartedAt: timestamp("failure_run_started_at", {
+      withTimezone: true,
+    }),
+    failureRunAttempts: integer("failure_run_attempts"),
+    // WHO SWITCHED IT OFF, which `enabled` alone cannot say. Auto-disable sets
+    // `enabled = false` AND stamps these; a customer pausing their own endpoint
+    // sets `enabled = false` and leaves them null. That asymmetry IS FR-009: a
+    // customer can tell a platform disablement from their own by whether the
+    // platform left its fingerprints.
+    disabledAt: timestamp("disabled_at", { withTimezone: true }),
+    disabledReason: text("disabled_reason"),
     createdAt: timestamp("created_at", { withTimezone: true })
       .notNull()
       .defaultNow(),
@@ -417,7 +436,33 @@ export const webhookEndpoints = pgTable(
     // the record of what it was survives its deletion.
     deletedAt: timestamp("deleted_at", { withTimezone: true }),
   },
-  (t) => [index("webhook_endpoints_environment_idx").on(t.environmentId)],
+  (t) => [
+    index("webhook_endpoints_environment_idx").on(t.environmentId),
+    // The SWEEP's only query (chapter 3.6, research R1): enabled endpoints with
+    // an open failure run, so the one that has outrun the hour can be found
+    // without reading every endpoint in the platform. Partial, for the reason
+    // 3.3's outbox index and 3.5's delivery index are partial — a healthy
+    // endpoint has a null run and costs nothing to keep out of it.
+    index("webhook_endpoints_failure_run_idx")
+      .on(t.failureRunStartedAt)
+      .where(sql`${t.enabled} AND ${t.failureRunStartedAt} IS NOT NULL`),
+    // The two halves of a run travel together, and the database says so rather
+    // than four call sites remembering to. A run with a start and no count, or a
+    // count and no start, is not a state this platform has a meaning for — and
+    // `shouldDisable` would read the missing half as zero, which is the shape of
+    // a bug that disables nothing and looks like a policy decision.
+    check(
+      "webhook_endpoints_failure_run_check",
+      sql`(${t.failureRunStartedAt} IS NULL) = (${t.failureRunAttempts} IS NULL)`,
+    ),
+    // Same argument for the disable stamp. FR-009 rests on `disabled_at` being
+    // the platform's fingerprint, so a reason without a timestamp would make the
+    // one distinction a customer needs unreadable.
+    check(
+      "webhook_endpoints_disabled_check",
+      sql`(${t.disabledAt} IS NULL) = (${t.disabledReason} IS NULL)`,
+    ),
+  ],
 );
 
 // The retry schedule — and it is chapter 3.3's outbox with one more column.
@@ -461,6 +506,34 @@ export const webhookDeliveries = pgTable(
     // scheduled. The relay's claim, in the shape `outbox.published_at` has.
     dispatchedAt: timestamp("dispatched_at", { withTimezone: true }),
     state: text("state").notNull().default("pending"),
+    // WHAT THE ENDPOINT ACTUALLY SAID on the most recent attempt (chapter 3.6).
+    //
+    // Chapter 3.5 recorded an attempt by MOVING the delivery — state, attempt,
+    // next_attempt_at — and threw the answer away, which was enough while the
+    // only reader was the retry schedule. Two things here need it back, and
+    // neither can get it from the attempt event: that publish is at-most-once by
+    // design (research R5).
+    //
+    //   * the test event reports what the endpoint answered to a caller who is
+    //     WAITING, and the attempt happens in the dispatcher's process (FR-016);
+    //   * the sweep writes a disablement's last observed error, and the sweep
+    //     fires precisely when no outcome is arriving — that is the whole of
+    //     research R1. Without this it could only write null, and "disabled,
+    //     cause unknown" is the notification a support engineer receives.
+    //
+    // `lastLatencyMs` is the third thing in this chapter to pick `latency_ms` up
+    // off the floor: it has crossed the internal seam on every attempt since 3.5
+    // and been discarded (research R6).
+    lastStatus: integer("last_status"),
+    lastError: text("last_error"),
+    lastLatencyMs: integer("last_latency_ms"),
+    // A TEST EVENT's delivery (chapter 3.6, FR-013). Three decisions branch on
+    // it — no retry schedule, no failure-run update, and delivery even to a
+    // disabled endpoint — which is why it is a column and not a `payload->>'type'`
+    // comparison against a customer-visible document. It also keeps the marker
+    // for the RECIPIENT (the envelope's `type` and `test`) separate from the
+    // marker for the PLATFORM, two audiences that happen to agree today.
+    synthetic: boolean("synthetic").notNull().default(false),
     createdAt: timestamp("created_at", { withTimezone: true })
       .notNull()
       .defaultNow(),
@@ -513,3 +586,62 @@ export const webhookDeadLetters = pgTable(
   },
   (t) => [index("webhook_dead_letters_environment_idx").on(t.environmentId)],
 );
+
+// DECISION (chapter 3.6, FR-011, research R7): one row per automatic
+// disablement — an OUTBOUND OBLIGATION the platform has not yet met.
+//
+// `deliveredAt` is the honest column, and it exists in this chapter solely in
+// order to be null. FR-WHK-07 asks for the endpoint to be disabled "and the
+// organisation notified by email", and this platform has no email transport of
+// any kind. Chapter 3.7 needs the same transport for quotas, so building one here
+// would mean building it for its second consumer first.
+//
+// A schema that recorded only the disablement would let a future reader believe
+// the requirement was finished. This one says, in a column, which half is
+// missing.
+//
+// Why a table rather than more columns on the endpoint: the endpoint gains
+// `disabledAt` and `disabledReason` regardless, because FR-009 needs a customer
+// to tell a platform disablement from their own. The notification is a different
+// kind of thing — a record with a lifecycle, which is what `deliveredAt` makes
+// visible.
+export const webhookDisableNotifications = pgTable(
+  "webhook_disable_notifications",
+  {
+    id: uuid("id").primaryKey(),
+    environmentId: uuid("environment_id")
+      .notNull()
+      .references(() => environments.id),
+    // DENORMALISED on purpose. The joins are available —
+    // environments.application_id → applications.organisation_id — so storing it
+    // looks redundant. It is stored because this row records an obligation AS IT
+    // STOOD when the endpoint was disabled, and an application moving between
+    // organisations later must not silently retarget a notification that was
+    // already owed to somebody else.
+    organisationId: uuid("organisation_id")
+      .notNull()
+      .references(() => organisations.id),
+    endpointId: uuid("endpoint_id")
+      .notNull()
+      .references(() => webhookEndpoints.id),
+    disabledAt: timestamp("disabled_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+    // The window that triggered it, copied rather than referenced: the endpoint's
+    // own run columns are cleared the moment a customer re-enables it, and a
+    // notification that lost its evidence on re-enablement would be unanswerable
+    // by the time anybody read it.
+    runStartedAt: timestamp("run_started_at", { withTimezone: true }).notNull(),
+    runAttempts: integer("run_attempts").notNull(),
+    lastStatus: integer("last_status"),
+    lastError: text("last_error"),
+    // NULL THROUGHOUT THIS CHAPTER. Set by whatever chapter builds a transport.
+    deliveredAt: timestamp("delivered_at", { withTimezone: true }),
+  },
+  (t) => [
+    // Nothing beyond the primary key and the tenant. Volume is one row per
+    // endpoint per outage, so an index for any other access pattern would be
+    // guessing at a query nobody has written.
+    index("webhook_disable_notifications_environment_idx").on(t.environmentId),
+  ],
+);

-- Chapter 3.6 — the failure run, automatic disablement, and its notification.
--
-- REVIEW DISPOSITION: drizzle-kit generated this from schema.ts and it was read
-- line by line before being applied (the ADR-16 workflow, and chapter 2.1's rule
-- after a generated migration was once applied unread). Nothing was rewritten;
-- two CHECK constraints were added to schema.ts and the file regenerated. Six
-- things were checked rather than assumed:
--
--   * EVERY NEW COLUMN ON webhook_endpoints IS NULLABLE WITH NO DEFAULT, and that
--     is the single most important line in this file. A default of `now()` on
--     failure_run_started_at would have opened a failure run for every endpoint
--     in the platform at deploy time, and an hour later the sweep would have
--     disabled all of them. Nullable-and-null means every existing endpoint is
--     healthy the moment this applies, which is true;
--   * `synthetic boolean DEFAULT false NOT NULL` is the one non-null addition,
--     and the backfill it implies is correct: every delivery that already exists
--     was a real event, not a test;
--   * webhook_disable_notifications carries environment_id NOT NULL with a
--     foreign key (constitution I). organisation_id is NOT NULL too and is
--     denormalised on purpose — the row records an obligation as it stood, and an
--     application moving between organisations must not retarget a notification
--     already owed to somebody else;
--   * all three foreign keys are ON DELETE NO ACTION, the choice 3.5 made and
--     for its reason: deletion is soft, and a cascade would erase records the
--     platform is required to keep;
--   * webhook_endpoints_failure_run_idx is PARTIAL, on failure_run_started_at
--     WHERE enabled AND the run is open. It covers the sweep's only query and
--     nothing else, so a healthy endpoint costs nothing to keep out of it — the
--     same shape as 3.3's outbox index and 3.5's delivery index;
--   * the two CHECK constraints make the run's halves and the disable stamp's
--     halves travel together. Neither was in the generated output, and neither is
--     decoration: `shouldDisable` reads a missing attempt count as zero, which
--     would disable nothing and look like a policy decision rather than a bug.
--
-- No statement here is destructive, and there is no down path (forward-only).
 
CREATE TABLE "webhook_disable_notifications" (
	"id" uuid PRIMARY KEY NOT NULL,
	"environment_id" uuid NOT NULL,
	"organisation_id" uuid NOT NULL,
	"endpoint_id" uuid NOT NULL,
	"disabled_at" timestamp with time zone DEFAULT now() NOT NULL,
	"run_started_at" timestamp with time zone NOT NULL,
	"run_attempts" integer NOT NULL,
	"last_status" integer,
	"last_error" text,
	"delivered_at" timestamp with time zone
);
--> statement-breakpoint
ALTER TABLE "webhook_deliveries" ADD COLUMN "last_status" integer;--> statement-breakpoint
ALTER TABLE "webhook_deliveries" ADD COLUMN "last_error" text;--> statement-breakpoint
ALTER TABLE "webhook_deliveries" ADD COLUMN "last_latency_ms" integer;--> statement-breakpoint
ALTER TABLE "webhook_deliveries" ADD COLUMN "synthetic" boolean DEFAULT false NOT NULL;--> statement-breakpoint
ALTER TABLE "webhook_endpoints" ADD COLUMN "failure_run_started_at" timestamp with time zone;--> statement-breakpoint
ALTER TABLE "webhook_endpoints" ADD COLUMN "failure_run_attempts" integer;--> statement-breakpoint
ALTER TABLE "webhook_endpoints" ADD COLUMN "disabled_at" timestamp with time zone;--> statement-breakpoint
ALTER TABLE "webhook_endpoints" ADD COLUMN "disabled_reason" text;--> statement-breakpoint
ALTER TABLE "webhook_disable_notifications" ADD CONSTRAINT "webhook_disable_notifications_environment_id_environments_id_fk" FOREIGN KEY ("environment_id") REFERENCES "public"."environments"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "webhook_disable_notifications" ADD CONSTRAINT "webhook_disable_notifications_organisation_id_organisations_id_fk" FOREIGN KEY ("organisation_id") REFERENCES "public"."organisations"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
ALTER TABLE "webhook_disable_notifications" ADD CONSTRAINT "webhook_disable_notifications_endpoint_id_webhook_endpoints_id_fk" FOREIGN KEY ("endpoint_id") REFERENCES "public"."webhook_endpoints"("id") ON DELETE no action ON UPDATE no action;--> statement-breakpoint
CREATE INDEX "webhook_disable_notifications_environment_idx" ON "webhook_disable_notifications" USING btree ("environment_id");--> statement-breakpoint
CREATE INDEX "webhook_endpoints_failure_run_idx" ON "webhook_endpoints" USING btree ("failure_run_started_at") WHERE "webhook_endpoints"."enabled" AND "webhook_endpoints"."failure_run_started_at" IS NOT NULL;--> statement-breakpoint
ALTER TABLE "webhook_endpoints" ADD CONSTRAINT "webhook_endpoints_failure_run_check" CHECK (("webhook_endpoints"."failure_run_started_at" IS NULL) = ("webhook_endpoints"."failure_run_attempts" IS NULL));--> statement-breakpoint
ALTER TABLE "webhook_endpoints" ADD CONSTRAINT "webhook_endpoints_disabled_check" CHECK (("webhook_endpoints"."disabled_at" IS NULL) = ("webhook_endpoints"."disabled_reason" IS NULL));

// When to stop trying (chapter 3.6, FR-007).
//
// This file is the chapter's only arithmetic, and it is separated from both places
// that call it deliberately. One trigger runs inside a database transaction and
// the other runs inside a background loop; a policy living in either would be
// testable only with a database, a clock and a broker, and "should this endpoint be
// switched off" is a question that needs none of the three.
//
// The two numbers are FIXED, not configuration, and that is FR-007's own wording
// rather than an omission. An operator who can lower the floor to one has an
// operator who can disable a paying customer's endpoint on a single bad response —
// and the failure mode of configuration here is silent, since nobody notices a
// threshold that is too aggressive until the endpoints are already off.
 
/** ONE HOUR, from FR-WHK-07's "failing continuously for more than an hour". */
export const DISABLE_AFTER_MS = 60 * 60 * 1_000;
 
/** FIVE attempts, and the number comes from measurement rather than taste
 * (research R3).
 *
 * The hour alone is not enough. Chapter 3.5's schedule reaches a two-hour gap
 * between attempts, so ONE failure followed by silence satisfies "failing for
 * more than an hour" with a single data point — and disabling on one bad response
 * is what the floor exists to prevent.
 *
 * Five is the largest floor a single failing delivery can still clear inside the
 * window. Computed against `RETRY_TIERS_MS = [0, 1s, 5s, 30s, 5min, 30min, 2h]`,
 * measured from the first attempt:
 *
 *     attempt 5  at +5m36s    inside the hour
 *     attempt 6  at +35m36s   inside the hour
 *     attempt 7  at +2h35m36s OUTSIDE it
 *
 * So a floor of 5 or 6 is reachable by one delivery and a floor of 7 is not: set
 * it to 7 and a single-delivery run could never trigger, which would leave the
 * quietest endpoints — the ones least likely to be watched — enabled and failing
 * for ever. Five leaves a margin of one.
 *
 * Below 5 the floor stops doing its job: two or three failures are reachable in
 * the first six seconds, and six seconds of trouble is a blip rather than an
 * outage. */
export const DISABLE_MIN_ATTEMPTS = 5;
 
export interface FailureRun {
  /** When the current unbroken run of failures began. Null when the endpoint is
   * healthy — any success clears it. */
  runStartedAt: Date | null;
  /** How many failures the run contains. Null when healthy. */
  runAttempts: number | null;
  /** Passed in rather than read, so this function has no clock of its own. Every
   * caller has one already: the on-outcome trigger is inside a transaction that
   * knows `now()`, and the sweep is a loop that just woke up. */
  now: Date;
}
 
/** Should this endpoint be switched off?
 *
 * Both conditions, never either: longer than an hour AND at least five failures.
 * The run must also exist — a healthy endpoint is not a candidate, and the two
 * nulls are how "healthy" is spelled. */
export function shouldDisable({
  runStartedAt,
  runAttempts,
  now,
}: FailureRun): boolean {
  if (runStartedAt === null || runAttempts === null) return false;
  if (runAttempts < DISABLE_MIN_ATTEMPTS) return false;
  return runWindowMs({ runStartedAt, now }) > DISABLE_AFTER_MS;
}
 
/** How long the run has been going, and never a negative number.
 *
 * The clamp is not defensive padding. A clock that moves backwards — an NTP
 * correction, a container resumed from a snapshot, a replica whose time differs
 * from the writer's — would otherwise produce a negative window, and a negative
 * window compared against a positive threshold reads as "not yet", which is the
 * safe direction only by accident. Clamping makes the answer "no time has passed"
 * rather than "time has run backwards", which is a claim the rest of the code can
 * reason about. Spec edge case, stated there before it was written here. */
export function runWindowMs({
  runStartedAt,
  now,
}: {
  runStartedAt: Date;
  now: Date;
}): number {
  return Math.max(0, now.getTime() - runStartedAt.getTime());
}
 
/** The sentence a customer reads, and the reason FR-009 asks for one.
 *
 * "Disabled" on its own invites a support ticket. This says what the platform saw,
 * over how long, and what the endpoint last answered — enough for the customer to
 * recognise their own outage without being told to check a dashboard that does not
 * exist yet. */
export function disableReason({
  runAttempts,
  windowMs,
  lastStatus,
  lastError,
}: {
  runAttempts: number;
  windowMs: number;
  lastStatus: number | null;
  lastError: string | null;
}): string {
  const answered =
    lastStatus !== null
      ? `last status ${lastStatus}`
      : lastError !== null
        ? // Bounded, because this string goes in a column a customer reads and the
          // error it quotes came from a machine we do not own. The seam already
          // caps the error at 2000 characters; a reason is a sentence, not a log.
          `no response (${lastError.slice(0, 200)})`
        : "no response";
  return `${runAttempts} consecutive failures over ${humanDuration(windowMs)}; ${answered}`;
}
 
/** `1h04m`. Minutes, because the threshold is an hour and seconds would be noise
 * — and because a reason that said `3862000ms` would be a message written for the
 * platform rather than for the person reading it. */
function humanDuration(ms: number): string {
  const minutes = Math.floor(ms / 60_000);
  const hours = Math.floor(minutes / 60);
  return `${hours}h${String(minutes % 60).padStart(2, "0")}m`;
}

import { describe, expect, it } from "vitest";
 
import { RETRY_TIERS_MS } from "./schedule";
import {
  DISABLE_AFTER_MS,
  DISABLE_MIN_ATTEMPTS,
  disableReason,
  runWindowMs,
  shouldDisable,
} from "./disable";
 
// The decision to switch off a paying customer's endpoint (chapter 3.6).
//
// Pinned at 100% branches in `vitest.coverage.config.mts`, because constitution VI
// names idempotency logic and this is the predicate the at-most-once disable rests
// on. It is also the only arithmetic in the chapter, and arithmetic is the kind of
// thing that is obviously right until somebody computes it.
 
const T0 = new Date("2026-08-18T08:00:00.000Z");
const after = (ms: number) => new Date(T0.getTime() + ms);
 
const MINUTE = 60_000;
const HOUR = 60 * MINUTE;
 
describe("shouldDisable requires an hour AND five failures", () => {
  it("disables a run past the hour with five failures", () => {
    expect(
      shouldDisable({
        runStartedAt: T0,
        runAttempts: 5,
        now: after(HOUR + MINUTE),
      }),
    ).toBe(true);
  });
 
  it("does NOT disable four failures, however long the run", () => {
    // The floor, doing its job. A whole day of a single failing delivery that
    // only ever managed four attempts is not five failures.
    expect(
      shouldDisable({
        runStartedAt: T0,
        runAttempts: DISABLE_MIN_ATTEMPTS - 1,
        now: after(24 * HOUR),
      }),
    ).toBe(false);
  });
 
  it("does NOT disable five failures inside the hour", () => {
    // The window, doing its job. Chapter 3.5's first five attempts all land
    // within 5m36s, so without the hour a single blip would disable an endpoint
    // six seconds into an outage that might already be over.
    expect(
      shouldDisable({
        runStartedAt: T0,
        runAttempts: 9,
        now: after(HOUR - 1),
      }),
    ).toBe(false);
  });
 
  it("does not disable at EXACTLY the hour — FR-007 says more than", () => {
    // A boundary worth pinning rather than leaving to whichever comparison
    // somebody typed. "More than an hour" is `>`, and one millisecond later is
    // the first moment that is true.
    const attempts = DISABLE_MIN_ATTEMPTS;
    expect(
      shouldDisable({ runStartedAt: T0, runAttempts: attempts, now: after(HOUR) }),
    ).toBe(false);
    expect(
      shouldDisable({
        runStartedAt: T0,
        runAttempts: attempts,
        now: after(HOUR + 1),
      }),
    ).toBe(true);
  });
 
  it("treats a healthy endpoint as no candidate at all", () => {
    // Both nulls are how "healthy" is spelled, and each null is checked because a
    // schema CHECK guarantees they agree but this function must not depend on the
    // schema to be correct.
    const now = after(48 * HOUR);
    expect(shouldDisable({ runStartedAt: null, runAttempts: null, now })).toBe(false);
    expect(shouldDisable({ runStartedAt: null, runAttempts: 99, now })).toBe(false);
    expect(shouldDisable({ runStartedAt: T0, runAttempts: null, now })).toBe(false);
  });
 
  it("does not disable a run of zero length", () => {
    // The first failure opens the run and is evaluated in the same transaction, so
    // this case happens on every single failure. It must be cheap and it must be
    // false.
    expect(shouldDisable({ runStartedAt: T0, runAttempts: 1, now: T0 })).toBe(false);
  });
 
  it("yields no negative window when the clock moves backwards", () => {
    // An NTP correction, a container resumed from a snapshot, a replica whose
    // clock differs from the writer's. A negative window would compare as "not
    // yet" — the safe direction, but only by accident, and this makes it a
    // property rather than luck. Spec edge case.
    const backwards = new Date(T0.getTime() - 5 * MINUTE);
    expect(runWindowMs({ runStartedAt: T0, now: backwards })).toBe(0);
    expect(
      shouldDisable({ runStartedAt: T0, runAttempts: 99, now: backwards }),
    ).toBe(false);
  });
});
 
describe("the floor is reachable by one failing delivery (research R3)", () => {
  // THE MEASUREMENT THAT CHOSE THE NUMBER, re-derived here from the real tier
  // table rather than restated. If a later chapter changes the schedule, this
  // fails and the floor gets re-examined — which is the whole reason it is
  // computed from `RETRY_TIERS_MS` instead of written down.
  const attemptOffsets = RETRY_TIERS_MS.reduce<number[]>(
    (acc, tier) => [...acc, (acc.at(-1) ?? 0) + tier],
    [],
  );
 
  it("reaches the floor inside the hour, with a margin of one attempt", () => {
    // Attempt N lands at offset N-1. The floor is cleared when the attempt at
    // index DISABLE_MIN_ATTEMPTS - 1 is still inside the window.
    const atFloor = attemptOffsets[DISABLE_MIN_ATTEMPTS - 1]!;
    expect(atFloor).toBeLessThan(DISABLE_AFTER_MS);
 
    // The margin: one more attempt also lands inside the hour.
    const oneMore = attemptOffsets[DISABLE_MIN_ATTEMPTS]!;
    expect(oneMore).toBeLessThan(DISABLE_AFTER_MS);
  });
 
  it("would be unreachable at a floor of seven, which is why it is not seven", () => {
    // The last attempt falls at +2h35m36s. A floor equal to the attempt count
    // would mean a single failing delivery never disables its endpoint, and the
    // quiet endpoint stays broken for ever — the exact failure research R1 found.
    const lastAttempt = attemptOffsets.at(-1)!;
    expect(lastAttempt).toBeGreaterThan(DISABLE_AFTER_MS);
    expect(attemptOffsets.length).toBeGreaterThan(DISABLE_MIN_ATTEMPTS);
  });
 
  it("is above the blip threshold: three failures arrive within seconds", () => {
    const third = attemptOffsets[2]!;
    expect(third).toBeLessThan(10_000);
    expect(DISABLE_MIN_ATTEMPTS).toBeGreaterThan(3);
  });
});
 
describe("disableReason says what happened", () => {
  it("names the count, the window and the last status", () => {
    expect(
      disableReason({
        runAttempts: 6,
        windowMs: HOUR + 4 * MINUTE,
        lastStatus: 503,
        lastError: null,
      }),
    ).toBe("6 consecutive failures over 1h04m; last status 503");
  });
 
  it("names the error when nothing answered", () => {
    expect(
      disableReason({
        runAttempts: 5,
        windowMs: 2 * HOUR,
        lastStatus: null,
        lastError: "timeout after 10000ms",
      }),
    ).toBe("5 consecutive failures over 2h00m; no response (timeout after 10000ms)");
  });
 
  it("says `no response` when there is neither", () => {
    // The sweep's case: it disables an endpoint whose failures stopped arriving,
    // and the last delivery may predate the columns that would have recorded them.
    expect(
      disableReason({
        runAttempts: 5,
        windowMs: HOUR + MINUTE,
        lastStatus: null,
        lastError: null,
      }),
    ).toBe("5 consecutive failures over 1h01m; no response");
  });
 
  it("bounds an enormous error rather than quoting all of it", () => {
    // Spec edge case: "an attempt whose error message is enormous, or contains a
    // customer's payload". This column is read by a person; the full error is
    // already capped at 2000 characters by the seam, and a reason is a sentence.
    const reason = disableReason({
      runAttempts: 5,
      windowMs: 2 * HOUR,
      lastStatus: null,
      lastError: "x".repeat(2000),
    });
    expect(reason.length).toBeLessThan(300);
    expect(reason).toContain("x".repeat(200));
    expect(reason).not.toContain("x".repeat(201));
  });
});

@@ -18,10 +18,18 @@ import {
   users,
   webhookDeadLetters,
   webhookDeliveries,
+  webhookDisableNotifications,
   webhookEndpoints,
 } from "./schema";
 import { messageCreatedEvent } from "../outbox/event";
 import { nextAttemptAt } from "../webhooks/schedule";
+import {
+  DISABLE_AFTER_MS,
+  DISABLE_MIN_ATTEMPTS,
+  disableReason,
+  runWindowMs,
+  shouldDisable,
+} from "../webhooks/disable";
 import { activeSigningSecrets } from "../webhooks/secret";
 import {
   mintApiKey,
@@ -393,6 +401,229 @@ export type DeliveryOutcome = "delivered" | "rescheduled" | "dead_lettered";
  * itself may duplicate, and the customer absorbs it on the event id, but the
  * SCHEDULE must not advance twice for one attempt or the tiers would collapse.
  */
+/** Open, extend or clear an endpoint's failure run, and disable it if the run has
+ * gone on long enough (chapter 3.6, FR-006, FR-007).
+ *
+ * Runs INSIDE the transaction that records the outcome, and takes
+ * `SELECT … FOR UPDATE` on the endpoint row.
+ *
+ * WHAT THAT LOCK IS ACTUALLY FOR, corrected by the sabotage battery. The comment
+ * here used to say it was "the whole of FR-008's concurrency story" — that without
+ * it, two dispatcher instances reporting outcomes for the same endpoint at the same
+ * moment would both decide to disable and produce two notifications. Dropping the
+ * lock and running the whole suite produced 46 passes. The claim was wrong: the
+ * `enabled = true` predicate in `disableEndpoint`'s update is sufficient for
+ * at-most-once on its own, and the lock was being credited for the predicate's
+ * work.
+ *
+ * The lock protects the COUNTER. Under READ COMMITTED both transactions read
+ * `runAttempts = 4`, both compute 5, and the second UPDATE waits for the first and
+ * then overwrites it with 5. That is a lost update, and the run undercounts by one
+ * per collision. Nothing reports it, nothing fails, and the endpoint quietly needs
+ * an extra failure to reach FR-007's floor of five — a threshold harder to reach
+ * than the requirement says, which is the kind of defect that survives for years.
+ *
+ * The lock is per endpoint and held for one small update, so two customers never
+ * contend (research R2).
+ *
+ * Lock ORDER is delivery-then-endpoint, everywhere, without exception. The caller
+ * has already locked the delivery row; anything that took these two in the other
+ * order would deadlock against this under concurrency, and a deadlock found in
+ * production is a deadlock found by a customer.
+ *
+ * Returns what it did, so the caller can report it and the tests can assert on it
+ * without reading the row back. */
+async function applyFailureRun(
+  tx: Db,
+  input: {
+    endpointId: string;
+    /** Did this attempt fail? A success clears the run outright. */
+    failed: boolean;
+    status: number | null;
+    error: string | null;
+  },
+): Promise<{ disabled: boolean }> {
+  const [endpoint] = await tx
+    .select({
+      id: webhookEndpoints.id,
+      environmentId: webhookEndpoints.environmentId,
+      enabled: webhookEndpoints.enabled,
+      runStartedAt: webhookEndpoints.failureRunStartedAt,
+      runAttempts: webhookEndpoints.failureRunAttempts,
+    })
+    .from(webhookEndpoints)
+    .where(eq(webhookEndpoints.id, input.endpointId))
+    .for("update");
+
+  // The endpoint was deleted between the delivery being claimed and its outcome
+  // being reported. Nothing to track and nothing to disable; the delivery's own
+  // state has already been settled by the caller.
+  if (!endpoint) return { disabled: false };
+
+  if (!input.failed) {
+    // ANY SUCCESS CLEARS THE RUN (FR-006). This is why an endpoint that succeeds
+    // once an hour is never disabled, and it is deliberately generous: a platform
+    // that switches off endpoints which sometimes work is a worse failure than one
+    // that keeps trying. Written unconditionally rather than behind a "was there a
+    // run" check — the update is the same cost either way, and the check is one
+    // more thing to get wrong.
+    await tx
+      .update(webhookEndpoints)
+      .set({ failureRunStartedAt: null, failureRunAttempts: null })
+      .where(eq(webhookEndpoints.id, endpoint.id));
+    return { disabled: false };
+  }
+
+  // Read the clock ONCE, from the database, so the window and the timestamps it is
+  // compared against come from the same source. An api process whose clock differs
+  // from Postgres's would otherwise measure a window against somebody else's idea
+  // of now.
+  // Coerced, not trusted. A raw `execute` hands back whatever the driver made of
+  // the column, and for `timestamptz` that is a string here rather than a Date —
+  // which drizzle then refuses to write back, with `value.toISOString is not a
+  // function` from deep inside its timestamp mapper and no mention of this line.
+  const [clock] = (await tx.execute(sql`SELECT now() AS now`))
+    .rows as { now: string | Date }[];
+  const now = new Date(clock!.now);
+
+  const runStartedAt = endpoint.runStartedAt ?? now;
+  const runAttempts = (endpoint.runAttempts ?? 0) + 1;
+
+  await tx
+    .update(webhookEndpoints)
+    .set({ failureRunStartedAt: runStartedAt, failureRunAttempts: runAttempts })
+    .where(eq(webhookEndpoints.id, endpoint.id));
+
+  if (!shouldDisable({ runStartedAt, runAttempts, now })) {
+    return { disabled: false };
+  }
+
+  return disableEndpoint(tx, {
+    endpointId: endpoint.id,
+    environmentId: endpoint.environmentId,
+    runStartedAt,
+    runAttempts,
+    now,
+    status: input.status,
+    error: input.error,
+  });
+}
+
+/** Switch an endpoint off, once, and record the obligation to tell somebody
+ * (FR-007, FR-008, FR-011).
+ *
+ * AT MOST ONCE PER RUN, and it is the STATEMENT that enforces it rather than a
+ * check somebody has to remember to write: the update carries `enabled = true` in
+ * its predicate, so a second disable matches zero rows and the notification below
+ * is never reached. Both triggers call this, so both inherit the property — which
+ * is contract invariant 12, and the reason there can safely be two of them. */
+async function disableEndpoint(
+  tx: Db,
+  input: {
+    endpointId: string;
+    environmentId: string;
+    runStartedAt: Date;
+    runAttempts: number;
+    now: Date;
+    status: number | null;
+    error: string | null;
+  },
+): Promise<{ disabled: boolean }> {
+  const windowMs = runWindowMs({ runStartedAt: input.runStartedAt, now: input.now });
+  const reason = disableReason({
+    runAttempts: input.runAttempts,
+    windowMs,
+    lastStatus: input.status,
+    lastError: input.error,
+  });
+
+  const updated = (await tx.execute(sql`
+    UPDATE webhook_endpoints
+       SET enabled = false,
+           disabled_at = ${input.now},
+           disabled_reason = ${reason}
+     WHERE id = ${input.endpointId}
+       AND enabled = true
+     RETURNING id`)) as unknown as { rows: { id: string }[] };
+
+  // Zero rows means somebody else got there first — a concurrent outcome report,
+  // or the sweep, or a customer who happened to pause the endpoint themselves. In
+  // every case the answer is the same: this call disabled nothing, so it owes no
+  // notification.
+  if (updated.rows.length === 0) return { disabled: false };
+
+  // The organisation is resolved HERE, at write time, through the two hops that
+  // already exist: environments.application_id → applications.organisation_id. It
+  // is stored rather than joined for later because this row records an obligation
+  // AS IT STOOD, and an application moving between organisations afterwards must
+  // not silently retarget a notification already owed to somebody else.
+  const [owner] = await tx
+    .select({ organisationId: applications.organisationId })
+    .from(environments)
+    .innerJoin(applications, eq(environments.applicationId, applications.id))
+    .where(eq(environments.id, input.environmentId));
+
+  await tx.insert(webhookDisableNotifications).values({
+    id: randomUUID(),
+    environmentId: input.environmentId,
+    organisationId: owner!.organisationId,
+    endpointId: input.endpointId,
+    disabledAt: input.now,
+    runStartedAt: input.runStartedAt,
+    runAttempts: input.runAttempts,
+    lastStatus: input.status,
+    lastError: input.error,
+    // NOT SET, and that is the point. FR-WHK-07 asks for the organisation to be
+    // notified by email and this platform has no email; `delivered_at` exists in
+    // order to be null until a transport does.
+  });
+
+  return { disabled: true };
+}
+
+/** What one recorded outcome yields its caller.
+ *
+ * The four identifiers are here because the ATTEMPT EVENT needs them and the
+ * dispatcher does not hold them: it knows a delivery id, a status and a latency,
+ * and nothing about which environment or event that delivery belongs to. Reading
+ * them back out with a second query would be a second query for data this
+ * transaction already had in hand.
+ *
+ * `recorded` is the field the analytics publish is conditional on. See the
+ * idempotent-replay branch below. */
+export interface RecordedOutcome {
+  outcome: DeliveryOutcome;
+  nextAttemptAt: Date | null;
+  /** True when this call actually moved the delivery. False when it recognised a
+   * report it had already processed. */
+  recorded: boolean;
+  endpointId: string;
+  environmentId: string;
+  eventId: string;
+  attempt: number;
+  synthetic: boolean;
+}
+
+/** The identifiers, lifted off the locked row so both return paths agree. */
+function identity(delivery: {
+  endpointId: string;
+  environmentId: string;
+  eventId: string;
+  attempt: number;
+  synthetic: boolean;
+}): Pick<
+  RecordedOutcome,
+  "endpointId" | "environmentId" | "eventId" | "attempt" | "synthetic"
+> {
+  return {
+    endpointId: delivery.endpointId,
+    environmentId: delivery.environmentId,
+    eventId: delivery.eventId,
+    attempt: delivery.attempt,
+    synthetic: delivery.synthetic,
+  };
+}
+
 export async function recordAttemptOutcome(
   db: Db,
   input: {
@@ -400,8 +631,13 @@ export async function recordAttemptOutcome(
     attempt: number;
     status?: number;
     error?: string;
+    /** How long the customer took to answer. Carried across the internal seam on
+     * every attempt since chapter 3.5 and discarded until 3.6 wanted it (research
+     * R6). Optional only so that callers written before it existed still compile;
+     * every real caller has it. */
+    latencyMs?: number;
   },
-): Promise<{ outcome: DeliveryOutcome; nextAttemptAt: Date | null }> {
+): Promise<RecordedOutcome> {
   return db.transaction(async (tx) => {
     const [delivery] = await tx
       .select({
@@ -414,6 +650,7 @@ export async function recordAttemptOutcome(
         state: webhookDeliveries.state,
         nextAttemptAt: webhookDeliveries.nextAttemptAt,
         dispatchedAt: webhookDeliveries.dispatchedAt,
+        synthetic: webhookDeliveries.synthetic,
       })
       .from(webhookDeliveries)
       .where(eq(webhookDeliveries.id, input.deliveryId))
@@ -434,21 +671,65 @@ export async function recordAttemptOutcome(
               : ("rescheduled" as const),
         nextAttemptAt:
           delivery.state === "pending" ? delivery.nextAttemptAt : null,
+        // NOT RECORDED. This branch changed no row, so nothing new happened and
+        // the caller must not publish an attempt event for it (contract invariant
+        // 1). The dispatcher posts, reports, then acknowledges, so a crash in the
+        // last gap makes a second report ordinary rather than exceptional — and
+        // nothing on the analytical path deduplicates, so publishing here would
+        // put a retry that never happened on a customer's dashboard.
+        recorded: false,
+        ...identity(delivery),
       };
     }
 
     const succeeded =
       input.status !== undefined && input.status >= 200 && input.status < 300;
 
+    // WHAT THE ENDPOINT SAID, kept on every recorded attempt whichever branch
+    // follows. Two things need it and neither can get it from the attempt event,
+    // whose publish is at-most-once by design: the test event answers a caller who
+    // is waiting (FR-016), and the sweep names a disablement's last error at a
+    // moment when no outcome is arriving (FR-009, research R1).
+    const lastOutcome = {
+      lastStatus: input.status ?? null,
+      lastError: input.error ?? null,
+      lastLatencyMs: input.latencyMs ?? null,
+    };
+
+    // THE FAILURE RUN, and a test event is exempt from it (contract invariant 13,
+    // research R8). A test event is a diagnostic rather than traffic: letting a
+    // failed test push an endpoint toward disablement would punish a customer for
+    // checking, and letting a successful one CLEAR the run would let a customer
+    // mask a real outage by testing until it passed. Both directions matter, which
+    // is why the exemption is on the call and not inside it.
+    if (!delivery.synthetic) {
+      await applyFailureRun(tx, {
+        endpointId: delivery.endpointId,
+        failed: !succeeded,
+        status: input.status ?? null,
+        error: input.error ?? null,
+      });
+    }
+
     if (succeeded) {
       await tx
         .update(webhookDeliveries)
-        .set({ state: "delivered", dispatchedAt: null })
+        .set({ state: "delivered", dispatchedAt: null, ...lastOutcome })
         .where(eq(webhookDeliveries.id, delivery.id));
-      return { outcome: "delivered" as const, nextAttemptAt: null };
+      return {
+        outcome: "delivered" as const,
+        nextAttemptAt: null,
+        recorded: true,
+        ...identity(delivery),
+      };
     }
 
-    const next = nextAttemptAt(delivery.attempt + 1);
+    // A TEST EVENT GETS ONE ATTEMPT AND NO SCHEDULE (research R8, FR-013). A
+    // caller is standing at their terminal waiting for the answer; a test that
+    // quietly retried for two hours would report a stale one, and a test event
+    // that kept coming back would be indistinguishable from real traffic to the
+    // customer trying to read their logs.
+    const next = delivery.synthetic ? null : nextAttemptAt(delivery.attempt + 1);
     if (next) {
       await tx
         .update(webhookDeliveries)
@@ -457,14 +738,40 @@ export async function recordAttemptOutcome(
           nextAttemptAt: next,
           // Cleared so the relay can pick it up again when it falls due.
           dispatchedAt: null,
+          ...lastOutcome,
         })
         .where(eq(webhookDeliveries.id, delivery.id));
-      return { outcome: "rescheduled" as const, nextAttemptAt: next };
+      return {
+        outcome: "rescheduled" as const,
+        nextAttemptAt: next,
+        recorded: true,
+        ...identity(delivery),
+      };
     }
 
     // Attempts exhausted. The dead letter and the state change commit together —
     // a delivery marked dead with no dead letter behind it would be a failure
     // with no record, which is exactly what FR-WHK-04's seven days are for.
+    // …unless it was a test. A dead letter is a customer-visible record retained
+    // for seven days and replayable (FR-WHK-04), and a test event is a diagnostic
+    // the customer asked for and already has the answer to. Writing one would put
+    // synthetic traffic in the store whose whole purpose is real traffic that
+    // failed to leave, and offer an operator a "replay" button that re-sends a
+    // test. The delivery is still marked dead, and the record of what happened is
+    // the response the caller received plus `last_status` on the row.
+    if (delivery.synthetic) {
+      await tx
+        .update(webhookDeliveries)
+        .set({ state: "dead", dispatchedAt: null, ...lastOutcome })
+        .where(eq(webhookDeliveries.id, delivery.id));
+      return {
+        outcome: "dead_lettered" as const,
+        nextAttemptAt: null,
+        recorded: true,
+        ...identity(delivery),
+      };
+    }
+
     await tx.insert(webhookDeadLetters).values({
       id: randomUUID(),
       environmentId: delivery.environmentId,
@@ -477,9 +784,226 @@ export async function recordAttemptOutcome(
     });
     await tx
       .update(webhookDeliveries)
-      .set({ state: "dead", dispatchedAt: null })
+      .set({ state: "dead", dispatchedAt: null, ...lastOutcome })
       .where(eq(webhookDeliveries.id, delivery.id));
-    return { outcome: "dead_lettered" as const, nextAttemptAt: null };
+    return {
+      outcome: "dead_lettered" as const,
+      nextAttemptAt: null,
+      recorded: true,
+      ...identity(delivery),
+    };
+  });
+}
+
+/** Create the one delivery a test event needs (chapter 3.6, FR-013, research R8).
+ *
+ * THREE DELIBERATE DEVIATIONS from `expandEventToDeliveries`, each with a reason,
+ * and they are the whole difference between a test event and a real one:
+ *
+ *   * ONE ENDPOINT, named by the caller, rather than every endpoint whose
+ *     subscription matches. A test is aimed. Fanning it out would send every other
+ *     endpoint in the environment a surprise event they did not ask for.
+ *   * DELIVERED EVEN WHEN DISABLED — `enabled` is not in the predicate below.
+ *     Testing is how a customer establishes their endpoint is fixed BEFORE
+ *     re-enabling it, and refusing here would make the disable-repair-re-enable
+ *     loop unclosable, which is the whole point of FR-WHK-09.
+ *   * NO CLAIM LEDGER. Expansion claims the event so a broker redelivery cannot
+ *     double a customer's webhooks; nothing redelivers a test, because a person
+ *     asked for it once over HTTP.
+ *
+ * Soft-deleted endpoints are still refused. A deleted endpoint is gone as far as
+ * the customer's own API is concerned, and delivering to one would be the platform
+ * reaching a url the customer believes it has forgotten.
+ *
+ * Everything else is ordinary: a real row, on the real schedule, delivered by the
+ * real dispatcher, signed by the real signing path. That is FR-014 — a test whose
+ * delivery worked differently would prove nothing about real deliveries. */
+export async function createTestDelivery(
+  db: Db,
+  input: { endpointId: string; environmentId: string },
+): Promise<{ deliveryId: string; eventId: string; payload: unknown } | null> {
+  const [endpoint] = await db
+    .select({ id: webhookEndpoints.id })
+    .from(webhookEndpoints)
+    .where(
+      and(
+        eq(webhookEndpoints.id, input.endpointId),
+        eq(webhookEndpoints.environmentId, input.environmentId),
+        isNull(webhookEndpoints.deletedAt),
+      ),
+    );
+  if (!endpoint) return null;
+
+  const eventId = randomUUID();
+  // MARKED TWICE, for two different readers (FR-015). A recipient switching on
+  // `type` and a recipient inspecting the body should each be able to tell this is
+  // synthetic without knowing about the other — and neither should have to know
+  // about `webhook_deliveries.synthetic`, which is the platform's own marker and
+  // not part of the contract.
+  const payload = {
+    id: eventId,
+    type: TEST_EVENT_TYPE,
+    environment_id: input.environmentId,
+    occurred_at: new Date().toISOString(),
+    test: true,
+    data: { message: "This is a test event from Relay." },
+  };
+
+  const deliveryId = randomUUID();
+  await db.insert(webhookDeliveries).values({
+    id: deliveryId,
+    environmentId: input.environmentId,
+    endpointId: endpoint.id,
+    eventId,
+    payload,
+    // Due immediately: a caller is waiting.
+    attempt: 1,
+    synthetic: true,
+  });
+
+  return { deliveryId, eventId, payload };
+}
+
+/** What a test event's envelope calls itself. Exported because the contract names
+ * it and a recipient may switch on it. */
+export const TEST_EVENT_TYPE = "webhook.test";
+
+/** What the endpoint answered, read back off the delivery the test created.
+ *
+ * The attempt happens in the DISPATCHER's process, so the route that is holding a
+ * customer's request cannot observe it directly — it waits for the row to move.
+ * `state` is the signal: `pending` means no outcome has been recorded yet. */
+export async function testDeliveryResult(
+  db: Db,
+  deliveryId: string,
+): Promise<{
+  settled: boolean;
+  delivered: boolean;
+  status: number | null;
+  error: string | null;
+  latencyMs: number | null;
+} | null> {
+  const [row] = await db
+    .select({
+      state: webhookDeliveries.state,
+      lastStatus: webhookDeliveries.lastStatus,
+      lastError: webhookDeliveries.lastError,
+      lastLatencyMs: webhookDeliveries.lastLatencyMs,
+    })
+    .from(webhookDeliveries)
+    .where(eq(webhookDeliveries.id, deliveryId));
+  if (!row) return null;
+
+  return {
+    settled: row.state !== "pending",
+    // A test event gets one attempt, so `delivered` and "the state is delivered"
+    // are the same fact. Reading the state rather than the status keeps that
+    // decision in one place — `recordAttemptOutcome` already decided what 2xx
+    // means, and a second opinion here is a second thing to get wrong.
+    delivered: row.state === "delivered",
+    status: row.lastStatus,
+    error: row.lastError,
+    latencyMs: row.lastLatencyMs,
+  };
+}
+
+/** Disable every endpoint whose failure run has outrun the hour (chapter 3.6,
+ * research R1, contract invariant 12).
+ *
+ * THE SECOND TRIGGER, and it is not belt-and-braces. The on-outcome check catches
+ * every endpoint that is still receiving attempts, and research R1 measured that
+ * this is not all of them. Against chapter 3.5's tier table, one failing delivery
+ * attempts at +35m36s and then not again until +2h35m36s — so nothing happens AT
+ * the hour, and a check that only runs when an outcome is recorded fires
+ * ninety-five minutes late. Worse: if that last attempt dead-letters and no
+ * further events arrive for the environment, no outcome is ever recorded again and
+ * the endpoint is never disabled at all. It sits enabled and failing for ever,
+ * which is the state FR-WHK-07 exists to end.
+ *
+ * The endpoint that stays broken silently is the QUIET one — the low-traffic
+ * customer, who is also the customer least likely to be watching.
+ *
+ * Rides the delivery relay's existing loop rather than adding a scheduler: one
+ * more statement per drain, in a worker that is already awake and already holds a
+ * connection (constitution VII). Its per-endpoint work goes through the same
+ * `disableEndpoint` the on-outcome path uses, so the at-most-once rule is one rule
+ * and not two implementations of it.
+ *
+ * Returns how many it disabled, so the relay can log a number rather than a claim.
+ */
+export async function sweepDisabledEndpoints(
+  db: Db,
+  limit = 100,
+): Promise<number> {
+  // An INTERVAL built from the same constant the pure policy uses, so the sweep and
+  // `shouldDisable` can never disagree about how long an hour is. Milliseconds
+  // rather than a literal `'1 hour'`: one definition, in `disable.ts`.
+  const disableCutoff = sql`now() - make_interval(secs => ${DISABLE_AFTER_MS / 1000})`;
+
+  return db.transaction(async (tx) => {
+    // The candidates, and the LAST THING each endpoint heard. `disableReason` and
+    // the notification both want a status the sweep does not have in hand — it
+    // fires precisely when no outcome is arriving — so it is read off the
+    // endpoint's most recent attempted delivery. LEFT JOIN LATERAL, because an
+    // endpoint whose deliveries have all been pruned still deserves to be switched
+    // off; it just gets "no response" as its reason.
+    //
+    // `FOR UPDATE OF e SKIP LOCKED` is chapter 3.3's pattern and here it does two
+    // jobs: it serialises the sweep against a concurrent outcome report on the same
+    // endpoint, and it lets two api instances sweep at once without either waiting
+    // — whichever skips simply finds nothing to do, which is correct.
+    const candidates = (await tx.execute(sql`
+      SELECT e.id,
+             e.environment_id,
+             e.failure_run_started_at AS run_started_at,
+             e.failure_run_attempts   AS run_attempts,
+             now()                    AS now,
+             last.last_status,
+             last.last_error
+        FROM webhook_endpoints e
+        LEFT JOIN LATERAL (
+          SELECT d.last_status, d.last_error
+            FROM webhook_deliveries d
+           WHERE d.endpoint_id = e.id
+             AND d.synthetic = false
+             AND d.last_latency_ms IS NOT NULL
+           ORDER BY d.next_attempt_at DESC, d.id DESC
+           LIMIT 1
+        ) last ON true
+       WHERE e.enabled = true
+         AND e.deleted_at IS NULL
+         AND e.failure_run_started_at IS NOT NULL
+         AND e.failure_run_attempts >= ${DISABLE_MIN_ATTEMPTS}
+         AND e.failure_run_started_at < ${disableCutoff}
+       ORDER BY e.failure_run_started_at
+       LIMIT ${limit}
+         FOR UPDATE OF e SKIP LOCKED`)) as unknown as {
+      rows: {
+        id: string;
+        environment_id: string;
+        run_started_at: string | Date;
+        run_attempts: number;
+        now: string | Date;
+        last_status: number | null;
+        last_error: string | null;
+      }[];
+    };
+
+    let disabled = 0;
+    for (const row of candidates.rows) {
+      const result = await disableEndpoint(tx, {
+        endpointId: row.id,
+        environmentId: row.environment_id,
+        // Same coercion, same reason as `applyFailureRun`'s clock read.
+        runStartedAt: new Date(row.run_started_at),
+        runAttempts: row.run_attempts,
+        now: new Date(row.now),
+        status: row.last_status,
+        error: row.last_error,
+      });
+      if (result.disabled) disabled++;
+    }
+    return disabled;
   });
 }
 
@@ -517,6 +1041,7 @@ export async function deliveryMaterial(
       url: webhookEndpoints.url,
       enabled: webhookEndpoints.enabled,
       deletedAt: webhookEndpoints.deletedAt,
+      synthetic: webhookDeliveries.synthetic,
       secretCiphertext: webhookEndpoints.secretCiphertext,
       secretPreviousCiphertext: webhookEndpoints.secretPreviousCiphertext,
       secretRotatedAt: webhookEndpoints.secretRotatedAt,
@@ -532,7 +1057,19 @@ export async function deliveryMaterial(
   // An endpoint paused or removed after the delivery was scheduled gets nothing.
   // The spec's edge case: events already in the retry schedule for a removed
   // endpoint must not be delivered.
-  if (!row.enabled || row.deletedAt) return null;
+  //
+  // A TEST EVENT IS THE EXCEPTION, and it is the only one (chapter 3.6, FR-013).
+  // Two requirements meet exactly here and pull opposite ways: invariant 9 says a
+  // disabled endpoint receives no attempts, and FR-013 says a customer may test a
+  // disabled endpoint — which is how they establish it is fixed BEFORE re-enabling
+  // it. `synthetic` is what tells them apart, and it is the reason that column is a
+  // column rather than a string comparison against a customer-visible payload.
+  //
+  // DELETED IS STILL DELETED. A soft-deleted endpoint is gone as far as the
+  // customer's own API is concerned, and delivering to one — test or not — would be
+  // the platform reaching a url the customer believes it has forgotten.
+  if (row.deletedAt) return null;
+  if (!row.enabled && !row.synthetic) return null;
 
   return {
     delivery_id: row.id,
@@ -1007,6 +1544,17 @@ export interface WebhookEndpointRow {
   enabled: boolean;
   secret_rotated_at: string | null;
   created_at: string;
+  /** Chapter 3.6, FR-009. `enabled: false` with `disabled_at: null` means the
+   * customer paused it themselves; both set means the platform did. Without this
+   * pair a customer looking at a disabled endpoint has no way to tell whether they
+   * are looking at their own decision or ours, and the support conversation starts
+   * from zero. */
+  disabled_at: string | null;
+  disabled_reason: string | null;
+  /** The run as it stands, so a customer can see a disablement COMING rather than
+   * only after it lands. Both null when the endpoint is healthy. */
+  failure_run_started_at: string | null;
+  failure_run_attempts: number | null;
 }
 
 /** Raised when an outcome names a delivery that is not there. A caller error,
@@ -1051,6 +1599,18 @@ export class Repository {
     private readonly environmentId: string,
   ) {}
 
+  /** The tenant this repository is scoped to, readable but not settable.
+   *
+   * Added in chapter 3.6 for the test event, which needs an UNSCOPED operation —
+   * `createTestDelivery` ignores subscriptions and `enabled`, so it cannot be a
+   * method here — but must still be told which environment is asking. Exposing the
+   * id rather than widening the operation keeps constitution I's shape: the
+   * environment still comes from a verified principal, never from a request body.
+   */
+  get environment(): string {
+    return this.environmentId;
+  }
+
   // ---------------------------------------------------------------------
   // Webhook endpoints (chapter 3.5). Scoped like everything else on this class:
   // the environment comes from the constructor and never from a caller, so a
@@ -1136,7 +1696,31 @@ export class Repository {
   ): Promise<WebhookEndpointRow | null> {
     await this.db
       .update(webhookEndpoints)
-      .set({ enabled })
+      .set({
+        enabled,
+        // RE-ENABLING CLEARS THE RUN, all four columns, in this one statement
+        // (chapter 3.6, FR-017). The hour is measured from the NEXT failure, not
+        // resumed from the old one — otherwise a customer who fixed their server
+        // and switched it back on would be disabled again by the first failure
+        // after that, on the strength of an outage they had already repaired.
+        //
+        // `disabled_at` and `disabled_reason` go too, because FR-009 reads them as
+        // "the platform switched this off" and after a re-enable that is no longer
+        // true. A schema CHECK requires those two to agree, so they must move
+        // together whatever else happens here.
+        //
+        // DISABLING clears nothing. A customer pausing their own endpoint has said
+        // nothing about whether it is healthy, and throwing away the run would let
+        // a disable/enable cycle launder an hour of failures.
+        ...(enabled
+          ? {
+              failureRunStartedAt: null,
+              failureRunAttempts: null,
+              disabledAt: null,
+              disabledReason: null,
+            }
+          : {}),
+      })
       .where(and(eq(webhookEndpoints.id, id), this.liveEndpoints));
     return this.getEndpoint(id);
   }
@@ -1227,6 +1811,10 @@ export class Repository {
         enabled: webhookEndpoints.enabled,
         secretRotatedAt: webhookEndpoints.secretRotatedAt,
         createdAt: webhookEndpoints.createdAt,
+        disabledAt: webhookEndpoints.disabledAt,
+        disabledReason: webhookEndpoints.disabledReason,
+        failureRunStartedAt: webhookEndpoints.failureRunStartedAt,
+        failureRunAttempts: webhookEndpoints.failureRunAttempts,
       })
       .from(webhookEndpoints)
       .where(where);
@@ -1239,6 +1827,10 @@ export class Repository {
       enabled: r.enabled,
       secret_rotated_at: r.secretRotatedAt?.toISOString() ?? null,
       created_at: r.createdAt.toISOString(),
+      disabled_at: r.disabledAt?.toISOString() ?? null,
+      disabled_reason: r.disabledReason,
+      failure_run_started_at: r.failureRunStartedAt?.toISOString() ?? null,
+      failure_run_attempts: r.failureRunAttempts,
     }));
   }
 

@@ -12,7 +12,11 @@ import {
 } from "nats";
 
 import type { Db } from "../db/client";
-import { drainDueDeliveries, type DueDeliveryRow } from "../db/repository";
+import {
+  drainDueDeliveries,
+  sweepDisabledEndpoints,
+  type DueDeliveryRow,
+} from "../db/repository";
 import type { Publisher } from "../outbox/publisher";
 
 // The second relay (chapter 3.5, research R13).
@@ -84,6 +88,8 @@ export interface DeliveryRelay {
   /** One pass, for tests and for the walk script — the same code path `start`
    * runs, so nothing is proven about a loop only tests exercise. */
   drainOnce(): Promise<number>;
+  /** One sweep, same argument. Returns how many endpoints it disabled. */
+  sweepOnce(): Promise<number>;
 }
 
 export function createDeliveryRelay({
@@ -92,12 +98,22 @@ export function createDeliveryRelay({
   logger,
   batchSize = BATCH_SIZE,
   intervalMs = IDLE_INTERVAL_MS,
+  sweepEnabled = process.env["RELAY_DISABLE_SWEEP"] !== "off",
 }: {
   db: Db;
   publisher: Publisher;
   logger: Logger;
   batchSize?: number;
   intervalMs?: number;
+  /** `RELAY_DISABLE_SWEEP=off` turns the second trigger off, and it exists for one
+   * purpose: quickstart V6 asks a reader to watch a quiet endpoint stay enabled
+   * and failing with the sweep off, and then be disabled with it on. The first
+   * half is what an outcome-only check ships, and reading it is the only way the
+   * second half means anything.
+   *
+   * DEFAULT ON. A flag whose default disabled a requirement would be a
+   * requirement nobody had built. */
+  sweepEnabled?: boolean;
 }): DeliveryRelay {
   let running = false;
   let loop: Promise<void> = Promise.resolve();
@@ -135,6 +151,35 @@ export function createDeliveryRelay({
     });
   }
 
+  /** The auto-disable sweep, riding this loop (chapter 3.6, research R1).
+   *
+   * Here rather than in a scheduler of its own because this worker is already
+   * awake, already owns a database connection, and already runs in the one service
+   * permitted to write (constitution VII, constitution IV). A third background
+   * process would be a third thing to deploy, monitor and reason about, for one
+   * statement.
+   *
+   * Its failure is logged and dropped for the same reason the drain's is: an
+   * endpoint that should have been disabled and was not is a cost measured in one
+   * more failed delivery, while a relay that stops draining is a cost measured in
+   * every customer's webhooks. */
+  async function sweepOnce(): Promise<number> {
+    if (!sweepEnabled) return 0;
+    try {
+      const disabled = await sweepDisabledEndpoints(db);
+      if (disabled > 0) {
+        // A COUNT, and only when it is not zero. This runs several times a second
+        // when the platform is idle, and a line per pass would bury every other
+        // line in the service.
+        logger.log("info", "webhooks.endpoints_disabled", { count: disabled });
+      }
+      return disabled;
+    } catch (error) {
+      logger.log("error", "webhooks.disable_sweep_failed", { error: String(error) });
+      return 0;
+    }
+  }
+
   async function run(): Promise<void> {
     while (running) {
       try {
@@ -151,6 +196,11 @@ export function createDeliveryRelay({
         // buffering 3.3's relay promises for events.
         logger.log("error", "deliveries.drain_failed", { error: String(error) });
       }
+      // AFTER the drain and only when there was nothing due, so a backlog is never
+      // made to wait behind a housekeeping query. An endpoint that has been failing
+      // for an hour can wait another quarter of a second; a customer's webhook
+      // cannot.
+      await sweepOnce();
       await new Promise((resolve) => setTimeout(resolve, intervalMs));
     }
   }
@@ -166,5 +216,6 @@ export function createDeliveryRelay({
       await loop;
     },
     drainOnce,
+    sweepOnce,
   };
 }

webhook_disable_notifications
  environment_id · organisation_id · endpoint_id
  disabled_at · run_started_at · run_attempts
  last_status · last_error
  delivered_at        <-- always null in this chapter

@@ -1,10 +1,17 @@
 import {
+  Inject,
   Injectable,
   NotFoundException,
   UnprocessableEntityException,
 } from "@nestjs/common";
 
-import { Repository, type WebhookEndpointRow } from "../db/repository";
+import type { Db } from "../db/client";
+import {
+  createTestDelivery,
+  Repository,
+  testDeliveryResult,
+  type WebhookEndpointRow,
+} from "../db/repository";
 import { encryptSecret, mintSigningSecret } from "./secret";
 
 // The management surface's rules (chapter 3.5). Two of them are worth stating
@@ -46,9 +53,31 @@ export interface EndpointWithSecret extends WebhookEndpointRow {
   secret: string;
 }
 
+/** How long a caller waits for a test event to come back.
+ *
+ * The attempt happens in the dispatcher's process, so this route can only watch
+ * the row. Ten seconds is above the dispatcher's own attempt timeout, so a
+ * customer whose server hangs gets the honest answer — "no response" — rather than
+ * this bound firing first and reporting an inconclusive result for a conclusive
+ * failure. */
+const TEST_EVENT_TIMEOUT_MS = 10_000;
+const TEST_EVENT_POLL_MS = 100;
+
+/** What a test event answered, or why there is no answer yet. */
+export interface TestEventResult {
+  delivered: boolean;
+  status: number | null;
+  latency_ms: number | null;
+  error: string | null;
+  event_id: string;
+}
+
 @Injectable()
 export class WebhooksService {
-  constructor(private readonly repo: Repository) {}
+  constructor(
+    private readonly repo: Repository,
+    @Inject("DB") private readonly db: Db,
+  ) {}
 
   async create(input: CreateEndpointInput): Promise<EndpointWithSecret> {
     this.assertDeliverableUrl(input.url);
@@ -98,6 +127,59 @@ export class WebhooksService {
     return row;
   }
 
+  /** Send one synthetic event to one endpoint and report what it answered
+   * (chapter 3.6, FR-013…FR-016, research R8).
+   *
+   * The delivery is REAL: a row on the ordinary schedule, published by the
+   * ordinary relay, posted and signed by the ordinary dispatcher, its outcome
+   * recorded by the ordinary seam. This method creates it and then WATCHES,
+   * because the attempt happens in another process and constitution IV does not
+   * let that process write its own answer anywhere else.
+   *
+   * That is why FR-014 is met rather than approximated. A route that posted the
+   * request itself would need a second copy of the signing code, and a test signed
+   * by a second copy proves nothing about real deliveries — which is the one thing
+   * a test event exists to prove. */
+  async sendTestEvent(endpointId: string): Promise<TestEventResult> {
+    const created = await createTestDelivery(this.db, {
+      endpointId,
+      environmentId: this.repo.environment,
+    });
+    // The same 404 a missing endpoint gets, so a probe cannot tell "no such
+    // endpoint" from "not yours" (FR-TEN-05).
+    if (!created) throw new NotFoundException("no such webhook endpoint");
+
+    const deadline = Date.now() + TEST_EVENT_TIMEOUT_MS;
+    for (;;) {
+      const result = await testDeliveryResult(this.db, created.deliveryId);
+      if (result?.settled) {
+        return {
+          delivered: result.delivered,
+          status: result.status,
+          latency_ms: result.latencyMs,
+          error: result.error,
+          event_id: created.eventId,
+        };
+      }
+      if (Date.now() >= deadline) {
+        // NOT an HTTP error, and the distinction is the contract's. A non-2xx from
+        // the customer means the test succeeded in finding out; this means the
+        // platform did not find out, and the `error` says which. Answering 500
+        // would conflate "we could not run the test" with "your endpoint is
+        // unhealthy", and the second is what a customer would read.
+        return {
+          delivered: false,
+          status: null,
+          latency_ms: null,
+          error:
+            "no attempt was reported within 10s — is the dispatcher running?",
+          event_id: created.eventId,
+        };
+      }
+      await new Promise((resolve) => setTimeout(resolve, TEST_EVENT_POLL_MS));
+    }
+  }
+
   async remove(id: string): Promise<void> {
     const deleted = await this.repo.deleteEndpoint(id);
     if (!deleted) throw new NotFoundException("no such webhook endpoint");

@@ -67,6 +67,23 @@ export class WebhooksController {
     return this.webhooks.setEnabled(id, false);
   }
 
+  /** Send a synthetic event to this endpoint and report what it answered
+   * (chapter 3.6, FR-013, FR-016).
+   *
+   * ALWAYS 200 when the test ran, whatever the endpoint said. A non-2xx from the
+   * customer's server is reported as `delivered: false` with the status, because
+   * the TEST succeeded — it found out. Answering 502 here would conflate "we could
+   * not run the test" with "your endpoint is unhealthy", and a customer debugging
+   * at 3 a.m. would read the second.
+   *
+   * 404 for an endpoint in another environment, the same answer a missing one
+   * gets. */
+  @Post(":id/test")
+  @HttpCode(200)
+  test(@Param("id") id: string) {
+    return this.webhooks.sendTestEvent(id);
+  }
+
   /** SOFT delete. The endpoint stops receiving deliveries and disappears from
    * every read; the row survives because its dead letters must (FR-WHK-04). */
   @Delete(":id")

$ node scripts/hostile-endpoint.mjs --mode=fail --quiet --secret=hunter2
MARKER listening url=http://127.0.0.1:4555/hook mode=fail

@@ -8,9 +8,13 @@
 //   node scripts/webhook-walk.mjs --print-signing-material
 //   node scripts/webhook-walk.mjs --fast-forward     # against --mode=fail
 //   node scripts/webhook-walk.mjs --send-only        # leave it for the real dispatcher
+//   node scripts/webhook-walk.mjs --fast-forward --watch-disable   # chapter 3.6
 //
 //   --url=http://127.0.0.1:4555/hook   where to point the endpoint
 //   --api-port=4141                    the api this walk spawns for itself
+//   --watch-disable                    chapter 3.6: print the failure run as it
+//                                      grows and the disablement when it lands.
+//                                      Ages the run rather than waiting an hour.
 //   --secret=SECRET                    pin the signing secret instead of minting
 //                                      one, so the endpoint can be started with
 //                                      the same value and verify what arrives:
@@ -59,6 +63,9 @@ const { createDispatcher } = await import(
 const { SIGNATURE_SCHEME } = await import(
   join(ROOT, "services", "dispatcher", "dist", "signature.js")
 );
+const { DISABLE_AFTER_MS, DISABLE_MIN_ATTEMPTS } = await import(
+  join(API_DIST, "webhooks", "disable.js")
+);
 const { RETRY_TIERS_MS, MAX_ATTEMPTS } = await import(
   join(API_DIST, "webhooks", "schedule.js")
 );
@@ -77,6 +84,7 @@ const API_PORT = Number(arg("api-port", "4141"));
 const PRINT_MATERIAL = flag("print-signing-material");
 const SEND_ONLY = flag("send-only");
 const FAST_FORWARD = flag("fast-forward");
+const WATCH_DISABLE = flag("watch-disable");
 const PINNED_SECRET = arg("secret", "");
 const CREDENTIAL = "rk_svc_walk_0123456789abcdef0123456789abcd";
 
@@ -292,6 +300,21 @@ await dispatcher.pollOnce();
 console.log("");
 await report();
 
+/** The endpoint's failure run, straight from the row auto-disable reads.
+ *
+ * Read with plain SQL because this script is a reader's tool and the columns are
+ * the point: `enabled` is what stops deliveries, and `disabled_at` is how a
+ * customer tells a platform disablement from their own (FR-009). */
+const runOf = async () => {
+  const { rows } = await pool.query(
+    `SELECT enabled, disabled_at, disabled_reason,
+            failure_run_started_at, failure_run_attempts
+       FROM webhook_endpoints WHERE id = $1`,
+    [endpoint.id],
+  );
+  return rows[0];
+};
+
 // ---------------------------------------------------------------------------
 if (FAST_FORWARD) {
   rule("5. the whole schedule, without waiting for it");
@@ -315,6 +338,7 @@ if (FAST_FORWARD) {
 
   const stateOf = async () => (await repo.listDeliveriesForEvent(eventId))[0];
 
+
   for (let i = 0; i < MAX_ATTEMPTS + 2; i++) {
     const before = await stateOf();
     if (!before || before.state !== "pending") break;
@@ -343,6 +367,16 @@ if (FAST_FORWARD) {
         ? `failed → rescheduled as attempt ${after.attempt}`
         : `failed → ${after ? after.state : "gone"}`,
     );
+
+    if (WATCH_DISABLE) {
+      const run = await runOf();
+      show(
+        "    endpoint",
+        run.failure_run_started_at
+          ? `run open · ${run.failure_run_attempts} failures · enabled=${run.enabled}`
+          : `no run · enabled=${run.enabled}`,
+      );
+    }
   }
 
   console.log("");
@@ -356,6 +390,75 @@ if (FAST_FORWARD) {
   }
 }
 
+// ---------------------------------------------------------------------------
+if (WATCH_DISABLE) {
+  rule("6. when to stop trying");
+
+  console.log("  The run above is what auto-disable reads — two columns on the endpoint,");
+  console.log("  never the attempt stream. A backlogged analytics path cannot delay a");
+  console.log("  disablement, and a broker being unwell cannot block one.\n");
+  console.log(`  The rule: longer than ${DISABLE_AFTER_MS / 60000} minutes AND at least`);
+  console.log(`  ${DISABLE_MIN_ATTEMPTS} failures. Both, never either — the hour alone would let one`);
+  console.log("  failure followed by a two-hour retry gap disable an endpoint.\n");
+
+  const before = await runOf();
+  show("failures in the run", before.failure_run_attempts ?? 0);
+  show("enabled", before.enabled);
+
+  if (!before.failure_run_started_at) {
+    console.log("\n  No open run — the endpoint answered 2xx at some point, which CLEARS it.");
+    console.log("  Point this at --mode=fail to watch a run survive long enough to matter.\n");
+  } else {
+    // AGED, not waited out. Same honesty as --fast-forward above: the clock moves,
+    // the logic does not. An hour and four minutes of real time would make this
+    // demonstration one nobody runs.
+    console.log("  aging the run past the hour (the clock moves, the rule does not)...\n");
+    await pool.query(
+      `UPDATE webhook_endpoints
+          SET failure_run_started_at = now() - interval '64 minutes'
+        WHERE id = $1`,
+      [endpoint.id],
+    );
+
+    // THE SWEEP, which is the trigger this endpoint needs. Its last attempt has
+    // already been made — the delivery dead-lettered above — so no further outcome
+    // will ever be reported and an on-outcome check would never fire again. That is
+    // research R1's quiet endpoint, and it is the reason there are two triggers.
+    const disabled = await relay.sweepOnce();
+    show("endpoints the sweep disabled", disabled);
+
+    const after = await runOf();
+    console.log("");
+    show("enabled", after.enabled);
+    show("disabled_at", after.disabled_at ? after.disabled_at.toISOString() : "null");
+    show("disabled_reason", after.disabled_reason ?? "null");
+
+    const { rows: notes } = await pool.query(
+      `SELECT run_attempts, last_status, last_error, delivered_at
+         FROM webhook_disable_notifications WHERE endpoint_id = $1`,
+      [endpoint.id],
+    );
+    console.log("");
+    show("notification rows", notes.length);
+    for (const n of notes) {
+      show(
+        `  run of ${n.run_attempts}`,
+        `last_status=${n.last_status ?? "none"} delivered_at=${n.delivered_at ?? "null"}`,
+      );
+    }
+    console.log("");
+    console.log("  `delivered_at` is null and stays null. FR-WHK-07 asks for the");
+    console.log("  organisation to be notified BY EMAIL, and this platform has no email");
+    console.log("  transport of any kind. The row is the obligation; the null is the");
+    console.log("  admission. Chapter 3.7 needs the same transport for quotas.\n");
+
+    // Running it again must change nothing. At most once per run, enforced by the
+    // `enabled = true` predicate in the update rather than by a check.
+    const second = await relay.sweepOnce();
+    show("a second sweep disables", second);
+  }
+}
+
 rule("what to take from this");
 console.log(`
   The delivery row is the schedule. Not a message the broker is holding — that

$ node scripts/webhook-walk.mjs --secret=hunter2 --fast-forward --watch-disable
 
=== 5. the whole schedule, without waiting for it =
  tiers: now → 1s → 5s → 30s → 300s → 1800s → 7200s
  7 attempts, then the delivery is dead-lettered.
 
  attempt 2                failed → rescheduled as attempt 3
    endpoint               run open · 2 failures · enabled=true
  attempt 3                failed → rescheduled as attempt 4
    endpoint               run open · 3 failures · enabled=true
  attempt 4                failed → rescheduled as attempt 5
    endpoint               run open · 4 failures · enabled=true
  attempt 5                failed → rescheduled as attempt 6
    endpoint               run open · 5 failures · enabled=true
  attempt 6                failed → rescheduled as attempt 7
    endpoint               run open · 6 failures · enabled=true
  attempt 7                failed → dead
    endpoint               run open · 7 failures · enabled=true
 
  delivery b9a68bee        attempt=7 state=dead next=2026-08-18T14:52:26.767Z
 
dead letters               1
  3e2acaf8                 attempts=7 last_status=500

=== 6. when to stop trying ========================
  The rule: longer than 60 minutes AND at least
  5 failures. Both, never either.
 
failures in the run        7
enabled                    true
  aging the run past the hour (the clock moves, the rule does not)...
 
endpoints the sweep disabled 1
 
enabled                    false
disabled_at                2026-08-18T14:52:29.830Z
disabled_reason            7 consecutive failures over 1h04m; last status 500
 
notification rows          1
  run of 7                 last_status=500 delivered_at=null
 
a second sweep disables    0

$ RELAY_DISABLE_SWEEP=off node scripts/webhook-walk.mjs --secret=hunter2 --fast-forward --watch-disable
 
failures in the run        7
enabled                    true
  aging the run past the hour (the clock moves, the rule does not)...
 
endpoints the sweep disabled 0
 
enabled                    true
disabled_at                null
disabled_reason            null
 
notification rows          0

$ node scripts/stream-info.mjs ANALYTICS
stream ANALYTICS
  messages           7
  bytes              0.0 MiB
  consumers          0
configuration
  subjects           ["analytics.>"]
  retention          limits   (immutable once created)
  storage            file    (immutable once created)
  replicas           1
  max_age            604800s
  max_bytes          1.00 GiB
  discard            old     (at the bound, drop the OLDEST)
  duplicate_window   120s   (the broker's dedupe, not ours)
consumers

@@ -5,8 +5,16 @@
 //
 //   docker compose up -d --wait nats
 //   RELAY_NATS_URL=nats://localhost:14222 node scripts/stream-info.mjs
+//   RELAY_NATS_URL=nats://localhost:14222 node scripts/stream-info.mjs ANALYTICS
+//
+// The stream is an ARGUMENT as of chapter 3.6, defaulting to the one stream that
+// existed when this was written. It had `"EVENTS"` in three places, and 3.6's
+// quickstart asks the reader to inspect `ANALYTICS` — which would have printed
+// the wrong stream's configuration and passed, since every field it shows exists
+// on both. A validation step that cannot fail is worse than no step.
 import { connect } from "../services/api/node_modules/nats/lib/src/mod.js";
 
+const stream = process.argv[2] ?? "EVENTS";
 const url = process.env.RELAY_NATS_URL ?? "nats://127.0.0.1:4222";
 const nc = await connect({ servers: url });
 const jsm = await nc.jetstreamManager();
@@ -15,10 +23,21 @@ const seconds = (ns) => (ns === 0 ? "unlimited" : `${ns / 1e9}s`);
 const bytes = (n) => (n === -1 ? "unlimited" : `${(n / 1024 ** 3).toFixed(2)} GiB`);
 const show = (label, value) => console.log(`${label.padEnd(20)} ${value}`);
 
-const info = await jsm.streams.info("EVENTS");
+const info = await jsm.streams.info(stream).catch(() => null);
+if (info === null) {
+  // Named loudly rather than crashed on. A stream that does not exist yet is the
+  // ordinary state before the api has started once, and "no such stream" is a
+  // more useful answer than a broker client's stack trace.
+  console.error(
+    `no stream named ${stream}. The api CREATES its streams on first publish —\n` +
+      "start it once, or check the name (EVENTS, DELIVERIES, ANALYTICS).",
+  );
+  await nc.drain();
+  process.exit(1);
+}
 const c = info.config;
 
-console.log("stream EVENTS");
+console.log(`stream ${stream}`);
 show("  messages", info.state.messages);
 show("  bytes", `${(info.state.bytes / 1024 ** 2).toFixed(1)} MiB`);
 show("  consumers", info.state.consumer_count);
@@ -27,13 +46,18 @@ show("  subjects", JSON.stringify(c.subjects));
 show("  retention", `${c.retention}   (immutable once created)`);
 show("  storage", `${c.storage}    (immutable once created)`);
 show("  replicas", c.num_replicas);
-show("  max_age", `${seconds(c.max_age)}   (NFR-REL-08 floor: 86400s)`);
+show(
+  "  max_age",
+  stream === "EVENTS"
+    ? `${seconds(c.max_age)}   (NFR-REL-08 floor: 86400s)`
+    : seconds(c.max_age),
+);
 show("  max_bytes", bytes(c.max_bytes));
 show("  discard", `${c.discard}     (at the bound, drop the OLDEST)`);
 show("  duplicate_window", `${seconds(c.duplicate_window)}   (the broker's dedupe, not ours)`);
 
 console.log("consumers");
-for await (const ci of jsm.consumers.list("EVENTS")) {
+for await (const ci of jsm.consumers.list(stream)) {
   show(
     `  ${ci.name}`,
     `pending=${ci.num_pending} ack_pending=${ci.num_ack_pending} redelivered=${ci.num_redelivered} max_deliver=${ci.config.max_deliver ?? "-"}`,

{
  "delivery_id": "eb8fa3c9-606f-49a1-b872-b577efefe793",
  "endpoint_id": "2678590a-addc-489c-9ef5-3f8a807ddb16",
  "environment_id": "b0fc170b-58dc-435d-b452-3f7df834e90f",
  "event_id": "9f0667a3-c049-459d-9291-c70cffb35871",
  "attempt": 1,
  "attempted_at": "2026-08-18T16:06:13.509Z",
  "status": 500,
  "latency_ms": 12,
  "outcome": "rescheduled"
}

import "reflect-metadata";
 
import { randomUUID } from "node:crypto";
 
import { Test } from "@nestjs/testing";
import type { INestApplication } from "@nestjs/common";
import { AckPolicy, connect, DeliverPolicy, type NatsConnection } from "nats";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
import { ALL_ANALYTICS_SUBJECT, ANALYTICS_STREAM } from "@relay/protocol";
 
import { AppModule } from "../app.module";
import { createDb, createPool, type Db } from "../db/client";
import { ensureAnalyticsStream } from "../outbox/jetstream.publisher";
import {
  createEnvironment,
  drainDueDeliveries,
  expandEventToDeliveries,
  recordAttemptOutcome,
  Repository,
} from "../db/repository";
import { encryptSecret, mintSigningSecret } from "./secret";
import { MAX_ATTEMPTS } from "./schedule";
 
// The attempt record, against a real broker and a real api (chapter 3.6).
//
// Invariants 1, 2, 3 and 5 of contracts/attempts.md live here. Invariant 4 is the
// swallowed publish failure and is pure, so it lives in analytics.test.ts.
//
// This suite drives the INTERNAL ROUTE over HTTP rather than calling
// `publishAttempt` itself, and that is the whole point of it being an integration
// test. The claim under test is not "the publisher works" — the unit lane holds
// that — it is "recording an outcome puts exactly one event on the stream, and the
// publish happens outside the transaction that recorded it". Neither half is
// visible from inside the publisher.
//
// Every environment is minted here, and every assertion is scoped to a subject
// carrying that environment's id. The stream is global and this lane runs beside
// other suites, so an assertion that counted messages on `analytics.>` would be
// counting somebody else's work (chapter 3.3's finding 4, again).
 
const CREDENTIAL =
  process.env["RELAY_INTERNAL_CREDENTIAL"] ??
  "rk_svc_attempts_itest_0123456789abcdef0123";
 
/** A per-run consumer name. A durable IS a position in a shared stream, and
 * chapter 3.6's own baseline is the reason this is a per-SUITE prefix rather than
 * a shared `itest-`: the api's consumer suite once swept every consumer whose name
 * began with that, and deleted a live one belonging to another suite. */
const SUITE = "itest-attempts";
 
describe("the attempt record", () => {
  let app: INestApplication;
  let url: string;
  let db: Db;
  let nats: NatsConnection;
  let env: { id: string };
  let repo: Repository;
  let durable: string;
 
  const seedEndpoint = async (
    scope: Repository,
    eventTypes = ["message.created"],
  ) => {
    const secret = mintSigningSecret();
    return scope.createEndpoint({
      url: `https://example.test/${randomUUID()}`,
      eventTypes,
      secretCiphertext: encryptSecret(secret),
    });
  };
 
  /** The relay's claim, with the publish stubbed out.
   *
   * `drainDueDeliveries` takes the publish as a callback — the seam that makes
   * chapter 3.3's outbox broker-agnostic — so claiming a delivery without putting
   * it on the DELIVERIES stream is a one-line stub rather than a mock. This suite
   * is about the ANALYTICS stream, and a real delivery publish here would hand
   * work to whatever dispatcher happens to be running beside it. */
  const claimOnly = () => drainDueDeliveries(db, 50, async () => {});
 
  /** Expand one event into deliveries and claim them, which is what leaves a row
   * an outcome can be reported against. The relay's claim is included because an
   * unclaimed delivery is not one the dispatcher would ever have posted. */
  const deliveryFor = async (environmentId: string, scope: Repository) => {
    const eventId = randomUUID();
    await expandEventToDeliveries(db, {
      eventId,
      environmentId,
      type: "message.created",
      payload: { id: eventId, type: "message.created" },
    });
    await claimOnly();
    const [row] = await scope.listDeliveriesForEvent(eventId);
    expect(row).toBeDefined();
    return { eventId, delivery: row! };
  };
 
  const report = (body: Record<string, unknown>) =>
    fetch(`${url}/internal/dispatch/outcome`, {
      method: "POST",
      headers: {
        "content-type": "application/json",
        authorization: `Bearer ${CREDENTIAL}`,
      },
      body: JSON.stringify(body),
    });
 
  /** Everything this suite has taken off the stream, filed by environment.
   *
   * A BUFFER rather than a filter, and the first version of this file got it
   * wrong in a way worth recording: it acknowledged every message it read and
   * returned only the ones matching the environment asked for. Reading
   * environment A's records therefore consumed and discarded environment B's, and
   * the two-tenant test failed looking for a record that had already been thrown
   * away by the assertion before it. A shared position needs shared bookkeeping. */
  const inbox = new Map<string, Record<string, unknown>[]>();
 
  /** Take whatever is available right now, file it, acknowledge it. Records from
   * other suites in this lane land here too and are acknowledged so they do not
   * come back; they simply never match an environment any test asks about. */
  const drain = async (): Promise<void> => {
    const consumer = await nats.jetstream().consumers.get(ANALYTICS_STREAM, durable);
    const messages = await consumer.fetch({ max_messages: 100, expires: 1_000 });
    for await (const msg of messages) {
      const payload = msg.json<Record<string, unknown>>();
      msg.ack();
      const key = String(payload["environment_id"]);
      inbox.set(key, [...(inbox.get(key) ?? []), payload]);
    }
  };
 
  /** Poll until this environment has at least `atLeast`, then answer with all of
   * them.
   *
   * Polls rather than peeks: the publish happens after the outcome commits, so a
   * single fetch races the broker — and a test that fetches once is testing
   * something the platform never does. Chapter 3.5's suite learned this and said
   * so in as many words. */
  const collected = async (
    environmentId: string,
    atLeast = 1,
    budgetMs = 15_000,
  ): Promise<Record<string, unknown>[]> => {
    const deadline = Date.now() + budgetMs;
    while (Date.now() < deadline) {
      await drain();
      if ((inbox.get(environmentId) ?? []).length >= atLeast) break;
    }
    return inbox.get(environmentId) ?? [];
  };
 
  /** Every environment this suite mints, so its leftovers can be settled.
   *
   * The relay's drain is GLOBAL and claims 50 due rows at a time, oldest first.
   * Reports of failure reschedule their delivery, so a suite that reports failures
   * and never delivers them leaves rows that are due again a second later and stay
   * that way. Enough of them starve a later suite's own delivery out of the window
   * — which is exactly how four `dispatcher.itest.ts` tests failed under the
   * coverage lane, with nothing delivered and no error anywhere. */
  const minted: string[] = [];
  const mintEnvironment = async (name: string) => {
    const created = await createEnvironment(db, { name });
    minted.push(created.id);
    return created;
  };
 
  beforeAll(async () => {
    process.env["RELAY_INTERNAL_CREDENTIAL"] = CREDENTIAL;
    db = createDb(createPool());
    env = await createEnvironment(db, { name: "attempts-itest" });
    minted.push(env.id);
    repo = new Repository(db, env.id);
 
    app = (
      await Test.createTestingModule({ imports: [AppModule] }).compile()
    ).createNestApplication({ logger: false });
    await app.listen(0);
    url = await app.getUrl();
 
    nats = await connect({
      servers: process.env["RELAY_NATS_URL"] ?? "nats://localhost:4222",
    });
 
    // The stream is created by the API SERVICE, not here — one definition of a
    // stream, as chapter 3.5 established for DELIVERIES. So the first outcome is
    // reported before any consumer exists, which is also the honest ordering: the
    // platform must not need a reader in order to write.
    await seedEndpoint(repo);
    const { delivery } = await deliveryFor(env.id, repo);
    await report({
      delivery_id: delivery.id,
      attempt: 1,
      status: 200,
      latency_ms: 12,
    });
 
    const jsm = await nats.jetstreamManager();
    durable = `${SUITE}-${randomUUID().slice(0, 8)}`;
    await jsm.consumers.add(ANALYTICS_STREAM, {
      durable_name: durable,
      ack_policy: AckPolicy.Explicit,
      // ALL, not New. The point of a seven-day retention is that a consumer
      // arriving late still finds what it missed, and this consumer is arriving
      // late on purpose — the record above was published before it existed.
      deliver_policy: DeliverPolicy.All,
      filter_subject: ALL_ANALYTICS_SUBJECT,
    });
  }, 60_000);
 
  afterAll(async () => {
    if (nats && !nats.isClosed()) {
      const jsm = await nats.jetstreamManager();
      // The durable goes with the run. Chapter 3.5's dispatcher suite leaked two
      // consumers per run onto a shared broker and 3.6's baseline found ninety of
      // them; this file does not add to that count.
      await jsm.consumers.delete(ANALYTICS_STREAM, durable).catch(() => undefined);
      // And the stream comes BACK, because the last test in this file deletes it.
      // Leaving it deleted made every later suite in the lane log a swallowed
      // publish failure — harmless by design, and still a whole lane of api
      // instances quietly recording nothing. The api cannot restore it itself: its
      // publisher ensures the stream when it opens a connection, and that
      // connection is cached for the process's life.
      //
      // Restored by calling THE API'S OWN `ensureAnalyticsStream`, not by declaring
      // the configuration again here. The first version of this teardown wrote its
      // own `streams.add` and left out `max_bytes`, so the stream a reader then
      // inspected with `stream-info.mjs` was the TEST's stream wearing the api's
      // name — unbounded where the api bounds it at a gigabyte. Two definitions of
      // one stream is a drift waiting for the day they disagree, and it took two
      // hours to arrive.
      await ensureAnalyticsStream(nats).catch(() => undefined);
      await nats.drain();
    }
    if (minted.length > 0) {
      const list = minted.map((id) => `'${id}'`).join(",");
      await db.execute(
        `UPDATE webhook_deliveries SET state = 'dead'
          WHERE state = 'pending' AND environment_id IN (${list})`,
      );
    }
    await app?.close();
  }, 60_000);
 
  it("invariant 1: the stream exists with the configuration the contract states", async () => {
    // Created by the api's own publisher on first use — nothing in this file
    // created it, and a test that created it would be asserting on its own work.
    const info = await (await nats.jetstreamManager()).streams.info(ANALYTICS_STREAM);
    expect(info.config.subjects).toEqual([ALL_ANALYTICS_SUBJECT]);
    expect(info.config.max_age).toBe(7 * 24 * 60 * 60 * 1_000_000_000);
    expect(info.config.discard).toBe("old");
    expect(info.config.retention).toBe("limits");
  });
 
  it("invariants 1, 2, 3: a recorded outcome publishes one event carrying the four identifiers", async () => {
    const scoped = await mintEnvironment("attempts-itest-one");
    const scopedRepo = new Repository(db, scoped.id);
    const endpoint = await seedEndpoint(scopedRepo);
    const { eventId, delivery } = await deliveryFor(scoped.id, scopedRepo);
 
    const response = await report({
      delivery_id: delivery.id,
      attempt: 1,
      status: 200,
      latency_ms: 143,
    });
    expect(response.status).toBe(200);
    expect(await response.json()).toEqual({ outcome: "delivered" });
 
    const events = await collected(scoped.id);
    expect(events).toHaveLength(1);
    const [event] = events;
    expect(event).toMatchObject({
      delivery_id: delivery.id,
      endpoint_id: endpoint.id,
      environment_id: scoped.id,
      event_id: eventId,
      attempt: 1,
      status: 200,
      latency_ms: 143,
      outcome: "delivered",
    });
    // FR-004: identifiers, statuses and durations. Nothing else, and asserted as
    // an exact key set so an added field fails here rather than shipping.
    expect(Object.keys(event!).sort()).toEqual([
      "attempt",
      "attempted_at",
      "delivery_id",
      "endpoint_id",
      "environment_id",
      "event_id",
      "latency_ms",
      "outcome",
      "status",
    ]);
    expect(typeof event!["attempted_at"]).toBe("string");
  });
 
  it("records a timeout with NO status rather than omitting the attempt", async () => {
    // FR-001's hardest case. Nothing answered, so there is no status to report —
    // and the attempt still happened, which is exactly what a customer asking
    // "what did you do on my behalf" needs to see.
    const scoped = await mintEnvironment("attempts-itest-timeout");
    const scopedRepo = new Repository(db, scoped.id);
    await seedEndpoint(scopedRepo);
    const { delivery } = await deliveryFor(scoped.id, scopedRepo);
 
    await report({
      delivery_id: delivery.id,
      attempt: 1,
      error: "timeout after 10000ms",
      latency_ms: 10_000,
    });
 
    const [event] = await collected(scoped.id);
    expect(event).toBeDefined();
    expect("status" in event!).toBe(false);
    expect(event!["error"]).toBe("timeout after 10000ms");
    expect(event!["latency_ms"]).toBe(10_000);
    expect(event!["outcome"]).toBe("rescheduled");
  });
 
  it("publishes one event per attempt across a whole exhausted schedule", async () => {
    // US1's third acceptance scenario, and the only place in the chapter that
    // drives a delivery to exhaustion on the stream. Seven attempts, seven
    // events, ascending, with the last one dead-lettered.
    //
    // The tiers are not waited out — `recordAttemptOutcome` is called directly
    // for attempts 2..7 and the schedule's clock is irrelevant to what gets
    // published. Waiting for the real 2h tier would make this test unrunnable,
    // and the thing under test is the record, not the delay.
    const scoped = await mintEnvironment("attempts-itest-full");
    const scopedRepo = new Repository(db, scoped.id);
    await seedEndpoint(scopedRepo);
    const { delivery } = await deliveryFor(scoped.id, scopedRepo);
 
    await report({
      delivery_id: delivery.id,
      attempt: 1,
      status: 500,
      latency_ms: 20,
    });
    // Attempts 2..7 go through the route too, so every one of them exercises the
    // publish. Each needs the row to be due and claimed again first.
    for (let attempt = 2; attempt <= MAX_ATTEMPTS; attempt++) {
      await db.execute(
        `UPDATE webhook_deliveries
            SET next_attempt_at = now() - interval '1 second', dispatched_at = NULL
          WHERE id = '${delivery.id}'`,
      );
      await claimOnly();
      const response = await report({
        delivery_id: delivery.id,
        attempt,
        status: 500,
        latency_ms: 20,
      });
      expect(response.status).toBe(200);
    }
 
    const events = await collected(scoped.id, MAX_ATTEMPTS, 30_000);
    expect(events).toHaveLength(MAX_ATTEMPTS);
    expect(events.map((e) => e["attempt"])).toEqual([1, 2, 3, 4, 5, 6, 7]);
    // Six reschedules and then the end of the road.
    expect(events.slice(0, MAX_ATTEMPTS - 1).map((e) => e["outcome"])).toEqual(
      Array(MAX_ATTEMPTS - 1).fill("rescheduled"),
    );
    expect(events.at(-1)!["outcome"]).toBe("dead_lettered");
  }, 60_000);
 
  it("invariant 1: a REPEATED report publishes nothing the second time", async () => {
    // The dispatcher posts, reports, then acknowledges, so a crash in the last
    // gap makes a second report ordinary. It changes no row — and nothing on the
    // analytical path deduplicates, so a second event here would show a customer
    // a retry that never happened.
    const scoped = await mintEnvironment("attempts-itest-repeat");
    const scopedRepo = new Repository(db, scoped.id);
    await seedEndpoint(scopedRepo);
    const { delivery } = await deliveryFor(scoped.id, scopedRepo);
 
    const body = {
      delivery_id: delivery.id,
      attempt: 1,
      status: 200,
      latency_ms: 30,
    };
    const first = await report(body);
    const second = await report(body);
    // The dispatcher is told the same thing both times — that is what idempotent
    // means here — so the repeat is invisible to it.
    expect(await first.json()).toEqual(await second.json());
 
    // Spend a real budget looking for a second event rather than checking once.
    const events = await collected(scoped.id, 2, 5_000);
    expect(events).toHaveLength(1);
  }, 30_000);
 
  it("FR-018: no attempt event crosses a tenant boundary", async () => {
    // Two environments, one report each, and each one's subject carries only its
    // own. The subject is the filter a future consumer will use, so a mismatch
    // between subject and payload is the shape a cross-tenant leak would take
    // here — nothing would error, and one customer's dashboard would show
    // another's traffic.
    const a = await mintEnvironment("attempts-itest-tenant-a");
    const b = await mintEnvironment("attempts-itest-tenant-b");
    const repoA = new Repository(db, a.id);
    const repoB = new Repository(db, b.id);
    await seedEndpoint(repoA);
    await seedEndpoint(repoB);
 
    const first = await deliveryFor(a.id, repoA);
    const second = await deliveryFor(b.id, repoB);
    await report({
      delivery_id: first.delivery.id,
      attempt: 1,
      status: 200,
      latency_ms: 10,
    });
    await report({
      delivery_id: second.delivery.id,
      attempt: 1,
      status: 500,
      latency_ms: 11,
    });
 
    const forA = await collected(a.id);
    const forB = await collected(b.id);
    expect(forA).toHaveLength(1);
    expect(forB).toHaveLength(1);
    expect(forA[0]!["delivery_id"]).toBe(first.delivery.id);
    expect(forB[0]!["delivery_id"]).toBe(second.delivery.id);
    // Neither environment's id appears in the other's record, in any field.
    expect(JSON.stringify(forA)).not.toContain(b.id);
    expect(JSON.stringify(forB)).not.toContain(a.id);
  }, 30_000);
 
  it("carries no payload, secret or signature, whatever the delivery held", async () => {
    // SC-006 end to end. The unit test proves `shape` is an allow-list; this
    // proves the thing actually on the stream contains none of a real delivery's
    // sensitive parts, including the endpoint's url and the event body.
    const scoped = await mintEnvironment("attempts-itest-secrets");
    const scopedRepo = new Repository(db, scoped.id);
    const secret = mintSigningSecret();
    const endpoint = await scopedRepo.createEndpoint({
      url: "https://customer.example/secret-path",
      eventTypes: ["message.created"],
      secretCiphertext: encryptSecret(secret),
    });
    const eventId = randomUUID();
    await expandEventToDeliveries(db, {
      eventId,
      environmentId: scoped.id,
      type: "message.created",
      payload: {
        id: eventId,
        type: "message.created",
        data: { text: "B2, north ramp" },
      },
    });
    await claimOnly();
    const [row] = await scopedRepo.listDeliveriesForEvent(eventId);
 
    await report({
      delivery_id: row!.id,
      attempt: 1,
      status: 200,
      latency_ms: 9,
    });
 
    const serialised = JSON.stringify(await collected(scoped.id));
    expect(serialised).toContain(endpoint.id);
    for (const forbidden of [
      secret,
      "B2, north ramp",
      "customer.example",
      "secret-path",
    ]) {
      expect(serialised).not.toContain(forbidden);
    }
  }, 30_000);
 
  it("FR-009: a customer disabling their own endpoint leaves no platform fingerprint", async () => {
    // The other half of the distinction the disable columns exist to draw, and it
    // is testable before auto-disable exists: `setEndpointEnabled(false)` must
    // leave `disabled_at` null.
    const scoped = await mintEnvironment("attempts-itest-manual");
    const scopedRepo = new Repository(db, scoped.id);
    const endpoint = await seedEndpoint(scopedRepo);
 
    await scopedRepo.setEndpointEnabled(endpoint.id, false);
 
    const [row] = (
      await db.execute(
        `SELECT enabled, disabled_at, disabled_reason,
                failure_run_started_at, failure_run_attempts
           FROM webhook_endpoints WHERE id = '${endpoint.id}'`,
      )
    ).rows as {
      enabled: boolean;
      disabled_at: string | null;
      disabled_reason: string | null;
      failure_run_started_at: string | null;
      failure_run_attempts: number | null;
    }[];
 
    expect(row!.enabled).toBe(false);
    expect(row!.disabled_at).toBeNull();
    expect(row!.disabled_reason).toBeNull();
    expect(row!.failure_run_started_at).toBeNull();
    expect(row!.failure_run_attempts).toBeNull();
  });
 
  it("persists what the endpoint answered on the delivery row", async () => {
    // The state the test event and the sweep both read (data-model.md). Chapter
    // 3.5 recorded an attempt by moving the delivery and discarded the answer;
    // this is the column that stops the sweep having to write "cause unknown".
    const scoped = await mintEnvironment("attempts-itest-last");
    const scopedRepo = new Repository(db, scoped.id);
    await seedEndpoint(scopedRepo);
    const { delivery } = await deliveryFor(scoped.id, scopedRepo);
 
    await recordAttemptOutcome(db, {
      deliveryId: delivery.id,
      attempt: 1,
      status: 503,
      latencyMs: 214,
    });
 
    const [row] = (
      await db.execute(
        `SELECT last_status, last_error, last_latency_ms
           FROM webhook_deliveries WHERE id = '${delivery.id}'`,
      )
    ).rows as {
      last_status: number | null;
      last_error: string | null;
      last_latency_ms: number | null;
    }[];
    expect(row!.last_status).toBe(503);
    expect(row!.last_latency_ms).toBe(214);
    expect(row!.last_error).toBeNull();
  });
  it("invariant 5: an outcome is recorded and answered with the ANALYTICS stream deleted", async () => {
    // THE CONSTITUTION III TEST, and the reason this chapter publishes after the
    // commit instead of inside it. Deleting the stream is a sharper instrument
    // than stopping the broker and needs no container restart: the connection is
    // healthy, the publish is refused.
    //
    // If a delivery can fail because analytics is unwell, the design is wrong —
    // not the test.
    //
    // LAST IN THE FILE ON PURPOSE. The api's publisher ensures its stream when it
    // opens a connection, and that connection is cached for the process's life —
    // so a stream deleted underneath it is NOT recreated, and every later publish
    // from this app instance fails. That is a real limitation and it is recorded
    // in the chapter rather than papered over here: an operator who deletes this
    // stream loses attempt records until the api restarts. Nothing after this test
    // may depend on the stream, so nothing is.
    const scoped = await mintEnvironment("attempts-itest-nostream");
    const scopedRepo = new Repository(db, scoped.id);
    await seedEndpoint(scopedRepo);
    const { delivery } = await deliveryFor(scoped.id, scopedRepo);
 
    const jsm = await nats.jetstreamManager();
    await jsm.streams.delete(ANALYTICS_STREAM);
 
    const started = Date.now();
    const response = await report({
      delivery_id: delivery.id,
      attempt: 1,
      status: 200,
      latency_ms: 55,
    });
    const elapsed = Date.now() - started;
 
    // Unchanged answer, unchanged row. The delivery does not care.
    expect(response.status).toBe(200);
    expect(await response.json()).toEqual({ outcome: "delivered" });
    const [row] = await scopedRepo.listDeliveriesForEvent(delivery.event_id);
    expect(row!.state).toBe("delivered");
    // And it did not sit waiting on a broker that was never going to answer. The
    // bound is generous — this asserts "not stalled", not a latency budget.
    expect(elapsed).toBeLessThan(15_000);
  }, 60_000);
});