@kentcdodds/kody-issue-triage

import postMessage from 'kody:@kentcdodds/discord/post-message'
import editMessage from 'kody:@kentcdodds/discord/edit-message'
import { kody } from 'kody:runtime'
import { gatherPeerContext } from './peer-context.ts'
import {
	PATCH_BOT_ID,
	PATCH_BOT_NAME,
	buildPatchTriagePrompt,
	wakePatchForTriage,
} from './patch-handoff.ts'
import { gatherLoopGuardSnapshot } from './spawn-snapshot.ts'
import { emptyLoopGuardSnapshot, type LoopGuardSnapshot } from './loop-guard.ts'
import {
	emptyPeerContext,
	handledSiblingForRecord,
	inFlightSiblingForRecord,
	type PeerContext,
} from './peer-match.ts'
import {
	SELF_KODY_ID,
	SKIPPED_EXECUTE_STATUS,
	discordChannelId,
	isFleetRateFingerprint,
	eventAnomalyThreshold,
	fingerprintAnomalyThreshold,
	hourBucket,
	maxAgentsPerHour,
	retriageCooldownMs,
	AGENT_EXECUTE_NOISE_NOTE,
	isSelfPackageRun,
	shouldIgnoreAgentExecuteNoise,
	shouldIgnorePlatformWeather,
	skipReason,
	spawnSkipReasonForRecord,
	truncate,
	PLATFORM_WEATHER_NOTE,
	type FingerprintRecord,
	type RunSnippet,
} from './shared.ts'
import {
	getCounter,
	incrementCounter,
	initSchema,
	isEnabled,
	listFingerprintsByStatuses,
	listStaleSpawned,
	releaseLease,
	tryAcquireLease,
	updateFingerprint,
	upsertFingerprintFromRun,
} from './storage.ts'
import {
	applyTriageToMatchingOpenRuns,
	softTriageForOutcome,
} from './apply-matching-runs.ts'
import { formatKodyIssueReport } from './format-discord-report.ts'


async function withTimeout<T>(promise: Promise<T>, ms: number, fallback: T): Promise<T> {
	let timer: ReturnType<typeof setTimeout> | undefined
	try {
		return await Promise.race([
			promise,
			new Promise<T>((resolve) => {
				timer = setTimeout(() => resolve(fallback), ms)
			}),
		])
	} finally {
		if (timer) clearTimeout(timer)
	}
}

export type SweepInput = {
	dryRun?: boolean
	/** Process at most this many queued fingerprints (default 1). */
	limit?: number
	/** When true, also scan current open run errors into the queue. */
	reconcile?: boolean
}

function discordId(posted: unknown) {
	const value = posted as { id?: string; messageId?: string; message_id?: string }
	return value?.id || value?.messageId || value?.message_id || null
}


async function postOrEditQueuedCard(
	record: FingerprintRecord,
	content: string,
	flags: unknown,
) {
	if (record.discord_message_id) {
		try {
			await editMessage({
				channelId: discordChannelId,
				messageId: record.discord_message_id,
				content,
				flags,
			})
			return record.discord_message_id
		} catch {
			// Message may have been deleted; post a replacement.
		}
	}
	const posted = await postMessage({
		channelId: discordChannelId,
		content,
		flags,
	})
	return discordId(posted)
}

/** Edit an existing Queued card off pure "Waking Patch…" after a failed/interrupted spawn. */
async function editCardRetryable(
	record: FingerprintRecord,
	discordMessageId: string,
	reason: string,
) {
	const fleetRate = isFleetRateFingerprint(record.fingerprint)
	const report = formatKodyIssueReport({
		status: 'queued',
		title: fleetRate ? 'Fleet package error rate elevated' : undefined,
		record,
		summary: record.sample_message
			? truncate(record.sample_message, 240)
			: null,
		extras: [reason],
	})
	try {
		await editMessage({
			channelId: discordChannelId,
			messageId: discordMessageId,
			content: report.content,
			flags: report.flags,
		})
	} catch {
		// Best-effort; stored discord_message_id still lets the next sweep edit in place.
	}
}


