feat(labeler): prolonged-error publisher notice via reconciliation cron (W10.5 follow-up)

apps/labeler/migrations/0010_assessment_error_escalations.sql

@@ -0,0 +1,36 @@
+-- Prolonged-error escalation tracking (plan W10.5 follow-up, ratified
+-- 2026-07-17). Drives the two-stage escalation ladder the reconciliation cron
+-- applies to an assessment stuck in the terminal `error` state (the labeler
+-- failed to assess, transient retries exhausted) that no newer run has
+-- superseded: at 24h an operator alert (an `operational_events` row) so
+-- operators can triage an infra-vs-publisher cause, then at 72h the publisher
+-- notice if the error is still the live run.
+--
+-- One row per escalated assessment makes each stage fire-once across the
+-- 5-minute cron ticks: the cron upserts the row, raises the operator alert once
+-- when `operator_alerted_at_epoch_ms` is null, and sends the publisher notice
+-- once past 72h when `publisher_notified_at_epoch_ms` is null. MUTABLE — the two
+-- mark columns flip from null to stamped — so no immutability trigger, mirroring
+-- the `notifications`/`notification_outbox` mutable-outbox tables.
+--
+-- Timestamp columns queries order/compare on carry an integer `*_epoch_ms`
+-- sibling (RFC 3339 strings compare incorrectly across timezone offsets in SQL),
+-- matching 0003-0008.
+
+CREATE TABLE assessment_error_escalations (
+	assessment_id TEXT PRIMARY KEY REFERENCES assessments(id),
+	subject_uri TEXT NOT NULL,
+	subject_cid TEXT NOT NULL,
+	operator_alerted_at_epoch_ms INTEGER,
+	publisher_notified_at_epoch_ms INTEGER,
+	created_at TEXT NOT NULL,
+	created_at_epoch_ms INTEGER NOT NULL
+);
+
+-- The escalation scan filters `state = 'error' AND completed_at_epoch_ms <= ?`
+-- and orders by `completed_at_epoch_ms`; the existing `idx_assessments_state_created`
+-- covers `(state, created_at_epoch_ms)`, not `completed_at`. Partial on the error
+-- state keeps this compact (errors are a small minority) and matches the cron's
+-- filter and order exactly, so a 5-minute tick seeks instead of full-scanning.
+CREATE INDEX idx_assessments_error_completed
+	ON assessments(completed_at_epoch_ms) WHERE state = 'error';

apps/labeler/src/assessment-error-escalations.ts

