feat: deliver monthly Featured lineups with search intelligence

docs/clawhub-search-intelligence.md

@@ -6,8 +6,8 @@ Companion to [CLAW-768](https://linear.app/my-openclaw/issue/CLAW-768) under
 ## Boundary and ownership
 
 `POST /api/clawhub-search-intelligence/weekly` accepts ClawHub's frozen
-`search_intelligence_weekly_v2` digest and previously frozen `plugin_search_weekly`
-digests. It uses the existing `CLAWHUB_HERMIT_TOKEN`
+`search_intelligence_weekly_v4` digest and previously frozen v3, v2 and
+`plugin_search_weekly` digests. It uses the existing `CLAWHUB_HERMIT_TOKEN`
 (fallback `CLAWHUB_BAN_APPEALS_TOKEN`) and `CLAWHUB_SITE_URL` trusted-origin
 configuration. Destination is `formSettings.clawhubAppealReviewChannelId`, the
 `maintainer-clawhub` channel. No role or user is mentioned.
@@ -68,12 +68,49 @@ claimed legacy week or cause a second message for it. Deploy receiver support
 before enabling the v2 sender. A read-only report/dry run does not call this POST
 endpoint; invoking it requests delivery.
 
+## Monthly Featured lineups (v4)
+
+V4 carries 16 slots per catalog. Plugins reserve eight editorial slots and use
+eight deduplicated telemetry slots; skills use 16 telemetry slots. ClawHub owns
+eligibility and order: installs across 30 completed UTC days, then installs in the
+final seven days, then stable artifact ID. Search counts, downloads, official
+status and current Featured membership do not improve that order. Editorial
+reservations retain their rationale; pending entries stay pending instead of
+being replaced with telemetry choices. A changed editorial revision is visibly
+stale and requires a regenerated report before approval.
+
+Each catalog carries the exact monthly window, aggregate scan start/end, scanned
+and imported row counts, and import dataset provenance. The scan is not an atomic
+snapshot. Search evidence still has its separate completed-week window and query
+privacy threshold. Missing adoption stays unavailable, not zero. Hermit neither
+re-ranks the report nor publishes Featured selections.
+
+The compact Carbon layout retains every proposed ID, slot, selection basis,
+30d/7d install count and rationale, metadata freshness, plus all pending reservations.
+Weekly search totals, source counts, coverage and bounded company-opportunity,
+official-gap and mover rows remain separate sections. The complete
+wire payload remains capped at 30,000 bytes. Reports that fit use one message;
+longer reports are packed deterministically into bounded parts, without dropping
+selection rows or reasons. Each part uses at most 4000 text characters and 40
+components, with a dashboard link for full search links and removal details.
+
+V4 freezes the complete digest hash and ordered rendered-part hashes at the same
+origin/week key. Each part reuses the existing message receipt, nonce and
+uncertain-send reconciliation owner at a derived part key. Only all confirmed
+parts yield overall success. A rejected part may retry; confirmed earlier parts
+are skipped. An uncertain part must reconcile before later parts can proceed.
+A changed report or rendering manifest conflicts instead of silently replacing a
+partially delivered week. Frozen v1/v2/v3 payloads retain their original validation,
+rendering, hashes and single-message receipts. Deploy receiver support before
+sending v4. Read-only dry runs must not call the delivery endpoint.
+
 ## Delivery state and failure semantics
 
 No migration is needed. The existing D1 `keyValue` primary key stores one receipt
-per trusted origin and UTC week. The record contains a version, canonical digest
-hash, delivery status, start timestamp, and confirmed Discord message ID; it
-contains no query text. Reads use a `first-primary` D1 session. Atomic
+per trusted origin and UTC week, plus v4 part receipts when applicable. Legacy
+message records contain a version, canonical digest
+hash, delivery status, start timestamp, and confirmed Discord message ID; receipts
+contain no query text. Reads use a `first-primary` D1 session. Atomic
 `INSERT ... ON CONFLICT DO NOTHING RETURNING` and compare-and-swap updates fence
 concurrent requests and freeze the weekly payload.
 

scripts/proof-search-intelligence.ts

@@ -122,36 +122,57 @@ try {
 			.bind(key)
 			.first<{ value: string }>()
 		const receipt = row ? JSON.parse(row.value) : null
-		proofStep = "read-confirmed-discord-message"
-		const message = receipt?.messageId
-			? ((await client.rest.get(
-					Routes.channelMessage(channel.id, receipt.messageId)
-				)) as {
-					id: string
-					author: { id: string }
-					flags: number
-					components: unknown[]
-					mentions: unknown[]
-					mention_roles: unknown[]
-					mention_everyone: boolean
-				})
-			: null
+		proofStep = "read-confirmed-discord-messages"
+		// V4 freezes the whole report at the weekly key; message IDs live on
+		// its deterministic part receipts. Older weeks keep their single receipt.
+		const partReceipts =
+			receipt?.version === 2
+				? await Promise.all(
+						receipt.partHashes.map(async (_hash: string, index: number) => {
+							const part = await proxy.env.DB.withSession("first-primary")
+								.prepare("SELECT value FROM keyValue WHERE key = ?")
+								.bind(`${key}:part:${index + 1}`)
+								.first<{ value: string }>()
+							return part ? JSON.parse(part.value) : null
+						})
+					)
+				: [receipt]
+		const confirmed = partReceipts.filter(
+			(part) => part?.status === "sent" && part.messageId
+		)
+		const messages = await Promise.all(
+			confirmed.map(
+				async (part) =>
+					(await client.rest.get(
+						Routes.channelMessage(channel.id, part.messageId)
+					)) as {
+						id: string
+						author: { id: string }
+						flags: number
+						components: unknown[]
+						mentions: unknown[]
+						mention_roles: unknown[]
+						mention_everyone: boolean
+					}
+			)
+		)
+		const discordParts = messages.map((message) => ({
+			url: `https://discord.com/channels/${channel.guild_id}/${channel.id}/${message.id}`,
+			botId: message.author.id,
+			flags: message.flags,
+			components: message.components,
+			mentionCount: message.mentions.length,
+			roleMentionCount: message.mention_roles.length,
+			mentionEveryone: message.mention_everyone
+		}))
+
 		const evidence = {
 			mode: "local Hermit production handler + local persistent D1 + real Discord test bot; NOT deployed production Hermit",
 			first: { status: first?.status, body: await first?.json() },
 			duplicate: { status: replay?.status, body: await replay?.json() },
 			receipt,
-			discord: message
-				? {
-						url: `https://discord.com/channels/${channel.guild_id}/${channel.id}/${message.id}`,
-						botId: message.author.id,
-						flags: message.flags,
-						components: message.components,
-						mentionCount: message.mentions.length,
-						roleMentionCount: message.mention_roles.length,
-						mentionEveryone: message.mention_everyone
-					}
-				: null
+			discord: discordParts[0] ?? null,
+			discordParts
 		}
 		await writeFile(
 			resolve(directory, "evidence.json"),
@@ -163,18 +184,24 @@ try {
 				duplicateStatus: replay?.status,
 				delivered: receipt?.status === "sent",
 				discordUrl: evidence.discord?.url,
+				discordUrls: discordParts.map((part) => part.url),
 				evidencePath: resolve(directory, "evidence.json")
 			})
 		)
 		if (
 			first?.status !== 200 ||
 			replay?.status !== 200 ||
-			!message ||
-			message.flags !== 32768 ||
-			message.author.id !== botId ||
-			message.mentions.length ||
-			message.mention_roles.length ||
-			message.mention_everyone
+			receipt?.status !== "sent" ||
+			messages.length !== partReceipts.length ||
+			!messages.length ||
+			messages.some(
+				(message) =>
+					message.flags !== 32768 ||
+					message.author.id !== botId ||
+					message.mentions.length ||
+					message.mention_roles.length ||
+					message.mention_everyone
+			)
 		)
 			process.exitCode = 1
 	}