function snippetFromListedRun(run: {
	id?: string
	surface?: string | null
	name?: string | null
	package_id?: string | null
	kody_id?: string | null
	job_id?: string | null
	error_name?: string | null
	error_message?: string | null
	started_at?: string | null
	metadata?: Record<string, unknown> | null
}): RunSnippet {
	return {
		id: run.id,
		surface: run.surface,
		name: run.name,
		package_id: run.package_id,
		kody_id: run.kody_id,
		job_id: run.job_id,
		error_name: run.error_name,
		error_message: run.error_message,
		started_at: run.started_at,
		metadata: run.metadata ?? null,
	}
}

const SELF_PACKAGE_NOTE =
	'Ignored: kody-issue-triage self-package run (UNIQUE SQLITE / KodyFetchGateway / own handler). Not a standing product failure.'

async function ignoreOpenRuns(runIds: string[], note: string, dryRun: boolean) {
	if (!runIds.length) return { ignored: 0, dryRun }
	if (dryRun) return { ignored: runIds.length, dryRun: true }
	let ignored = 0
	for (let i = 0; i < runIds.length; i += 100) {
		try {
			const result = await kody.runUpdateBulk({
				run_ids: runIds.slice(i, i + 100),
				triage: 'ignored',
				note,
			})
			ignored += result.updated_count || 0
		} catch {
			// Best-effort Activity drain.
		}
	}
	return { ignored, dryRun: false }
}

async function drainRecordedOutcomes(dryRun: boolean) {
	const records = await listFingerprintsByStatuses(
		['fixed', 'ignored', 'resolved_noise', 'loop_detected'],
		40,
	)
	let updated = 0
	for (const record of records) {
		const soft = softTriageForOutcome(record.outcome || record.status)
		if (!soft) continue
		if (dryRun) {
			updated += record.sample_run_ids.length
			continue
		}
		const applied = await applyTriageToMatchingOpenRuns(
			record,
			soft,
			`kody-issue-triage sweep drain ${record.outcome || record.status}`,
		)
		updated += applied.updated
	}
	return { fingerprints: records.length, updated, dryRun }
}

async function reconcileOpenErrors(dryRun: boolean) {
	const page = await kody.runList({
		status: 'error',
		error_triage: 'open',
		limit: 100,
	})
	const knownRuns = new Map<string, RunSnippet>()
	const executeIgnoreIds: string[] = []
	const selfIgnoreIds: string[] = []
	const weatherIgnoreIds: string[] = []
	const standingIgnoreIds: string[] = []
	let created = 0
	let queued = 0
	let skippedExecute = 0
	let skippedSelf = 0
	let skippedWeather = 0
	for (const run of page.runs || []) {
		const snippet = snippetFromListedRun(run)
		if (snippet.id) knownRuns.set(snippet.id, snippet)
		const skip = skipReason(snippet)
		if (skip === 'agent-execute') {
			skippedExecute++
			if (snippet.id && shouldIgnoreAgentExecuteNoise(snippet)) {
				executeIgnoreIds.push(snippet.id)
			}
			continue
		}
		if (skip === 'self-package' || isSelfPackageRun(snippet)) {
			skippedSelf++
			if (snippet.id) selfIgnoreIds.push(snippet.id)
			continue
		}
		if (skip) continue
		const result = await upsertFingerprintFromRun(snippet)
		const weather = shouldIgnorePlatformWeather(snippet)
		const standing =
			result.nextStatus === 'ignored' || result.nextStatus === 'resolved_noise'
		if (weather || standing) {
			skippedWeather++
			if (weather && !dryRun) {
				await updateFingerprint(result.fingerprint, {
					status: 'resolved_noise',
					outcome: 'resolved_noise',
					classification: 'noise',
					completed_at: new Date().toISOString(),
					summary: PLATFORM_WEATHER_NOTE,
				})
			}
			if (snippet.id) {
				if (weather) weatherIgnoreIds.push(snippet.id)
				else standingIgnoreIds.push(snippet.id)
			}
			continue
		}
		queued++
		if (result.created) created++
	}
	const ignoredExecute = await ignoreOpenRuns(
		executeIgnoreIds,
		AGENT_EXECUTE_NOISE_NOTE,
		dryRun,
	)
	const ignoredSelf = await ignoreOpenRuns(selfIgnoreIds, SELF_PACKAGE_NOTE, dryRun)
	const ignoredWeather = await ignoreOpenRuns(
		weatherIgnoreIds,
		PLATFORM_WEATHER_NOTE,
		dryRun,
	)
	const ignoredStanding = await ignoreOpenRuns(
		standingIgnoreIds,
		'Standing ignored/resolved_noise fingerprint; do not refill Activity.',
		dryRun,
	)
	return {
		scanned: (page.runs || []).length,
		queued,
		created,
		skippedExecute,
		skippedSelf,
		skippedWeather,
		ignoredExecuteNoise: ignoredExecute.ignored,
		ignoredSelfPackage: ignoredSelf.ignored,
		ignoredExecuteTimeouts: ignoredExecute.ignored,
		ignoredPlatformWeather: ignoredWeather.ignored,
		ignoredStandingNoise: ignoredStanding.ignored,
		knownRuns,
	}
}