@@ -0,0 +1,183 @@
+/**
+ * Persistence for the prolonged-error escalation ladder (plan W10.5 follow-up).
+ * `assessment_error_escalations` (migration 0010) tracks, per errored
+ * assessment, whether the 24h operator alert and the 72h publisher notice have
+ * already fired — the tracking table that makes each stage idempotent across the
+ * 5-minute reconciliation cron ticks.
+ *
+ * The escalation query deliberately does NOT reuse the pointer-based
+ * `isSuperseded` (an `error` run never moves the `current_assessments` pointer,
+ * spec §10): supersession here means simply "a newer run exists for the same
+ * `(uri, cid)`", detected by `created_at_epoch_ms`.
+ */
+
+import { PROLONGED_ERROR_OPERATOR_THRESHOLD_MS, PROLONGED_ERROR_SCAN_BATCH } from "./constants.js";
+
+export interface EscalatableError {
+	id: string;
+	uri: string;
+	cid: string;
+	completedAtEpochMs: number;
+}
+
+export interface AssessmentErrorEscalation {
+	assessmentId: string;
+	subjectUri: string;
+	subjectCid: string;
+	operatorAlertedAtEpochMs: number | null;
+	publisherNotifiedAtEpochMs: number | null;
+	createdAtEpochMs: number;
+}
+
+interface EscalatableErrorRow {
+	id: string;
+	uri: string;
+	cid: string;
+	completed_at_epoch_ms: number;
+}
+
+interface EscalationRow {
+	assessment_id: string;
+	subject_uri: string;
+	subject_cid: string;
+	operator_alerted_at_epoch_ms: number | null;
+	publisher_notified_at_epoch_ms: number | null;
+	created_at_epoch_ms: number;
+}
+
+/**
+ * Terminal `error` assessments whose `completed_at` is at least the operator
+ * threshold (24h) old and that no newer run has superseded — the candidate set
+ * for the escalation ladder. Skips subjects tombstoned at the source (a deleted
+ * release warrants no publisher chase) and rows whose escalation is already
+ * complete (both marks stamped), so a fully-escalated row leaves the
+ * `completed_at`-ordered window and newer errors behind it are reached during a
+ * backlog. Capped at {@link PROLONGED_ERROR_SCAN_BATCH}; a full page is logged so
+ * a persistent backlog is visible.
+ */
+export async function findEscalatableErrors(
+	db: D1Database,
+	now: Date,
+): Promise<EscalatableError[]> {
+	const operatorBefore = now.getTime() - PROLONGED_ERROR_OPERATOR_THRESHOLD_MS;
+	const rows = await db
+		.prepare(
+			`SELECT a.id, a.uri, a.cid, a.completed_at_epoch_ms
+			 FROM assessments a
+			 JOIN subjects s ON s.uri = a.uri AND s.cid = a.cid
+			 LEFT JOIN assessment_error_escalations e ON e.assessment_id = a.id
+			 WHERE a.state = 'error'
+			   AND a.completed_at_epoch_ms IS NOT NULL
+			   AND a.completed_at_epoch_ms <= ?
+			   AND s.deleted_at IS NULL
+			   AND (
+				e.assessment_id IS NULL
+				OR e.operator_alerted_at_epoch_ms IS NULL
+				OR e.publisher_notified_at_epoch_ms IS NULL
+			   )
+			   AND NOT EXISTS (
+				SELECT 1 FROM assessments b
+				WHERE b.uri = a.uri AND b.cid = a.cid AND b.created_at_epoch_ms > a.created_at_epoch_ms
+			   )
+			 ORDER BY a.completed_at_epoch_ms ASC
+			 LIMIT ?`,
+		)
+		.bind(operatorBefore, PROLONGED_ERROR_SCAN_BATCH)
+		.all<EscalatableErrorRow>();
+	const results = rows.results ?? [];
+	if (results.length >= PROLONGED_ERROR_SCAN_BATCH)
+		console.error("[labeler] prolonged-error escalation scan hit the batch cap", {
+			cap: PROLONGED_ERROR_SCAN_BATCH,
+		});
+	return results.map((row) => ({
+		id: row.id,
+		uri: row.uri,
+		cid: row.cid,
+		completedAtEpochMs: row.completed_at_epoch_ms,
+	}));
+}
+
+/** Guarantee a tracking row exists for an errored assessment, leaving both mark
+ * columns null. Idempotent: a repeated cron tick keeps the first row's marks. */
+export async function ensureEscalationRow(
+	db: D1Database,
+	input: { assessmentId: string; subjectUri: string; subjectCid: string; now: Date },
+): Promise<void> {
+	await db
+		.prepare(
+			`INSERT INTO assessment_error_escalations
+				(assessment_id, subject_uri, subject_cid, created_at, created_at_epoch_ms)
+			 VALUES (?, ?, ?, ?, ?)
+			 ON CONFLICT(assessment_id) DO NOTHING`,
+		)
+		.bind(
+			input.assessmentId,
+			input.subjectUri,
+			input.subjectCid,
+			input.now.toISOString(),
+			input.now.getTime(),
+		)
+		.run();
+}
+
+export async function getEscalation(
+	db: D1Database,
+	assessmentId: string,
+): Promise<AssessmentErrorEscalation | null> {
+	const row = await db
+		.prepare(
+			`SELECT assessment_id, subject_uri, subject_cid,
+			 operator_alerted_at_epoch_ms, publisher_notified_at_epoch_ms, created_at_epoch_ms
+			 FROM assessment_error_escalations WHERE assessment_id = ?`,
+		)
+		.bind(assessmentId)
+		.first<EscalationRow>();
+	return row
+		? {
+				assessmentId: row.assessment_id,
+				subjectUri: row.subject_uri,
+				subjectCid: row.subject_cid,
+				operatorAlertedAtEpochMs: row.operator_alerted_at_epoch_ms,
+				publisherNotifiedAtEpochMs: row.publisher_notified_at_epoch_ms,
+				createdAtEpochMs: row.created_at_epoch_ms,
+			}
+		: null;
+}
+
+/**
+ * The statement that stamps `operator_alerted_at`, guarded on it still being
+ * null. Returned rather than executed so the caller can batch it atomically with
+ * the operational-event insert — a crash can then never leave the alert raised
+ * without the mark (which would re-alert on the next tick).
+ */
+export function buildMarkOperatorAlerted(
+	db: D1Database,
+	assessmentId: string,
+	now: Date,
+): D1PreparedStatement {
+	return db
+		.prepare(
+			`UPDATE assessment_error_escalations
+			 SET operator_alerted_at_epoch_ms = ?
+			 WHERE assessment_id = ? AND operator_alerted_at_epoch_ms IS NULL`,
+		)
+		.bind(now.getTime(), assessmentId);
+}
+
+/** Stamp `publisher_notified_at`, guarded on it still being null. Runs after the
+ * notice trigger returns; the notice's own `(issuance, id)` dedup makes a crash
+ * between send and mark self-heal (the next tick re-sends nothing, then marks). */
+export async function markPublisherNotified(
+	db: D1Database,
+	assessmentId: string,
+	now: Date,
+): Promise<void> {
+	await db
+		.prepare(
+			`UPDATE assessment_error_escalations
+			 SET publisher_notified_at_epoch_ms = ?
+			 WHERE assessment_id = ? AND publisher_notified_at_epoch_ms IS NULL`,
+		)
+		.bind(now.getTime(), assessmentId)
+		.run();
+}

