fix(insights): validate saved definitions and retire obsolete investigations

.agents/skills/databuddy-internal/SKILL.md

@@ -19,6 +19,7 @@ Keep additions **minimal**: one bullet, a new `rg` hint, or a routing note—eno
 
 ## Quick Map
 
+- Before any PR merge, follow the AGENTS.md review-feedback gate: wait for configured reviewers on the final head, read all comment/review/thread pages, address each finding with evidence, and re-fetch to verify no unresolved feedback. Review bots can finish several minutes after a draft becomes ready; green CI does not establish completed review.
 - Prod infrastructure repo is local at `/Users/iza/Documents/GitHub/databuddy-infra` (`databuddy-analytics/infra`); ClickHouse cluster inventory is `clickhouse/ansible/inventory.yml`, not `/Users/iza/Dev/Databuddy/infra` or `DatabuddyOPS`.
 - Never use production/customer data as tests, fixtures, snapshots, examples, or copied output. Tests must use placeholders/mocks only (example.com, example IDs). If production ClickHouse is queried for investigation, summarize anonymized aggregates and do not paste customer domains, client IDs, emails, or other identifiers into code or responses.
 - `@databuddy/test/env` targets local `databuddy_test` unless `CI=true`, so a normal `db:push` may update a different database; sync that test database explicitly before debugging removed-column failures.

AGENTS.md

@@ -160,6 +160,7 @@ For picker controls, use the component that matches the interaction:
 - **Start fresh**: Check for an existing PR that owns the same surface, public contract, schema, or deployment configuration, then create the branch from an up-to-date `origin/staging`. Do not use an unmerged feature branch as a base unless the dependency is explicit, approved, and named as `Depends on #…` in both PRs.
 - **Make ownership visible**: Push and open a draft PR against `staging` once the slice has a first commit. State its scope, dependencies, and known overlaps.
 - **Keep integration linear**: Rebase a slice onto current `origin/staging` before it is ready for review; do not merge `staging` into the slice merely to refresh it. Request fresh review when a rebase changes reviewed code.
+- **Resolve all review feedback before merging**: Mark the PR ready and wait for configured reviewers to finish on the final head; green CI alone is insufficient. Read every page of general comments, reviews, and inline threads, including outdated threads. Fix actionable findings or document a supported reason for declining them, then resolve each thread. Immediately before merging, re-fetch feedback and verify zero unresolved threads and no unaddressed comments or pending reviews. Never merge immediately after marking a draft ready or pushing review fixes while reviewers are still running.
 - **Isolate parallel work**: Use one worktree per active branch. Never let two agents or contributors mutate the same branch or reuse a task branch for a different concern.
 - **Retire completed work**: Merged PR source branches are automatically deleted. Delete closed PR branches manually, remove clean finished worktrees, and create a new branch from current `staging` for any follow-up—never revive or repurpose an old PR branch.
 

apps/dashboard/app/(main)/organizations/components/measurement-plan-editor.tsx