async function sampleRunForRecord(
	record: FingerprintRecord,
	knownRuns: Map<string, RunSnippet>,
): Promise<RunSnippet | null> {
	const sampleId = record.sample_run_id
	if (!sampleId) return null
	const known = knownRuns.get(sampleId)
	if (known) return known
	try {
		const got = await kody.runGet({ run_id: sampleId })
		const run = got.run
		if (!run) return null
		return snippetFromListedRun(run)
	} catch {
		return null
	}
}

async function closeResolvedNoiseFingerprint(
	record: FingerprintRecord,
	summary: string,
	dryRun: boolean,
) {
	if (!dryRun) {
		await updateFingerprint(record.fingerprint, {
			status: 'resolved_noise',
			outcome: 'resolved_noise',
			classification: 'noise',
			completed_at: new Date().toISOString(),
			summary,
		})
	}
	return {
		ok: true,
		skipped: 'platform-weather' as const,
		fingerprint: record.fingerprint,
		dryRun,
	}
}

async function closeAgentExecuteFingerprint(
	record: FingerprintRecord,
	dryRun: boolean,
) {
	const summary =
		'Skipped spawn: sample run originated from an agent execute call; Activity soft-ignored.'
	if (!dryRun) {
		await updateFingerprint(record.fingerprint, {
			status: SKIPPED_EXECUTE_STATUS,
			summary,
		})
		await applyTriageToMatchingOpenRuns(
			record,
			'ignored',
			AGENT_EXECUTE_NOISE_NOTE,
		)
		const sampleIds = [
			...(record.sample_run_id ? [record.sample_run_id] : []),
			...(record.sample_run_ids || []),
		]
		await ignoreOpenRuns([...new Set(sampleIds)], AGENT_EXECUTE_NOISE_NOTE, false)
	}
	return {
		ok: true,
		skipped: 'agent-execute' as const,
		fingerprint: record.fingerprint,
		dryRun,
	}
}

function canRetriage(record: FingerprintRecord) {
	if (record.status === 'queued' || record.status === 'claimed') return true
	if (record.status === 'spawned') return false
	if (!record.completed_at) return true
	return Date.now() - Date.parse(record.completed_at) >= retriageCooldownMs
}

async function maybeAlert(kind: string, content: string, dryRun: boolean) {
	const key = `alerted:${kind}:${hourBucket()}`
	const already = await getCounter(key)
	if (already > 0) return { sent: false }
	if (dryRun) return { sent: false, dryRun: true }
	await incrementCounter(key)
	const alert = formatKodyIssueReport({
		status: 'queued',
		title: kind === 'cap' ? 'Hourly spawn cap reached' : 'Anomaly breaker paused spawning',
		record: {
			fingerprint: kind,
			surface: 'sweep',
			owner: 'kody-issue-triage',
			count: 0,
			activity_url: null,
			agent_url: null,
			kody_agent_url: null,
			classification: null,
			sample_message: content,
			error_family: kind,
		},
		summary: content,
	})
	await postMessage({
		channelId: discordChannelId,
		content: alert.content,
		flags: alert.flags,
	})
	return { sent: true }
}