apps/labeler/src/constants.ts

@@ -40,3 +40,18 @@ export const NOTIFICATION_MAX_SEND_ATTEMPTS = 5;
 export const NOTIFICATION_STUCK_PENDING_MS = 15 * 60 * 1000;
 export const NOTIFICATION_RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
 export const NOTIFICATION_SWEEP_BATCH = 200;
+
+/**
+ * Prolonged-error escalation thresholds (plan W10.5 follow-up). Versioned as a
+ * set: bump together if the ladder is retuned. A terminal `error` assessment
+ * (retries exhausted) that no newer run has superseded escalates in two stages
+ * measured from `completed_at_epoch_ms`: at {@link PROLONGED_ERROR_OPERATOR_THRESHOLD_MS}
+ * (24h) an operator alert so operators can triage an infra-vs-publisher cause;
+ * at {@link PROLONGED_ERROR_PUBLISHER_THRESHOLD_MS} (72h) the publisher notice,
+ * only if the error is still live. {@link PROLONGED_ERROR_SCAN_BATCH} caps one
+ * cron pass's scan of escalatable errors.
+ */
+export const PROLONGED_ERROR_ESCALATION_VERSION = 1;
+export const PROLONGED_ERROR_OPERATOR_THRESHOLD_MS = 24 * 60 * 60 * 1000;
+export const PROLONGED_ERROR_PUBLISHER_THRESHOLD_MS = 72 * 60 * 60 * 1000;
+export const PROLONGED_ERROR_SCAN_BATCH = 200;

apps/labeler/src/index.ts

@@ -13,6 +13,7 @@ import { didDocumentResponse, policyDocumentResponse } from "./identity.js";
 import { handleNotificationRequest, isNotificationPath } from "./notification-endpoints.js";
 import { runNotificationSweep } from "./notification-sweep.js";
 import { createNotifyDeps, type NotifyDeps } from "./notification-triggers.js";
+import { runProlongedErrorEscalation } from "./prolonged-error.js";
 import { queryLabels } from "./query-labels.js";
 import { reconcileAssessments } from "./reconciliation.js";
 import { createRuntimeSigner, getRuntimeSigningSecret } from "./signing-runtime.js";
@@ -124,6 +125,24 @@ export default {
 				}
 			})(),
 		);
+
+		// Prolonged-error escalation (plan W10.5 follow-up): 24h operator alert then
+		// 72h publisher notice for an errored, unsuperseded run. Its own branch with
+		// the non-throwing deps builder, so a notify-deps failure skips the pass
+		// rather than disturbing the sweep or reconciliation passes above.
+		ctx.waitUntil(
+			(async () => {
+				try {
+					const deps = await safeCreateNotifyDeps(env);
+					if (!deps) return;
+					await runProlongedErrorEscalation(deps, new Date());
+				} catch (err) {
+					console.error("[labeler] prolonged-error escalation failed", {
+						error: err instanceof Error ? err.message : String(err),
+					});
+				}
+			})(),
+		);
 	},
 };
 

apps/labeler/src/notification-triggers.ts

@@ -231,6 +231,26 @@ function reconsiderationOutcomeNoticeContent(
 			};
 }
 