@@ -71,8 +71,8 @@ export function MeasurementPlanEditor({
 	return (
 		<Card>
 			<Card.Header>
-				<Card.Title>Activation and return</Card.Title>
-				<Card.Description>
+				<Card.Title className="text-balance">Activation and return</Card.Title>
+				<Card.Description className="text-pretty">
 					Choose the events that mean someone got value and came back. Saved
 					definitions guide automatic investigations. Only identified profiles
 					can be measured.
@@ -88,7 +88,7 @@ export function MeasurementPlanEditor({
 							className="flex items-center justify-between gap-2 text-xs"
 							key={item.websiteId}
 						>
-							<p className="text-muted-foreground">
+							<p className="text-pretty text-muted-foreground">
 								{item.name || item.domain}: website unavailable. This definition
 								is inactive.
 							</p>
@@ -120,33 +120,35 @@ export function MeasurementPlanEditor({
 									className="space-y-2 break-words text-xs"
 									key={item.websiteId}
 								>
-									<p className="font-medium">
+									<p className="text-pretty font-medium">
 										{item.name || "Unnamed outcome"}
 									</p>
-									<p className="text-muted-foreground">{item.domain}</p>
+									<p className="text-pretty text-muted-foreground">
+										{item.domain}
+									</p>
 									{site && site.domain !== item.domain && (
-										<p className="text-destructive">
+										<p className="text-pretty text-destructive">
 											Website domain changed to {site.domain}. This definition
 											is inactive until updated.
 										</p>
 									)}
-									<p>
+									<p className="text-pretty">
 										Activation: <code>{item.activationEvent || "Not set"}</code>
 									</p>
-									<p>
+									<p className="text-pretty">
 										Return: <code>{item.returnEvent || "Not set"}</code> within{" "}
 										{item.horizonDays} days
 									</p>
 									{item.namespace && (
-										<p>
+										<p className="text-pretty">
 											Namespace: <code>{item.namespace}</code>
 										</p>
 									)}
 								</div>
 							);
 						})
 					) : (
-						<p className="text-muted-foreground text-xs">
+						<p className="text-pretty text-muted-foreground text-xs">
 							No definitions configured.
 						</p>
 					)
@@ -182,7 +184,7 @@ export function MeasurementPlanEditor({
 									</DropdownMenu.Content>
 								</DropdownMenu>
 							) : (
-								<p className="break-all text-muted-foreground text-xs">
+								<p className="text-pretty break-all text-muted-foreground text-xs">
 									{website.domain}
 								</p>
 							)}
@@ -200,7 +202,7 @@ export function MeasurementPlanEditor({
 							<div className="space-y-4" key={website.id}>
 								{domainMismatch && (
 									<div className="space-y-2" role="alert">
-										<p className="break-words text-destructive text-xs">
+										<p className="text-pretty break-words text-destructive text-xs">
 											This definition is bound to {plan.domain}. Update it to{" "}
 											{website.domain} before saving.
 										</p>
@@ -233,7 +235,7 @@ export function MeasurementPlanEditor({
 												suggestions={events}
 												value={plan[key]}
 											/>
-											<Field.Description>
+											<Field.Description className="text-pretty">
 												{catalog.isError
 													? "Catalog unavailable; enter an exact name."
 													: catalog.isPending
@@ -293,15 +295,15 @@ export function MeasurementPlanEditor({
 								</Accordion>
 							</div>
 						) : (
-							<p className="text-muted-foreground text-xs">
+							<p className="text-pretty text-muted-foreground text-xs">
 								{plans.length >= 20
 									? "Up to 20 website definitions are supported."
 									: "No definition for this website. Add one to choose the outcome and events."}
 							</p>
 						)}
 					</>
 				) : (
-					<p className="text-muted-foreground text-xs">
+					<p className="text-pretty text-muted-foreground text-xs">
 						Add a website to define activation and return.
 					</p>
 				)}

apps/insights/src/generation.ts

@@ -108,6 +108,7 @@ import type { WebsiteInvestigation } from "./persistence";
 import {
 	isInterruptingInvestigation,
 	persistInvestigation,
+	retireObsoleteRetentionObservation,
 } from "./persistence";
 import {
 	captureInsightsError,
@@ -536,7 +537,7 @@ function annotationEvidence(rows: InvestigationAnnotation[]): string | null {
 	return value.length <= 500 ? value : `${value.slice(0, 499).trimEnd()}…`;
 }
 
-async function discoverWebsiteSignals(
+export async function discoverWebsiteSignals(
 	input: InvestigateWebsiteInput,
 	runtime: InvestigationRuntime,
 	options: { allowCoolingFallback?: boolean } = {}
@@ -657,6 +658,17 @@ async function discoverWebsiteSignals(
 			`Insight detection was incomplete (${metricDiagnostics.failedFamilies} metric families and ${definitionDiagnostics.failedDefinitions} conversion definitions failed)`
 		);
 	}
+	const retiredDue =
+		due &&
+		!remeasuredDue &&
+		runtime.mode === "production" &&
+		(await retireObsoleteRetentionObservation({
+			asOf: asOf.toDate(),
+			domain: input.domain,
+			observation: due,
+			organizationId: input.organizationId,
+			websiteId: input.websiteId,
+		}));
 	const signalsByKey = new Map<string, DetectedSignal>();
 	for (const signal of [
 		...(remeasuredDue ? [remeasuredDue] : []),
@@ -670,12 +682,16 @@ async function discoverWebsiteSignals(
 			signalsByKey.set(key, signal);
 		}
 	}
+	if (retiredDue && due) {
+		// A parallel detector may have read the definition before it was edited.
+		signalsByKey.delete(due.signal.signalKey);
+	}
 	const detectedSignals = rankSignals([...signalsByKey.values()]);
 	if (detectedSignals.length === 0) {
 		const coverage = emptyInvestigationCoverage(
-			due ? "due_recheck_unmeasurable" : "no_detected_signals"
+			due && !retiredDue ? "due_recheck_unmeasurable" : "no_detected_signals"
 		);
-		if (due) {
+		if (due && !retiredDue) {
 			if (runtime.mode === "production") {
 				emitInsightsEvent(
 					"info",
@@ -735,7 +751,8 @@ async function discoverWebsiteSignals(
 		: candidateAutomaticEligibleSignals;
 	const hasDetectedCandidate = detectedSignals.some(isInvestigationCandidate);
 	const hasPlannableCandidate = eligibleSignals.length > 0;
-	const hasUnmeasuredDue = due !== null && remeasuredDue === null;
+	const hasUnmeasuredDue =
+		due !== null && remeasuredDue === null && !retiredDue;
 	if (
 		(hasUnmeasuredDue && !hasPlannableCandidate) ||
 		(eligibleSignals.length === 0 && !options.allowCoolingFallback)

apps/insights/src/observations.ts

@@ -53,6 +53,9 @@ export type LatestInsightObservation = Pick<
 
 export interface DueOpenInvestigation extends LatestInsightObservation {
 	evidence: string[];
+	// Synthetic shadow observations have no persisted identity.
+	id?: string;
+	insightId?: string | null;
 }
 
 export function nextRecheckAt(
@@ -158,6 +161,8 @@ export async function loadDueOpenInvestigation(params: {
 }): Promise<DueOpenInvestigation | null> {
 	const rows = await db
 		.selectDistinctOn([insightObservations.signalKey], {
+			id: insightObservations.id,
+			insightId: insightObservations.insightId,
 			evidence: insightObservations.evidence,
 			outcome: insightObservations.outcome,
 			recheckAt: insightObservations.recheckAt,

apps/insights/src/persistence.ts

@@ -1,7 +1,22 @@
 import type { BusinessScope } from "@databuddy/ai/lib/business-context";
 import { assertBusinessScopeCurrent } from "./business-context";
-import { and, db, desc, eq, isNotNull, lte, or, sql } from "@databuddy/db";
-import { analyticsInsights, insightObservations } from "@databuddy/db/schema";
+import {
+	and,
+	db,
+	desc,
+	eq,
+	isNotNull,
+	isNull,
+	lte,
+	or,
+	sql,
+} from "@databuddy/db";
+import {
+	analyticsInsights,
+	insightObservations,
+	organization,
+	websites,
+} from "@databuddy/db/schema";
 import {
 	invalidateAgentContextSnapshotsForWebsite,
 	invalidateInsightsCachesForOrganization,
@@ -10,9 +25,157 @@ import type {
 	InvestigationOutcome,
 	InvestigationSignal,
 } from "@databuddy/shared/insights";
+import { organizationBusinessContextSchema } from "@databuddy/shared/organization-business-context";
 import { randomUUIDv7 } from "bun";
+import { z } from "zod";
 import { normalizedErrorSubject } from "./investigation";
 import { captureInsightsError, emitInsightsEvent } from "./lib/evlog-insights";
+import { measurementPlanKey } from "./measurement-plan";
+import type { DueOpenInvestigation } from "./observations";
+
+export async function retireObsoleteRetentionObservation(params: {
+	asOf: Date;
+	domain: string;
+	observation: DueOpenInvestigation;
+	organizationId: string;
+	websiteId: string;
+}): Promise<boolean> {
+	const { observation } = params;
+	const signalKey = observation.signal.signalKey;
+	const insightId = observation.insightId;
+	if (!(signalKey.startsWith("retention:") && observation.id && insightId)) {
+		return false;
+	}
+	const retired = await db.transaction(async (tx) => {
+		// Match settings-save lock order and hold the canonical definition stable
+		// through the transition. A failed read must roll back, never imply removal.
+		const [owner] = await tx
+			.select({ metadata: organization.metadata })
+			.from(organization)
+			.where(eq(organization.id, params.organizationId))
+			.for("no key update");
+		const [site] = await tx
+			.select({ id: websites.id })
+			.from(websites)
+			.where(
+				and(
+					eq(websites.id, params.websiteId),
+					eq(websites.organizationId, params.organizationId),
+					eq(websites.domain, params.domain),
+					isNull(websites.deletedAt)
+				)
+			)
+			.for("update");
+		if (!(owner?.metadata && site)) {
+			return false;
+		}
+		const { businessContext } = z
+			.object({ businessContext: organizationBusinessContextSchema.optional() })
+			.parse(JSON.parse(owner.metadata));
+		const profile = businessContext?.profile;
+		if (
+			!profile?.measurementPlans ||
+			Date.parse(profile.updatedAt) > params.asOf.getTime() ||
+			profile.measurementPlans.some(
+				(plan) =>
+					plan.websiteId === params.websiteId &&
+					plan.domain === params.domain &&
+					measurementPlanKey(plan) === signalKey
+			)
+		) {
+			return false;
+		}
+		const scope = and(
+			eq(analyticsInsights.id, insightId),
+			eq(analyticsInsights.organizationId, params.organizationId),
+			eq(analyticsInsights.websiteId, params.websiteId),
+			eq(analyticsInsights.subjectKey, signalKey),
+			eq(analyticsInsights.status, "open"),
+			lte(analyticsInsights.createdAt, params.asOf)
+		);
+		const [current] = await tx
+			.select({ id: analyticsInsights.id })
+			.from(analyticsInsights)
+			.where(scope)
+			.for("update");
+		if (!current) {
+			return false;
+		}
+		const [latest] = await tx
+			.select()
+			.from(insightObservations)
+			.where(
+				and(
+					eq(insightObservations.organizationId, params.organizationId),
+					eq(insightObservations.websiteId, params.websiteId),
+					eq(insightObservations.signalKey, signalKey)
+				)
+			)
+			.orderBy(
+				desc(insightObservations.asOf),
+				desc(insightObservations.createdAt)
+			)
+			.limit(1);
+		if (
+			latest?.id !== observation.id ||
+			latest.insightId !== current.id ||
+			latest.asOf > params.asOf ||
+			latest.createdAt > params.asOf ||
+			latest.recheckAt > params.asOf ||
+			latest.outcome.next.type === "resolve"
+		) {
+			return false;
+		}
+		const reason =
+			"The saved activation and return definition was removed or changed. This investigation's measurement no longer applies; recovery was not measured.";
+		await tx
+			.update(analyticsInsights)
+			.set({
+				// Supersede even an in-flight write with this exact snapshot time.
+				// The existing UPDATE/UPSERT fences both compare createdAt with <=.
+				createdAt: new Date(params.asOf.getTime() + 1),
+				status: "resolved",
+				resolvedAt: params.asOf,
+				resolvedReason: "stale",
+			})
+			.where(scope);
+		await tx.insert(insightObservations).values({
+			id: randomUUIDv7(),
+			insightId: current.id,
+			organizationId: params.organizationId,
+			websiteId: params.websiteId,
+			signalKey,
+			signal: latest.signal,
+			evidence: [reason],
+			outcome: {
+				title: latest.outcome.title,
+				summary: reason,
+				evidence: [reason],
+				rootCause: null,
+				impact: null,
+				publish: false,
+				next: { type: "resolve", reason },
+			},
+			asOf: params.asOf,
+			recheckAt: params.asOf,
+		});
+		return true;
+	});
+	if (retired) {
+		try {
+			await Promise.all([
+				invalidateInsightsCachesForOrganization(params.organizationId),
+				invalidateAgentContextSnapshotsForWebsite(params.websiteId),
+			]);
+		} catch (error) {
+			captureInsightsError(error, "generation.cache_invalidation.failed", {
+				organization_id: params.organizationId,
+				website_id: params.websiteId,
+			});
+		}
+	}
+	return retired;
+}
 
 export interface WebsiteInvestigation {
 	id: string;