async function spawnOne(record: FingerprintRecord, dryRun: boolean) {
	if (!dryRun) {
		const lease = await tryAcquireLease(record.fingerprint, SELF_KODY_ID)
		if (!lease.ok) {
			return { ok: true, queued: true, skipped: 'lease-held', heldBy: lease.heldBy }
		}
	}

	const bucket = hourBucket()
	const spawned = await getCounter(`spawned:${bucket}`)
	if (spawned >= maxAgentsPerHour) {
		await releaseLease(record.fingerprint)
		await maybeAlert(
			'cap',
			`Kody issue triage hourly cap (${maxAgentsPerHour}) reached. Fingerprint \`${record.fingerprint}\` stays queued.`,
			dryRun,
		)
		return { ok: true, skipped: 'hourly-cap', spawned }
	}

	let peerContext: PeerContext = emptyPeerContext()
	try {
		peerContext = await withTimeout(gatherPeerContext(record), 8_000, emptyPeerContext())
	} catch {
		peerContext = emptyPeerContext()
	}
	if (peerContext.highConfidenceLiveSibling) {
		await releaseLease(record.fingerprint)
		return {
			ok: true,
			queued: true,
			skipped: 'live-peer',
			fingerprint: record.fingerprint,
			peerAgentId: peerContext.highConfidenceLiveSibling.id,
			peerAgentUrl: peerContext.highConfidenceLiveSibling.url,
		}
	}

	const inFlightSibling = inFlightSiblingForRecord(peerContext.siblings, record)
	if (inFlightSibling) {
		await releaseLease(record.fingerprint)
		return {
			ok: true,
			queued: true,
			skipped: 'in-flight-sibling',
			fingerprint: record.fingerprint,
			sibling: inFlightSibling.fingerprint,
		}
	}

	const handledSibling = handledSiblingForRecord(peerContext.siblings, record)
	if (handledSibling) {
		const summary = `Same-owner sibling ${handledSibling.fingerprint} already ${handledSibling.outcome || handledSibling.status}; skipped spawn.`
		if (!dryRun) {
			await updateFingerprint(record.fingerprint, {
				status: 'resolved_noise',
				outcome: 'resolved_noise',
				classification: 'noise',
				completed_at: new Date().toISOString(),
				summary,
			})
		}
		await releaseLease(record.fingerprint)
		return {
			ok: true,
			skipped: 'handled-peer',
			fingerprint: record.fingerprint,
			sibling: handledSibling.fingerprint,
			siblingOutcome: handledSibling.outcome,
			dryRun,
		}
	}

	let loopGuard: LoopGuardSnapshot = emptyLoopGuardSnapshot()
	try {
		loopGuard = await withTimeout(
			gatherLoopGuardSnapshot(record),
			8_000,
			emptyLoopGuardSnapshot(),
		)
	} catch {
		loopGuard = emptyLoopGuardSnapshot()
	}

	const fleetRate = isFleetRateFingerprint(record.fingerprint)
	const report = formatKodyIssueReport({
		status: 'queued',
		title: fleetRate
			? 'Fleet package error rate elevated'
			: undefined,
		record,
		summary: record.sample_message
			? truncate(record.sample_message, 240)
			: null,
		extras: ['Waking Patch…'],
	})

	if (dryRun) {
		await releaseLease(record.fingerprint)
		return {
			ok: true,
			dryRun: true,
			fingerprint: record.fingerprint,
			wouldSpawn: true,
			promptChars: buildPatchTriagePrompt({
				record,
				discordMessageId: 'dry-run',
				peerContext,
				loopGuard,
				fleetRate,
			}).length,
			wouldWake: 'patch',
			peerAgents: peerContext.agents.length,
			peerPullRequests: peerContext.pullRequests.length,
			matchingOpenRuns: loopGuard.matchingOpenRunCount,
		}
	}

	// Post/edit Discord first, then persist discord_message_id BEFORE wake.
	// A DO reset between Discord write and storage was leaving orphan Queued
	// cards (no stored id → next hour posted a second card).
	let discordMessageId: string | null = null
	try {
		const postedPromise = postOrEditQueuedCard(
			record,
			report.content,
			report.flags,
		)
		discordMessageId = await withTimeout(postedPromise, 12_000, null)
		if (!discordMessageId) {
			// Timeout must not abandon an in-flight Discord write — await it so
			// a late success is still persisted (never orphan with no stored id).
			discordMessageId = await withTimeout(postedPromise, 20_000, null)
		}
		if (!discordMessageId) {
			await releaseLease(record.fingerprint)
			return {
				ok: false,
				fingerprint: record.fingerprint,
				error: 'discord-post-timeout',
			}
		}
		await updateFingerprint(record.fingerprint, {
			status: 'claimed',
			discord_message_id: discordMessageId,
		})
	} catch (error) {
		if (discordMessageId) {
			try {
				await updateFingerprint(record.fingerprint, {
					status: 'queued',
					discord_message_id: discordMessageId,
				})
			} catch {
				// Persist best-effort; Discord edit below still marks retryable.
			}
			await editCardRetryable(
				record,
				discordMessageId,
				'Spawn interrupted after Discord post — will retry next sweep.',
			)
		}
		await releaseLease(record.fingerprint)
		throw error
	}

	try {
		const prompt = buildPatchTriagePrompt({
			record,
			discordMessageId: discordMessageId || 'unknown',
			openSiblingCount: record.count,
			peerContext,
			loopGuard,
			fleetRate,
		})
		const { ok, timedOut, woke } = await wakePatchForTriage({
			prompt,
			discordChannelId,
			discordMessageId,
		})
		if (!ok) {
			// Keep discord_message_id so the next sweep edits in place.
			await updateFingerprint(record.fingerprint, { status: 'queued' })
			await editCardRetryable(
				record,
				discordMessageId!,
				'Patch wake failed — will retry next sweep.',
			)
			await releaseLease(record.fingerprint)
			return {
				ok: false,
				fingerprint: record.fingerprint,
				error: 'patch-wake-failed',
				wake: woke,
				discordMessageId,
			}
		}
		await incrementCounter(`spawned:${bucket}`)
		const agentUrl = `grok-bot:patch`
		await updateFingerprint(record.fingerprint, {
			status: 'spawned',
			agent_id: PATCH_BOT_ID,
			agent_url: agentUrl,
			spawned_at: new Date().toISOString(),
			discord_message_id: discordMessageId,
		})
		return {
			ok: true,
			fingerprint: record.fingerprint,
			agentId: PATCH_BOT_ID,
			agentUrl,
			wokePatch: true,
			timedOut,
			bot: PATCH_BOT_NAME,
			discordMessageId,
		}
	} catch (error) {
		// Keep discord_message_id; requeue for edit-in-place retry.
		await updateFingerprint(record.fingerprint, { status: 'queued' })
		await editCardRetryable(
			record,
			discordMessageId!,
			'Spawn interrupted — will retry next sweep.',
		)
		await releaseLease(record.fingerprint)
		throw error
	}
}

