feat(logging): give failed runs and caught failures their own error lines

A wide event carries one action and the last annotation wins, so a hop
that failed before the next provider answered, one failed account among
six, or a caught error the code carried on past was either a meta key on
someone else's info line or nothing at all. `emitSignal` writes one extra
line under a stable action and level, joined to the surrounding event by
request and trace id; `logCaught` / `caughtAs` are its warn-level shape
for a catch that degrades on purpose.

Every failed pg-boss run now writes `job.run.failed` at error with the
queue and how many runs of it in a row have failed, counted from pg-boss's
own rows (`readConsecutiveRunFailures`). `reportJobRun` takes a fan-out
pass's whole candidate list: an all-failed run is `job.run.failed`, a
partial one `job.run.partial` at warn, and the run reports `all_failed` in
its facts so a completed night with nothing done still counts toward the
streak. The repeated-failure alert keys on that one action and number.

Retention is per queue now. `failureReaderRetention` refuses a queue
retention under 96 hours (the 72-hour reader window plus a day), and a
guard refuses any `deleteAfterSeconds` that is neither that helper nor a
literal at or above the floor.

A refusal the request was meant to get (any error carrying a 4xx status)
is recorded at warn instead of error. Expired-token 401s were most of the
error lines and hid the real failures between them.

src/__tests__/job-retention-floor-guard.test.ts

@@ -0,0 +1,105 @@
+/**
+ * No queue may forget its failed jobs before the failure readers look.
+ *
+ * `job-failures.ts` reads pg-boss's own `state = 'failed'` rows over a 72-hour
+ * window: the admin status card, the per-account "last run failed" hint and
+ * the repeated-failure count behind the `job.run.failed` alert all rest on
+ * those rows still existing. pg-boss deletes a terminal row
+ * `deleteAfterSeconds` after it finished. v1.42 shortens that option on the
+ * pure-volume queues, and a value under the reader window would make a
+ * failing queue read as healthy, with nothing anywhere saying so.
+ *
+ * The rule: a `deleteAfterSeconds` in the source is either produced by
+ * `failureReaderRetention(hours)` (which throws under the floor) or written as
+ * a plain literal at or above `FAILED_ROW_AVAILABILITY_FLOOR_HOURS`. Anything
+ * else, a named constant included, fails here, because a constant's value is
+ * not checkable from the call site and the helper exists to make it so.
+ *
+ * Mutation check, run when this guard was written: a planted
+ * `deleteAfterSeconds: 48 * 60 * 60` in a job file fails the second case, and
+ * `deleteAfterSeconds: VOLUME_SECONDS` fails it too; the matcher self-test
+ * below proves the scan sees both shapes, so an empty sweep is not mistaken
+ * for a clean one.
+ */
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import { describe, expect, it } from "vitest";
+
+import { walkSourceFiles } from "./helpers/source-files";
+import { FAILED_ROW_AVAILABILITY_FLOOR_HOURS } from "@/lib/jobs/job-failures";
+
+const SRC = join(process.cwd(), "src");
+const FLOOR_SECONDS = FAILED_ROW_AVAILABILITY_FLOOR_HOURS * 3600;
+
+/** The one file allowed to spell the option out: the helper itself. */
+const HELPER_FILE = "lib/jobs/job-failures.ts";
+
+const OPTION = /\bdeleteAfterSeconds\s*:\s*([^,}\n]+)/g;
+
+/** Evaluate a sum of products of integer literals; null for anything else. */
+function literalSeconds(expression: string): number | null {
+  const cleaned = expression.trim().replace(/_/g, "");
+  if (!/^\d+(\s*[*+]\s*\d+)*$/.test(cleaned)) return null;
+  return cleaned
+    .split("+")
+    .map((term) =>
+      term
+        .split("*")
+        .map((factor) => Number(factor.trim()))
+        .reduce((a, b) => a * b, 1),
+    )
+    .reduce((a, b) => a + b, 0);
+}
+
+interface Finding {
+  file: string;
+  expression: string;
+}
+
+function scan(file: string, text: string): Finding[] {
+  const out: Finding[] = [];
+  for (const match of text.matchAll(OPTION)) {
+    out.push({ file, expression: match[1].trim() });
+  }
+  return out;
+}
+
+function violations(findings: readonly Finding[]): string[] {
+  return findings
+    .filter((f) => {
+      const seconds = literalSeconds(f.expression);
+      return seconds === null || seconds < FLOOR_SECONDS;
+    })
+    .map((f) => `${f.file}: deleteAfterSeconds: ${f.expression}`);
+}
+
+function sourceFindings(): Finding[] {
+  return walkSourceFiles(SRC, { floor: 3000 })
+    .filter((p) => !p.startsWith("generated/"))
+    .filter((p) => !p.includes("__tests__"))
+    .filter((p) => !p.endsWith(".test.ts") && !p.endsWith(".test.tsx"))
+    .filter((p) => p !== HELPER_FILE)
+    .flatMap((p) => scan(p, readFileSync(join(SRC, p), "utf8")));
+}
+
+describe("job retention floor", () => {
+  it("the matcher sees both a short literal and a named constant", () => {
+    const planted = scan(
+      "planted.ts",
+      [
+        "await boss.createQueue(Q, { deleteAfterSeconds: 48 * 60 * 60 });",
+        "await boss.createQueue(R, { deleteAfterSeconds: VOLUME_SECONDS });",
+        "await boss.createQueue(S, { deleteAfterSeconds: 7 * 24 * 3600 });",
+      ].join("\n"),
+    );
+    expect(planted).toHaveLength(3);
+    expect(violations(planted)).toEqual([
+      "planted.ts: deleteAfterSeconds: 48 * 60 * 60",
+      "planted.ts: deleteAfterSeconds: VOLUME_SECONDS",
+    ]);
+  });
+
+  it("every queue retention in the tree keeps failed rows past the reader window", () => {
+    expect(violations(sourceFindings())).toEqual([]);
+  });
+});