+/** Prolonged-error notice for an assessment stuck in `error` past the 72h
+ * threshold (plan W10.5 follow-up). Neutral and actionable: it states the
+ * labeler could not complete its own assessment, that no label change is
+ * implied, and gives the publisher the one thing they can act on — checking the
+ * release's artifact URL is reachable. Carries NO findings or private detail. */
+function prolongedErrorNoticeContent(
+	urls: NoticeUrls,
+	input: { uri: string; cid: string },
+): NoticeContent {
+	return {
+		subject: "We couldn't complete the security assessment of your plugin release",
+		publicSummary:
+			"The automated security assessment of this release repeatedly failed to complete.",
+		effect:
+			"No label change is implied by this failure. If the release's artifact URL is unavailable or has changed, please verify it is reachable so the assessment can complete.",
+		assessmentUrl: assessmentUrl(urls.serviceUrl, input.uri, input.cid),
+		reconsiderationUrl: urls.reconsiderationUrl,
+	};
+}
+
 // ── Live trigger entry points ───────────────────────────────────────────────
 
 /** Automated block/warning notice from a finalized assessment run. Source is the
@@ -326,6 +346,30 @@ export async function notifyReconsiderationOutcome(
 	);
 }
 
+/**
+ * Prolonged-error publisher notice, fired by the reconciliation cron's 72h stage
+ * (plan W10.5 follow-up) — never at finalization (`assessmentNoticeContent`
+ * stays null for `error`, so `notifyAssessmentOutcome` no-ops on it). Source is
+ * the errored assessment id: an errored run never produces a block/warn notice,
+ * so the `(issuance, id)` key never collides with a finalization notice, and the
+ * claim dedups a crash-retry between the send and the escalation-row mark.
+ */
+export async function notifyProlongedError(
+	deps: NotifyDeps,
+	assessment: Assessment,
+): Promise<boolean> {
+	const target = contactTargetFromUri(assessment.uri);
+	// An unparseable URI is terminal, not transient (the URI never changes), so it
+	// counts as processed — the cron marks it rather than re-attempting forever.
+	if (!target) return true;
+	return runTrigger(
+		deps,
+		{ type: "issuance", id: assessment.id },
+		target,
+		prolongedErrorNoticeContent(deps, { uri: assessment.uri, cid: assessment.cid }),
+	);
+}
+
 /**
  * Rebuild a NOTICE's public content from its source row, for the retry sweep.
  * Nothing about the notice is persisted (plaintext minimization), so a retry
@@ -341,7 +385,10 @@ export async function resolveNoticeForSource(
 ): Promise<NoticeContent | null> {
 	if (sourceType === "issuance") {
 		const assessment = await loadAssessmentSafe(deps.db, sourceId);
-		return assessment ? assessmentNoticeContent(deps, assessment) : null;
+		if (!assessment) return null;
+		if (assessment.state === "error")
+			return prolongedErrorNoticeContent(deps, { uri: assessment.uri, cid: assessment.cid });
+		return assessmentNoticeContent(deps, assessment);
 	}
 	const action = await getOperatorActionById(deps.db, sourceId);
 	if (!action || action.subjectUri === null) return null;
@@ -423,18 +470,26 @@ function operatorActionNeg(metadataJson: string): boolean {
  * Dedup, resolve verification (fail-closed), send, swallow+log. Shared by every
  * trigger so the dedup and verified-skip policy is applied uniformly and a
  * notification failure can never escape into the label path.
+ *
+ * Returns whether the trigger reached a TERMINAL outcome — a dedup hit or a
+ * normal `sendNotification` return (sent, confirmation-sent, undeliverable, or a
+ * claimed-then-failed row the sweep now owns) — versus a thrown TRANSIENT error
+ * (an aggregator read or a pre-claim D1 write that failed before any row was
+ * claimed). The prolonged-error cron uses this to decide whether to stamp its
+ * fire-once mark: a transient failure returns `false` so the next tick retries
+ * instead of being silently swallowed. Fire-and-forget callers ignore it.
  */
 async function runTrigger(
 	deps: NotifyDeps,
 	source: NotificationSource,
 	target: ContactTarget,
 	notice: NoticeContent,
-): Promise<void> {
+): Promise<boolean> {
 	const now = deps.now ?? (() => new Date());
 	try {
 		if (await sourceAlreadyProcessed(deps.db, source)) {
 			logTrigger(source, target.did, "deduped");
-			return;
+			return true;
 		}
 		const verifiedPublisher = await isVerifiedPublisher(deps.aggregator, target.did, now());
 		const ctx: SendContext = {
@@ -449,13 +504,15 @@ async function runTrigger(
 		const request: NotificationRequest = { source, target, notice };
 		const outcome = await sendNotification(ctx, request);
 		logTrigger(source, target.did, outcome.status);
+		return true;
 	} catch (error) {
 		console.error("[notifications] trigger failed", {
 			sourceType: source.type,
 			sourceId: source.id,
 			did: target.did,
 			error: error instanceof Error ? error.message : String(error),
 		});
+		return false;
 	}
 }