/**
 * Bounded spawn path. Default: wake Patch for at most one fingerprint. Safe to invoke manually.
 */
export default async function sweep(input: SweepInput = {}) {
	const dryRun = input.dryRun === true
	const limit = Math.min(Math.max(input.limit ?? 1, 1), 3)
	const reconcile = input.reconcile !== false

	await initSchema()
	if (!(await isEnabled())) {
		return { ok: true, skipped: 'paused' }
	}

	const recovered = []
	if (!dryRun) {
		for (const stale of await listStaleSpawned()) {
			const summary =
				'Sweeper marked failed: Patch never recorded an outcome within the stale-spawn window.'
			await updateFingerprint(stale.fingerprint, {
				status: 'failed',
				outcome: 'failed',
				completed_at: new Date().toISOString(),
				summary,
			})
			await releaseLease(stale.fingerprint)
			recovered.push(stale.fingerprint)
		}
	}

	// Wake Patch first. Reconcile/drain can burn the whole execute budget
	// soft-triaging Activity rows, which previously timed out before spawn.
	const bucket = hourBucket()
	const fingerprintsThisHour = await getCounter(`fingerprints:${bucket}`)
	const eventsThisHour = await getCounter(`events:${bucket}`)
	if (
		fingerprintsThisHour > fingerprintAnomalyThreshold ||
		eventsThisHour > eventAnomalyThreshold
	) {
		await maybeAlert(
			'anomaly',
			`Kody issue triage paused this hour (anomaly): ${fingerprintsThisHour} new fingerprints, ${eventsThisHour} events. Caps are ${fingerprintAnomalyThreshold} / ${eventAnomalyThreshold}.`,
			dryRun,
		)
		return {
			ok: true,
			skipped: 'anomaly-breaker',
			fingerprintsThisHour,
			eventsThisHour,
			recovered,
		}
	}

	// Include claimed: DO reset after persist-but-before-wake left fingerprints
	// claimed with a stored discord_message_id that queued-only sweeps skipped,
	// while a missing id would post a duplicate Queued card. Edit-existing-first
	// reuses the card when the id is present.
	const queued = (
		await listFingerprintsByStatuses(['queued', 'claimed'], 20)
	).sort((a, b) => String(a.first_seen).localeCompare(String(b.first_seen)))
	const actions: Array<Record<string, unknown>> = []
	let spawnedCount = 0
	for (const record of queued.filter(canRetriage)) {
		if (spawnedCount >= limit) break
		const sampleRun = await sampleRunForRecord(record, new Map())
		const executeSkip = spawnSkipReasonForRecord(record, sampleRun)
		if (executeSkip) {
			actions.push(await closeAgentExecuteFingerprint(record, dryRun))
			continue
		}
		if (sampleRun && shouldIgnorePlatformWeather(sampleRun)) {
			actions.push(
				await closeResolvedNoiseFingerprint(record, PLATFORM_WEATHER_NOTE, dryRun),
			)
			continue
		}
		try {
			const action = await spawnOne(record, dryRun)
			actions.push(action)
			if (action.skipped === 'hourly-cap') break
			if (action.agentId || action.wouldSpawn) spawnedCount++
		} catch (error) {
			actions.push({
				ok: false,
				fingerprint: record.fingerprint,
				error: String((error as Error)?.message || error),
			})
		}
	}

	let reconcileResult: {
		scanned?: number
		queued?: number
		created?: number
		skippedExecute?: number
		ignoredExecuteTimeouts?: number
		error?: string
		timedOut?: boolean
	} | null = null
	if (reconcile) {
		try {
			const reconciled = await withTimeout(
				reconcileOpenErrors(dryRun),
				20_000,
				{
					scanned: 0,
					queued: 0,
					created: 0,
					skippedExecute: 0,
					skippedSelf: 0,
					skippedWeather: 0,
					ignoredExecuteNoise: 0,
					ignoredSelfPackage: 0,
					ignoredExecuteTimeouts: 0,
					ignoredPlatformWeather: 0,
					ignoredStandingNoise: 0,
					knownRuns: new Map(),
				},
			)
			reconcileResult = {
				scanned: reconciled.scanned,
				queued: reconciled.queued,
				created: reconciled.created,
				skippedExecute: reconciled.skippedExecute,
				ignoredExecuteTimeouts: reconciled.ignoredExecuteTimeouts,
				timedOut: reconciled.scanned === 0 && reconciled.queued === 0,
			}
		} catch (error) {
			reconcileResult = { error: String((error as Error)?.message || error) }
		}
	}

	let drainResult: {
		fingerprints?: number
		updated?: number
		error?: string
		timedOut?: boolean
	} | null = null
	try {
		drainResult = await withTimeout(
			drainRecordedOutcomes(dryRun),
			20_000,
			{ fingerprints: 0, updated: 0, dryRun, timedOut: true },
		)
	} catch (error) {
		drainResult = { error: String((error as Error)?.message || error) }
	}

	return {
		ok: true,
		dryRun,
		spawned: spawnedCount,
		queued: queued.length,
		actions,
		reconcile: reconcileResult,
		drain: drainResult,
		recovered,
		fingerprintsThisHour,
		eventsThisHour,
	}
}