src/lib/jobs/__tests__/job-failures-retention.test.ts

@@ -3,8 +3,10 @@ import { join } from "node:path";
 import { describe, expect, it } from "vitest";
 
 import {
+  FAILED_ROW_AVAILABILITY_FLOOR_HOURS,
   JOB_FAILURE_WINDOW_HOURS,
-  PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS,
+  PG_BOSS_DEFAULT_FAILED_ROW_AVAILABILITY_HOURS,
+  failureReaderRetention,
 } from "@/lib/jobs/job-failures";
 
 const PG_BOSS_ROOT = join(process.cwd(), "node_modules/pg-boss");
@@ -26,10 +28,28 @@ describe("pg-boss failed-row ledger retention", () => {
     );
     expect(plans).toContain("CLOCK_FUNCTION_BODY = 'SELECT pg_catalog.now();'");
     expect(JOB_FAILURE_WINDOW_HOURS).toBeLessThan(
-      PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS,
+      PG_BOSS_DEFAULT_FAILED_ROW_AVAILABILITY_HOURS,
     );
   });
 
+  it("keeps the per-queue floor above the reader window, with a day of slack", () => {
+    expect(JOB_FAILURE_WINDOW_HOURS).toBe(72);
+    expect(FAILED_ROW_AVAILABILITY_FLOOR_HOURS).toBe(96);
+    expect(FAILED_ROW_AVAILABILITY_FLOOR_HOURS).toBeLessThanOrEqual(
+      PG_BOSS_DEFAULT_FAILED_ROW_AVAILABILITY_HOURS,
+    );
+  });
+
+  it("refuses a queue retention that would hide failures from the readers", () => {
+    expect(failureReaderRetention(96)).toEqual({ deleteAfterSeconds: 345_600 });
+    expect(failureReaderRetention(7 * 24)).toEqual({
+      deleteAfterSeconds: 604_800,
+    });
+    expect(() => failureReaderRetention(95)).toThrow(RangeError);
+    expect(() => failureReaderRetention(24)).toThrow(RangeError);
+    expect(() => failureReaderRetention(Number.NaN)).toThrow(RangeError);
+  });
+
   it("pins the queue defaults the data-backup window had to escape", () => {
     const plans = readFileSync(join(PG_BOSS_ROOT, "dist/plans.js"), "utf8");
 

src/lib/jobs/__tests__/job-run-report.test.ts

@@ -0,0 +1,142 @@
+/**
+ * v1.42 — `reportJobRun`: one line per run with every candidate's outcome,
+ * and a level that follows the run as a whole.
+ *
+ * The nightly briefing warm read as "all-failed" on seven nights because the
+ * last account of each run happened to fail. These cases pin the verdict
+ * (what counts as attempted, what makes a run failed versus partial) and the
+ * two signals an operator and the repeated-failure alert key on.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const emitSignal = vi.fn();
+const annotate = vi.fn();
+const readConsecutiveRunFailures = vi.fn(async (_queue: string) => 3);
+
+vi.mock("@/lib/logging/signal", () => ({
+  emitSignal: (...a: unknown[]) => emitSignal(...a),
+}));
+vi.mock("@/lib/logging/context", () => ({
+  annotate: (...a: unknown[]) => annotate(...a),
+}));
+vi.mock("@/lib/jobs/job-failures", () => ({
+  readConsecutiveRunFailures: (queue: string) =>
+    readConsecutiveRunFailures(queue),
+}));
+
+import { judgeJobRun, reportJobRun } from "../job-run-report";
+
+beforeEach(() => {
+  vi.clearAllMocks();
+});
+
+describe("judgeJobRun", () => {
+  it("leaves skipped and deferred candidates out of the attempted set", () => {
+    expect(
+      judgeJobRun([
+        { key: "a", outcome: "skipped" },
+        { key: "b", outcome: "deferred" },
+        { key: "c", outcome: "failed" },
+      ]),
+    ).toEqual({
+      total: 3,
+      attempted: 1,
+      failed: 1,
+      allFailed: true,
+      partial: false,
+    });
+  });
+
+  it("counts a dead credential as a failure", () => {
+    expect(
+      judgeJobRun([
+        { key: "a", outcome: "auth_failed" },
+        { key: "b", outcome: "failed" },
+      ]).allFailed,
+    ).toBe(true);
+  });
+
+  it("reads one success among failures as partial, not failed", () => {
+    const verdict = judgeJobRun([
+      { key: "a", outcome: "ok" },
+      { key: "b", outcome: "failed" },
+    ]);
+    expect(verdict.allFailed).toBe(false);
+    expect(verdict.partial).toBe(true);
+  });
+
+  it("reads a withheld result as partial: the provider answered", () => {
+    const verdict = judgeJobRun([{ key: "a", outcome: "screened" }]);
+    expect(verdict.allFailed).toBe(false);
+    expect(verdict.partial).toBe(true);
+  });
+
+  it("is neither failed nor partial when nothing was attempted", () => {
+    const verdict = judgeJobRun([{ key: "a", outcome: "skipped" }]);
+    expect(verdict.allFailed).toBe(false);
+    expect(verdict.partial).toBe(false);
+  });
+});
+
+describe("reportJobRun", () => {
+  it("emits job.run.failed at error with the streak when every attempt failed", async () => {
+    const verdict = await reportJobRun({
+      queue: "insight-pregenerate",
+      runId: "r1",
+      candidates: [
+        { key: "a", outcome: "auth_failed", cause: "all-providers-failed" },
+        { key: "b", outcome: "skipped", cause: "no-consent" },
+      ],
+    });
+    expect(verdict.allFailed).toBe(true);
+    expect(readConsecutiveRunFailures).toHaveBeenCalledWith(
+      "insight-pregenerate",
+    );
+    expect(emitSignal).toHaveBeenCalledTimes(1);
+    expect(emitSignal.mock.calls[0][0]).toMatchObject({
+      action: "job.run.failed",
+      level: "error",
+      meta: {
+        queue: "insight-pregenerate",
+        consecutiveFailures: 3,
+        failed: 1,
+        total: 1,
+        causes: ["auth_failed:all-providers-failed"],
+      },
+    });
+  });
+
+  it("emits job.run.partial at warn when only some failed", async () => {
+    await reportJobRun({
+      queue: "q",
+      runId: "r",
+      candidates: [
+        { key: "a", outcome: "ok" },
+        { key: "b", outcome: "failed", cause: "timeout" },
+      ],
+    });
+    expect(readConsecutiveRunFailures).not.toHaveBeenCalled();
+    expect(emitSignal.mock.calls[0][0]).toMatchObject({
+      action: "job.run.partial",
+      level: "warn",
+      meta: { queue: "q", failed: 1, total: 2 },
+    });
+  });
+
+  it("emits nothing extra for a clean run, and lists every outcome on the run line", async () => {
+    await reportJobRun({
+      queue: "q",
+      runId: "r",
+      candidates: [
+        { key: "a", outcome: "ok" },
+        { key: "b", outcome: "ok" },
+        { key: "c", outcome: "skipped", cause: "no-provider" },
+      ],
+    });
+    expect(emitSignal).not.toHaveBeenCalled();
+    const meta = annotate.mock.calls[0][0].meta;
+    expect(meta.outcomes).toEqual({ ok: 2, skipped: 1 });
+    expect(meta.candidate_outcomes).toHaveLength(3);
+    expect(meta.all_failed).toBe(false);
+  });
+});

src/lib/jobs/__tests__/run-job.test.ts

@@ -17,6 +17,19 @@ vi.mock("../report-worker-error", () => ({
   reportWorkerError: (...a: unknown[]) => reportWorkerError(...a),
 }));
 
+// v1.42 — the failed run's own line and the streak behind it.
+const emitSignal = vi.fn();
+const readConsecutiveRunFailures = vi.fn(async (_queue: string) => 2);
+vi.mock("@/lib/logging/signal", () => ({
+  emitSignal: (...a: unknown[]) => emitSignal(...a),
+}));
+vi.mock("../job-failures", async () => ({
+  ...(await vi.importActual<typeof import("../job-failures")>(
+    "../job-failures",
+  )),
+  readConsecutiveRunFailures: (q: string) => readConsecutiveRunFailures(q),
+}));
+
 import { jobDone, jobFailed } from "../job-outcome";
 import { JobFailure, runJob } from "../run-job";
 
@@ -171,3 +184,47 @@ describe("runJob observation (#1031)", () => {
     observe.mockRestore();
   });
 });
+
+describe("runJob failure signal (v1.42)", () => {
+  it("emits job.run.failed at error with the queue's streak for a failed outcome", async () => {
+    const wrapped = runJob("environment-fetch", async () =>
+      jobFailed("fetch_failed", new Error("open-meteo 503")),
+    );
+    await expect(wrapped(noJobs)).rejects.toBeInstanceOf(JobFailure);
+    expect(readConsecutiveRunFailures).toHaveBeenCalledWith(
+      "environment-fetch",
+    );
+    expect(emitSignal).toHaveBeenCalledTimes(1);
+    expect(emitSignal.mock.calls[0][0]).toMatchObject({
+      action: "job.run.failed",
+      level: "error",
+      meta: {
+        queue: "environment-fetch",
+        reason: "fetch_failed",
+        consecutiveFailures: 2,
+      },
+    });
+  });
+
+  it("emits it for a handler that throws, and still rethrows the original error", async () => {
+    const original = new Error("boom");
+    const wrapped = runJob("data-backup", async () => {
+      throw original;
+    });
+    await expect(wrapped(noJobs)).rejects.toBe(original);
+    expect(emitSignal.mock.calls[0][0]).toMatchObject({
+      action: "job.run.failed",
+      level: "error",
+      meta: { queue: "data-backup", reason: "handler_threw" },
+      error: original,
+    });
+  });
+
+  it("emits nothing for a successful run", async () => {
+    const wrapped = runJob("rate-limit-cleanup", async () =>
+      jobDone({ deleted: 1 }),
+    );
+    await wrapped(noJobs);
+    expect(emitSignal).not.toHaveBeenCalled();
+  });
+});

src/lib/jobs/job-failures.ts

@@ -28,22 +28,61 @@ import { prisma } from "@/lib/db";
 import { redactSecrets } from "@/lib/logging/redact";
 
 /**
- * pg-boss 12.34.0 stores terminal rows until
- * `completed_on + deletion_seconds` (`dist/plans.js`). Its queue default is
- * 604800 seconds, while `retention_seconds` defaults to 1209600 seconds; job
- * insertion copies those queue values and maintenance applies the terminal-row
- * deletion clock. The deployed queue-table audit found those same values on
- * every queue, and no job submitter overrides either option.
+ * How long pg-boss keeps a terminal row, and why the readers below can rely
+ * on it.
  *
- * Seven days is therefore the verified availability floor for failed rows.
- * The reader stays at 72 hours, safely inside that floor, and pg-boss remains
- * the single failure ledger.
+ * pg-boss 12.34.0 stores terminal rows until `completed_on +
+ * deletion_seconds` (`dist/plans.js`). The queue default is 604800 seconds
+ * (seven days), `retention_seconds` defaults to 1209600 seconds, and job
+ * insertion copies the queue's values onto each row. A queue created without
+ * a `deleteAfterSeconds` option therefore keeps its failed rows for seven
+ * days.
+ *
+ * That is no longer true for every queue. v1.42 shortens the retention of
+ * the pure-volume queues (a sampler that ticks every five minutes, the rollup
+ * recompute fan-out) so their completed rows stop bloating `job_common`. A
+ * failed row on such a queue lives only as long as that queue's own
+ * retention, and a reader that looks back 72 hours over a queue that forgets
+ * after 48 would report "nothing failed" about failures it can no longer see.
+ *
+ * So the availability is per queue now, and it has a floor: no queue may
+ * keep its terminal rows for less than `FAILED_ROW_AVAILABILITY_FLOOR_HOURS`,
+ * which is the 72-hour reader window plus a day of slack for a run that
+ * fails just before a maintenance pass. Every queue option that shortens
+ * retention goes through `failureReaderRetention`, which refuses a value
+ * under the floor, and `src/__tests__/job-retention-floor-guard.test.ts`
+ * refuses a `deleteAfterSeconds` anywhere that is neither a call to it nor a
+ * literal at or above the floor. pg-boss remains the single failure ledger.
  */
-export const PG_BOSS_FAILED_ROW_AVAILABILITY_HOURS = 7 * 24;
+export const PG_BOSS_DEFAULT_FAILED_ROW_AVAILABILITY_HOURS = 7 * 24;
 
 /** How far back a failure still counts as news. */
 export const JOB_FAILURE_WINDOW_HOURS = 72;
 
+/**
+ * The shortest terminal-row retention any queue may have: the reader window
+ * plus a day. See the block above.
+ */
+export const FAILED_ROW_AVAILABILITY_FLOOR_HOURS =
+  JOB_FAILURE_WINDOW_HOURS + 24;
+
+/**
+ * The `deleteAfterSeconds` queue option for a queue that keeps its terminal
+ * rows for `hours`. Throws below the floor, so a retention change that would
+ * blind the failure readers fails at boot and in the test suite rather than
+ * silently three days later.
+ */
+export function failureReaderRetention(hours: number): {
+  deleteAfterSeconds: number;
+} {
+  if (!Number.isFinite(hours) || hours < FAILED_ROW_AVAILABILITY_FLOOR_HOURS) {
+    throw new RangeError(
+      `queue retention of ${hours} h is below the ${FAILED_ROW_AVAILABILITY_FLOOR_HOURS} h the failure readers need`,
+    );
+  }
+  return { deleteAfterSeconds: Math.round(hours * 3600) };
+}
+
 /** Queues named in one report — enough to see a pattern, bounded for the wire. */
 const MAX_QUEUES = 20;
 
@@ -373,3 +412,52 @@ export async function readQueueRunningSince(
     return null;
   }
 }
+
+/** Newest terminal rows `readConsecutiveRunFailures` looks at. */
+const CONSECUTIVE_LOOKBACK_ROWS = 50;
+
+/**
+ * How many runs of `queue` in a row ended in failure, newest first, counting
+ * the run that is failing right now (so the result is at least 1).
+ *
+ * A run counts as failed when pg-boss failed it (`state = 'failed'`), or when
+ * it completed but said in its own facts that every candidate it attempted
+ * failed (`did.all_failed`, written through `reportJobRun`). A fan-out pass
+ * like the nightly briefing warm completes even when nothing worked, because
+ * failing the queue would retry the whole cohort over one account; its
+ * all-failed nights are still a streak an operator wants to hear about.
+ *
+ * Completed runs that do not report `all_failed` at all (a queue's on-demand
+ * single-account jobs share it with the nightly pass) are not evidence either
+ * way and are skipped. The streak ends at the first completed run that says
+ * it did not all fail.
+ *
+ * Fails soft to 1: when the queue schema cannot be read, this run is still a
+ * failure, and one is the only count that is known.
+ */
+export async function readConsecutiveRunFailures(
+  queue: string,
+): Promise<number> {
+  try {
+    const rows = await prisma.$queryRaw<
+      Array<{ state: string; all_failed: string | null }>
+    >`
+      SELECT state, output->'did'->>'all_failed' AS all_failed
+      FROM pgboss.job
+      WHERE name = ${queue}
+        AND state IN ('completed', 'failed')
+        AND completed_on IS NOT NULL
+        AND (state = 'failed' OR output->'did'->>'all_failed' IS NOT NULL)
+      ORDER BY completed_on DESC
+      LIMIT ${CONSECUTIVE_LOOKBACK_ROWS}
+    `;
+    let streak = 1;
+    for (const row of rows) {
+      if (row.state === "failed" || row.all_failed === "true") streak += 1;
+      else break;
+    }
+    return streak;
+  } catch {
+    return 1;
+  }
+}

src/lib/jobs/job-outcome.ts

@@ -47,6 +47,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
   "ai_runs_deleted",
   "ai_runs_timed_out",
   "ai_runs_worker_unavailable",
+  // v1.42 — every candidate a fan-out pass attempted failed (`reportJobRun`).
+  "all_failed",
   // An admission delivery that found its import still running from an
   // earlier delivery of the same job and did nothing.
   "already_running",

src/lib/jobs/job-run-report.ts

@@ -3,16 +3,31 @@
  *
  * A pass over many accounts used to annotate its wide event once per
  * candidate, and the last write won: a run where six accounts worked and one
- * failed read as whatever the last account did. `reportJobRun` takes the
- * whole candidate list instead, so the run's line carries every outcome and
- * the level follows the worst of them.
+ * failed read as whatever the last account did. The nightly briefing warm read
+ * as `all-failed` on seven nights in a row when it had in fact completed every
+ * night, because the last account of each run happened to fail. `reportJobRun`
+ * takes the whole candidate list instead, so the run's line carries every
+ * outcome and the level follows the run as a whole:
  *
- * Contract stub: the signature is fixed for its callers (the insight
- * pre-generation, the tombstone purge, the measurement maintenance). Until
- * the implementation lands it annotates the run's counts the way the callers
- * do today.
+ *   - every attempted candidate failed → `job.run.failed` at `error`, with how
+ *     many runs of the queue in a row have now failed (`consecutiveFailures`,
+ *     the number the repeated-failure alert keys on);
+ *   - some failed, or a result was withheld → `job.run.partial` at `warn`;
+ *   - otherwise nothing extra: the run's own line says what it did.
+ *
+ * "Attempted" leaves out `skipped` and `deferred` candidates: an account
+ * without consent or one the job budget did not reach is not a failure of the
+ * run. `screened` (the model answered and the safety screen withheld it) is
+ * attempted but is not a provider failure, so a run of only screened
+ * candidates is partial, not failed.
+ *
+ * The verdict comes back to the caller, which writes `all_failed` into its
+ * `jobDone` facts. That fact is what lets `readConsecutiveRunFailures` count a
+ * completed-but-all-failed night as part of a streak.
  */
 import { annotate } from "@/lib/logging/context";
+import { emitSignal } from "@/lib/logging/signal";
+import { readConsecutiveRunFailures } from "@/lib/jobs/job-failures";
 
 /** How one candidate of a run ended. */
 export type JobRunOutcome =
@@ -26,20 +41,118 @@ export interface JobRunCandidate {
   cause?: string;
 }
 
-export function reportJobRun(input: {
+export interface JobRunVerdict {
+  /** Candidates the run looked at. */
+  total: number;
+  /** Candidates it actually attempted (not skipped, not deferred). */
+  attempted: number;
+  /** `failed` + `auth_failed`. */
+  failed: number;
+  /** Every attempted candidate failed (and at least one was attempted). */
+  allFailed: boolean;
+  /** Something failed or was withheld, but not everything. */
+  partial: boolean;
+}
+
+/** Candidates listed on the run's line; the counts always cover all of them. */
+const MAX_LISTED_CANDIDATES = 50;
+
+const FAILED: ReadonlySet<JobRunOutcome> = new Set(["failed", "auth_failed"]);
+const NOT_ATTEMPTED: ReadonlySet<JobRunOutcome> = new Set([
+  "skipped",
+  "deferred",
+]);
+
+/** The verdict alone, without logging. Pure, for callers and tests. */
+export function judgeJobRun(
+  candidates: readonly JobRunCandidate[],
+): JobRunVerdict {
+  const total = candidates.length;
+  const attempted = candidates.filter(
+    (c) => !NOT_ATTEMPTED.has(c.outcome),
+  ).length;
+  const failed = candidates.filter((c) => FAILED.has(c.outcome)).length;
+  const screened = candidates.filter((c) => c.outcome === "screened").length;
+  const allFailed = attempted > 0 && failed === attempted;
+  return {
+    total,
+    attempted,
+    failed,
+    allFailed,
+    partial: !allFailed && (failed > 0 || screened > 0),
+  };
+}
+
+export async function reportJobRun(input: {
   queue: string;
   runId: string;
   candidates: readonly JobRunCandidate[];
-}): void {
-  const failed = input.candidates.filter(
-    (c) => c.outcome === "failed" || c.outcome === "auth_failed",
-  ).length;
+}): Promise<JobRunVerdict> {
+  const verdict = judgeJobRun(input.candidates);
+  const byOutcome: Record<string, number> = {};
+  for (const c of input.candidates) {
+    byOutcome[c.outcome] = (byOutcome[c.outcome] ?? 0) + 1;
+  }
+
   annotate({
     meta: {
       queue: input.queue,
       runId: input.runId,
-      candidates: input.candidates.length,
-      failed,
+      candidates: verdict.total,
+      attempted: verdict.attempted,
+      failed: verdict.failed,
+      all_failed: verdict.allFailed,
+      outcomes: byOutcome,
+      candidate_outcomes: input.candidates
+        .slice(0, MAX_LISTED_CANDIDATES)
+        .map((c) => ({
+          key: c.key,
+          outcome: c.outcome,
+          ...(c.cause ? { cause: c.cause } : {}),
+        })),
     },
   });
+
+  if (verdict.allFailed) {
+    const consecutiveFailures = await readConsecutiveRunFailures(input.queue);
+    emitSignal({
+      action: "job.run.failed",
+      level: "error",
+      meta: {
+        queue: input.queue,
+        runId: input.runId,
+        consecutiveFailures,
+        failed: verdict.failed,
+        total: verdict.attempted,
+        causes: distinctCauses(input.candidates),
+      },
+    });
+  } else if (verdict.partial) {
+    emitSignal({
+      action: "job.run.partial",
+      level: "warn",
+      meta: {
+        queue: input.queue,
+        runId: input.runId,
+        failed: verdict.failed,
+        total: verdict.attempted,
+        outcomes: byOutcome,
+        causes: distinctCauses(input.candidates),
+      },
+    });
+  }
+  return verdict;
+}
+
+/** The distinct causes of the run's failed or withheld candidates, bounded. */
+function distinctCauses(candidates: readonly JobRunCandidate[]): string[] {
+  const causes = new Set<string>();
+  for (const c of candidates) {
+    if (c.outcome === "ok" || NOT_ATTEMPTED.has(c.outcome) || !c.cause) {
+      continue;
+    }
+    causes.add(`${c.outcome}:${c.cause}`);
+    if (causes.size >= 10) break;
+  }
+  return [...causes];
 }

src/lib/jobs/run-job.ts

@@ -29,6 +29,8 @@ import {
 } from "./job-outcome";
 import { reportWorkerError } from "./report-worker-error";
 import { observeJob } from "./job-observer";
+import { readConsecutiveRunFailures } from "./job-failures";
+import { emitSignal } from "@/lib/logging/signal";
 
 /** A converted handler: it says what it did instead of just finishing. */
 export type JobHandler<T> = (jobs: Job<T>[]) => Promise<JobOutcome>;
@@ -67,14 +69,34 @@ function factMeta(did: JobFacts | undefined): Record<string, JobFact> {
   return meta;
 }
 
+/**
+ * The failed run's own line: `job.run.failed` at `error`, naming the queue and
+ * how many runs of it in a row have now failed. Every queue reaches it, so
+ * the repeated-failure alert keys on one action and one number instead of on
+ * the handler's own task name, which differs per queue, or on the stderr line
+ * `reportWorkerError` writes, which is not structured.
+ */
+async function signalRunFailed(
+  queue: string,
+  reason: string,
+  error: unknown,
+): Promise<void> {
+  const consecutiveFailures = await readConsecutiveRunFailures(queue);
+  emitSignal({
+    action: "job.run.failed",
+    level: "error",
+    meta: { queue, reason, consecutiveFailures },
+    error,
+  });
+}
+
 /**
  * Wrap a handler for `boss.work`. Resolves with the bounded, redacted
  * persistence representation on `ok: true`; reports and rethrows on
  * `ok: false`.
  *
- * A handler that throws is left alone — the throw already fails the job, and
- * re-wrapping it would only bury the original stack. `runJob` exists for the
- * case the throw never happens.
+ * A handler that throws keeps its own error — re-wrapping it would only bury
+ * the original stack — but its run still gets the `job.run.failed` line.
  */
 export function runJob<J>(
   queue: string,
@@ -83,11 +105,17 @@ export function runJob<J>(
   return async (jobs: J[]): Promise<SerializedJobOutcome> => {
     // Observed, so a long run, an expiry and a cut-off leave a line in the
     // log naming the queue and how far it got (see `job-observer.ts`).
-    const outcome = await observeJob(
-      queue,
-      jobs as ReadonlyArray<{ id?: string; expireInSeconds?: number }>,
-      () => handler(jobs),
-    );
+    let outcome: JobOutcome;
+    try {
+      outcome = await observeJob(
+        queue,
+        jobs as ReadonlyArray<{ id?: string; expireInSeconds?: number }>,
+        () => handler(jobs),
+      );
+    } catch (error) {
+      await signalRunFailed(queue, "handler_threw", error);
+      throw error;
+    }
     if (outcome.ok) return serializeJobOutcome(outcome);
 
     const failure = new JobFailure(queue, outcome.reason, outcome.cause);
@@ -102,6 +130,7 @@ export function runJob<J>(
       ...factMeta(outcome.did),
       ...(causeMessage === undefined ? {} : { cause: causeMessage }),
     });
+    await signalRunFailed(queue, outcome.reason, failure);
 
     throw failure;
   };

src/lib/logging/__tests__/signal.test.ts

@@ -0,0 +1,101 @@
+/**
+ * v1.42 — the logging pieces the error-log discipline rests on.
+ *
+ *   - `emitSignal` writes its own line with a stable action and level, joined
+ *     to the surrounding event by request and trace id, and leaves that event
+ *     alone.
+ *   - `setError` keeps `error` for real failures and records a refusal the
+ *     request was meant to get (any 4xx) at `warn`. Expired-token 401s made up
+ *     most of the error lines and hid the real ones.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const emitted: Array<Record<string, unknown>> = [];
+vi.mock("../transports", () => ({
+  emitIfSampled: (event: Record<string, unknown>) => emitted.push(event),
+}));
+
+import { WideEventBuilder } from "../event-builder";
+import { eventStorage } from "../context";
+import { emitSignal, logCaught } from "../signal";
+
+beforeEach(() => {
+  emitted.length = 0;
+});
+
+describe("emitSignal", () => {
+  it("writes one line with its own action and level, joined to the parent by id", () => {
+    const parent = new WideEventBuilder("http");
+    parent.setAction({ name: "insights.generate" });
+    eventStorage.run(parent, () =>
+      emitSignal({
+        action: "ai.chain.link_failed",
+        level: "error",
+        meta: { provider: "admin-openai", status: 500 },
+      }),
+    );
+    expect(emitted).toHaveLength(1);
+    const line = emitted[0] as {
+      action: { name: string };
+      level: string;
+      request_id: string;
+      trace_id: string;
+      meta: Record<string, unknown>;
+    };
+    expect(line.action.name).toBe("ai.chain.link_failed");
+    expect(line.level).toBe("error");
+    expect(line.request_id).toBe(parent.getRequestId());
+    expect(line.trace_id).toBe(parent.getTraceId());
+    expect(line.meta).toMatchObject({ provider: "admin-openai", status: 500 });
+    // The parent keeps its own action and level.
+    expect(parent.toJSON().action?.name).toBe("insights.generate");
+    expect(parent.getLevel()).toBe("info");
+  });
+
+  it("logCaught carries the error's type and a bounded, redacted message at warn", () => {
+    logCaught(
+      "jobs.enqueue.failed",
+      new Error(`boom Bearer sk-abcdefghijklmnopqrstuvwxyz ${"x".repeat(500)}`),
+      { queue: "insight-pregenerate" },
+    );
+    const line = emitted[0] as { level: string; meta: Record<string, string> };
+    expect(line.level).toBe("warn");
+    expect(line.meta.error_type).toBe("Error");
+    expect(line.meta.error_message.length).toBeLessThanOrEqual(240);
+    expect(line.meta.error_message).not.toContain("sk-abcdefghijklmnop");
+  });
+
+  it("never throws, even when emitting fails", () => {
+    expect(() =>
+      emitSignal({
+        action: "x.y.z",
+        level: "warn",
+        meta: {
+          get bad(): never {
+            throw new Error("getter");
+          },
+        },
+      }),
+    ).not.toThrow();
+  });
+});
+
+describe("setError level", () => {
+  it("records a 4xx refusal at warn", () => {
+    const evt = new WideEventBuilder("http");
+    evt.setError(
+      Object.assign(new Error("Token expired"), { statusCode: 401 }),
+    );
+    expect(evt.getLevel()).toBe("warn");
+    expect(evt.toJSON().error?.message).toBe("Token expired");
+  });
+
+  it("keeps a real failure at error", () => {
+    const evt = new WideEventBuilder("http");
+    evt.setError(new Error("database unreachable"));
+    expect(evt.getLevel()).toBe("error");
+    const five = new WideEventBuilder("http");
+    five.setError(Object.assign(new Error("bad gateway"), { statusCode: 502 }));
+    expect(five.getLevel()).toBe("error");
+  });
+});

src/lib/logging/event-builder.ts

@@ -167,7 +167,18 @@ export class WideEventBuilder {
   }
 
   setError(err: unknown): this {
-    this.elevateLevel("error");
+    // v1.42 — a refusal the request was meant to get (an expired session,
+    // a missing permission, a validation error: anything that carries a 4xx
+    // status) is a `warn`, not an `error`. Every 401 from an expired token
+    // used to land at `error`, about nine in ten of all error lines, so a
+    // query or an alert on `level=error` could not see the real failures
+    // between them. The error details are recorded either way.
+    const status = (err as { statusCode?: unknown } | null)?.statusCode;
+    this.elevateLevel(
+      typeof status === "number" && status >= 400 && status < 500
+        ? "warn"
+        : "error",
+    );
     if (err instanceof Error) {
       this.event.error = {
         type: err.constructor.name,

src/lib/logging/signal.ts

@@ -0,0 +1,99 @@
+import { WideEventBuilder } from "./event-builder";
+import { getEvent } from "./context";
+import { emitIfSampled } from "./transports";
+
+/**
+ * A failure that deserves its own log line, under its own action name.
+ *
+ * A wide event carries one action, and the last `annotate()` to name one wins.
+ * That is the right shape for "what did this request do", and the wrong one
+ * for "what went wrong on the way": a provider hop that failed before the
+ * next one answered, a job candidate that failed among six that worked, a
+ * Coach turn that ran out of budget, a caught error the code chose to carry
+ * on past. Each of those used to be either a meta key on someone else's line
+ * at level `info`, or nothing at all, so a query for `level=error` could not
+ * find it and an alert rule had no stable name to key on.
+ *
+ * `emitSignal` writes one extra line with the action, the level and a small
+ * pinned-shape meta bag. It carries the surrounding event's request and trace
+ * ids, so the signal and the line it happened inside join on one id; the
+ * surrounding event itself is not touched (its own level still describes its
+ * own outcome).
+ *
+ * Levels: `error` for a real failure nobody downstream recovered from, `warn`
+ * for one the code recovered from (a fallback answered, a degraded result was
+ * served) or a refusal by design that an operator should still be able to
+ * count. Nothing here is for `info`: an expected outcome belongs on the
+ * request's own line.
+ *
+ * Never throws. A log line must not break the work it describes.
+ */
+export interface SignalInput {
+  /** Stable `<surface>.<noun>.<verb>` name, e.g. `ai.chain.link_failed`. */
+  action: string;
+  level: "warn" | "error";
+  /** Scalars and small arrays only; redacted on the way in like any meta. */
+  meta?: Record<string, unknown>;
+  /** The caught value, when there is one. Only its type and message leave. */
+  error?: unknown;
+}
+
+/** Longest error message a signal carries. */
+const MAX_MESSAGE_CHARS = 240;
+
+export function emitSignal(input: SignalInput): void {
+  try {
+    const event = new WideEventBuilder("background");
+    const parent = getEvent();
+    if (parent) {
+      event.setRequestId(parent.getRequestId());
+      event.setTraceId(parent.getTraceId());
+    }
+    event.setBackground({ task_name: input.action });
+    event.setAction({ name: input.action });
+    for (const [key, value] of Object.entries(input.meta ?? {})) {
+      event.addMeta(key, value);
+    }
+    if (input.error !== undefined) {
+      const err = input.error;
+      event.addMeta(
+        "error_type",
+        err instanceof Error ? err.name || "Error" : typeof err,
+      );
+      const message = err instanceof Error ? err.message : String(err);
+      event.addMeta("error_message", message.slice(0, MAX_MESSAGE_CHARS));
+    }
+    event.elevateLevel(input.level);
+    event.finish();
+    emitIfSampled(event.toJSON());
+  } catch {
+    // A log line must never break the work it describes.
+  }
+}
+
+/**
+ * The common case of a `catch` that carries on: say what failed, at `warn`,
+ * under a stable action, with the caught error attached. Use it where the
+ * code deliberately degrades (an enqueue that the next tick repeats, a
+ * best-effort side write) instead of dropping the error or folding it into an
+ * `info` annotation.
+ */
+export function logCaught(
+  action: string,
+  error: unknown,
+  meta?: Record<string, unknown>,
+): void {
+  emitSignal({ action, level: "warn", error, meta });
+}
+
+/**
+ * `logCaught` shaped for a promise's `.catch`: `.catch(caughtAs("x.y.failed"))`
+ * in place of `.catch(() => {})`. The promise still settles, so a best-effort
+ * side write stays best-effort, and its failure leaves a `warn` line.
+ */
+export function caughtAs(
+  action: string,
+  meta?: Record<string, unknown>,
+): (error: unknown) => void {
+  return (error: unknown) => logCaught(action, error, meta);
+}

src/lib/logging/types.ts

@@ -115,6 +115,12 @@ export interface WideEvent {
     duration_ms: number;
     status?: number;
     error?: string;
+    /**
+     * v1.42 — an expected non-2xx the caller recovers from by design (a
+     * filter dialect probe), recorded without `error` so a query for
+     * failed calls does not find it.
+     */
+    note?: string;
   }>;
 
   // Hintergrundaufgabe (nur bei kind === "background")

tests/integration/job-failures.test.ts

@@ -17,6 +17,7 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
 import {
   readActiveJobsStartedBefore,
+  readConsecutiveRunFailures,
   readFailingQueues,
   readQueueFailureForUser,
 } from "@/lib/jobs/job-failures";
@@ -189,6 +190,46 @@ describe("readActiveJobsStartedBefore (#1031)", () => {
   });
 });
 
+describe("readConsecutiveRunFailures (v1.42)", () => {
+  const STREAK_QUEUE = "insight-pregenerate";
+
+  /** Complete one job with the facts a handler returned. */
+  async function completeOneJob(did: Record<string, unknown>) {
+    const jobId = await boss.send(STREAK_QUEUE, {}, { retryLimit: 0 });
+    const [job] = await boss.fetch(STREAK_QUEUE);
+    if (!job || job.id !== jobId) throw new Error("nothing to fetch");
+    await boss.complete(STREAK_QUEUE, job.id, { ok: true, did });
+    return jobId!;
+  }
+
+  /** Pin the finishing order so the test does not race the clock. */
+  async function finishedMinutesAgo(jobId: string, minutes: number) {
+    await getPrismaClient().$executeRawUnsafe(
+      `UPDATE pgboss.job SET completed_on = now() - make_interval(mins => ${minutes}) WHERE id = '${jobId}'`,
+    );
+  }
+
+  it("counts failed runs and all-failed nights, skips runs that do not say, and stops at a good night", async () => {
+    await boss.createQueue(STREAK_QUEUE);
+    // Oldest first: a good night, then a pg-boss failure, an on-demand job
+    // that reports nothing about the run, and an all-failed night.
+    await finishedMinutesAgo(await completeOneJob({ all_failed: false }), 40);
+    await finishedMinutesAgo(
+      await failOneJob(STREAK_QUEUE, {}, "worker died"),
+      30,
+    );
+    await finishedMinutesAgo(await completeOneJob({ forced: 1 }), 20);
+    await finishedMinutesAgo(await completeOneJob({ all_failed: true }), 10);
+
+    // Two earlier failures plus the run failing now.
+    await expect(readConsecutiveRunFailures(STREAK_QUEUE)).resolves.toBe(3);
+  });
+
+  it("answers 1 for a queue with no history: the run failing now", async () => {
+    await expect(readConsecutiveRunFailures("never-ran")).resolves.toBe(1);
+  });
+});
+
 describe("no queue schema at all", () => {
   it("answers null rather than throwing, so the status page still renders", async () => {
     const prisma = getPrismaClient();