src/clawhubSearchIntelligence/delivery.ts

@@ -107,32 +107,111 @@ const findDeliveredMessage = async (
 	return null
 }
 
+type DigestIdentity = {
+	kind?: string
+	weekStart: number
+	weekEnd: number
+	dashboardUrl: string
+}
+type MessageBody = ReturnType<typeof serializePayload>
+const fingerprint = (body: MessageBody, value: string): MessageBody => ({
+	...body,
+	components: [
+		...(body.components ?? []),
+		...(serializePayload({
+			components: [new TextDisplay(`-# Report ${value}`)]
+		}).components ?? [])
+	]
+})
 export const deliverWeeklyDigest = async (
 	client: Client,
-	digest: {
-		kind?: string
-		weekStart: number
-		weekEnd: number
-		dashboardUrl: string
-	},
-	body: ReturnType<typeof serializePayload>
+	digest: DigestIdentity,
+	rendered: MessageBody | MessageBody[]
 ): Promise<Response> => {
-	const db = getRuntimeEnv().DB.withSession("first-primary")
 	const key = `clawhub-search-weekly:${new URL(digest.dashboardUrl).origin}:${digest.weekStart}`
 	const payloadHash = await hash(canonical(digest))
-	if (digest.kind === "search_intelligence_weekly_v3") {
-		// Long candidate URLs may share a dashboard button. The full payload's
-		// fingerprint preserves exact report identity during uncertain-send recovery.
-		body = {
-			...body,
-			components: [
-				...(body.components ?? []),
-				...(serializePayload({
-					components: [new TextDisplay(`-# Report ${payloadHash}`)]
-				}).components ?? [])
-			]
-		}
+	if (!Array.isArray(rendered))
+		return deliverMessage(
+			client,
+			digest,
+			digest.kind === "search_intelligence_weekly_v3"
+				? fingerprint(rendered, payloadHash)
+				: rendered,
+			key,
+			payloadHash
+		)
+	// The existing weekly key freezes the whole report. Each deterministic part
+	// uses the same claim/reconciliation owner; a lost part never resends prior parts.
+	const db = getRuntimeEnv().DB.withSession("first-primary")
+	const startedAt = Date.now()
+	const partHashes = await Promise.all(
+		rendered.map((body) => hash(canonical(body)))
+	)
+	const claim = {
+		version: 2,
+		hash: payloadHash,
+		status: "sending",
+		startedAt,
+		partHashes
 	}
+	const serialized = JSON.stringify(claim)
+	await db
+		.prepare(
+			"INSERT INTO keyValue (key, value, createdAt, updatedAt) VALUES (?, ?, ?, ?) ON CONFLICT(key) DO NOTHING RETURNING key"
+		)
+		.bind(key, serialized, startedAt, startedAt)
+		.first()
+	const existing = await db
+		.prepare("SELECT value FROM keyValue WHERE key = ?")
+		.bind(key)
+		.first<{ value: string }>()
+	if (!existing) return response({ error: "Delivery state unavailable" }, 503)
+	const state = JSON.parse(existing.value) as typeof claim
+	if (
+		state.hash !== payloadHash ||
+		state.version !== 2 ||
+		canonical(state.partHashes) !== canonical(partHashes)
+	)
+		return response({ error: "Weekly payload conflict" }, 409)
+	const success = () =>
+		response({ ok: true, delivered: true, weekEnd: digest.weekEnd })
+	if (state.status === "sent") return success()
+	for (let index = 0; index < rendered.length; index++) {
+		const result = await deliverMessage(
+			client,
+			digest,
+			fingerprint(
+				rendered[index],
+				`${payloadHash} · ${index + 1}/${rendered.length}`
+			),
+			`${key}:part:${index + 1}`,
+			payloadHash
+		)
+		if (!result.ok) return result
+	}
+	const saved = await db
+		.prepare(
+			"UPDATE keyValue SET value = ?, updatedAt = ? WHERE key = ? AND value = ? RETURNING key"
+		)
+		.bind(
+			JSON.stringify({ ...state, status: "sent" }),
+			Date.now(),
+			key,
+			existing.value
+		)
+		.first()
+	return saved
+		? success()
+		: response({ error: "Delivery receipt pending" }, 503)
+}
+const deliverMessage = async (
+	client: Client,
+	digest: DigestIdentity,
+	body: MessageBody,
+	key: string,
+	payloadHash: string
+): Promise<Response> => {
+	const db = getRuntimeEnv().DB.withSession("first-primary")
 	const claim: Delivery = {
 		version: 1,
 		hash: payloadHash,