← back to Homesonspec
Stage-3 verification engine: freshness decay + relabel, source-health auto-pause (SourceRun log + counters), cross-source dedup, verify CLI + 9 unit tests, wired into pipeline runs
3cb66354890e769d016804e1712d9f3606c12044 · 2026-07-22 17:37:25 -0700 · Steve
Files touched
M apps/workers/package.jsonM apps/workers/src/cli.tsM apps/workers/src/pipeline.tsA apps/workers/src/verify-cli.tsA apps/workers/src/verify.test.tsA apps/workers/src/verify.tsA packages/database/prisma/migrations/20260723003309_stage3_verification_engine/migration.sqlM packages/database/prisma/schema.prisma
Diff
commit 3cb66354890e769d016804e1712d9f3606c12044
Author: Steve <steve@designerwallcoverings.com>
Date: Wed Jul 22 17:37:25 2026 -0700
Stage-3 verification engine: freshness decay + relabel, source-health auto-pause (SourceRun log + counters), cross-source dedup, verify CLI + 9 unit tests, wired into pipeline runs
---
apps/workers/package.json | 1 +
apps/workers/src/cli.ts | 12 +-
apps/workers/src/pipeline.ts | 19 +-
apps/workers/src/verify-cli.ts | 81 +++++++
apps/workers/src/verify.test.ts | 68 ++++++
apps/workers/src/verify.ts | 251 +++++++++++++++++++++
.../migration.sql | 31 +++
packages/database/prisma/schema.prisma | 29 +++
8 files changed, 487 insertions(+), 5 deletions(-)
diff --git a/apps/workers/package.json b/apps/workers/package.json
index b361a75..6c8201b 100644
--- a/apps/workers/package.json
+++ b/apps/workers/package.json
@@ -5,6 +5,7 @@
"type": "module",
"scripts": {
"pipeline": "tsx src/cli.ts",
+ "verify": "tsx src/verify-cli.ts",
"start": "tsx src/index.ts",
"test": "vitest run --passWithNoTests",
"typecheck": "tsc --noEmit",
diff --git a/apps/workers/src/cli.ts b/apps/workers/src/cli.ts
index 623debe..8e76607 100644
--- a/apps/workers/src/cli.ts
+++ b/apps/workers/src/cli.ts
@@ -3,6 +3,7 @@ import { meridianHomesAdapter } from "@spechomes/collector-meridian-homes";
import { tollBrothersAdapter } from "@spechomes/collector-toll-brothers";
import { lennarAdapter } from "@spechomes/collector-lennar";
import { runPipeline } from "./pipeline";
+import { recordSourceRun } from "./verify";
/**
* Direct pipeline runner — no queue required.
@@ -23,8 +24,15 @@ async function main() {
process.exit(1);
}
console.log(`Running pipeline for ${key} (extractor v${adapter.version})…`);
- const summary = await runPipeline(adapter);
- console.log(JSON.stringify(summary, null, 2));
+ try {
+ const summary = await runPipeline(adapter);
+ console.log(JSON.stringify(summary, null, 2));
+ } catch (err) {
+ // Record the failure so the failure streak advances toward auto-pause.
+ const message = err instanceof Error ? err.message : String(err);
+ await recordSourceRun(key, { ok: false, kind: "pipeline", message }).catch(() => {});
+ throw err;
+ }
}
main()
diff --git a/apps/workers/src/pipeline.ts b/apps/workers/src/pipeline.ts
index 406ecb0..38cf6d9 100644
--- a/apps/workers/src/pipeline.ts
+++ b/apps/workers/src/pipeline.ts
@@ -17,6 +17,7 @@ import {
} from "@spechomes/validation";
import type { RawPage, SourceAdapter } from "@spechomes/collectors-common";
import { publishStagedRecord } from "@spechomes/publisher";
+import { recordSourceRun } from "./verify";
export { publishStagedRecord };
@@ -233,6 +234,7 @@ export async function runPipeline(adapter: SourceAdapter) {
const source = await prisma.sourceRegistry.findUniqueOrThrow({ where: { key: adapter.key } });
if (!source.active) throw new Error(`source ${source.key} is inactive — refusing to run`);
+ const startedAt = Date.now();
const summary = {
pages: 0,
changed: 0,
@@ -298,9 +300,20 @@ export async function runPipeline(adapter: SourceAdapter) {
}
}
- await prisma.sourceRegistry.update({
- where: { id: source.id },
- data: { health: summary.errors.length > 0 ? "DEGRADED" : "HEALTHY" },
+ // Log the run + roll source health forward (Stage-3). A clean run resets any
+ // failure streak; errors mark the source DEGRADED. Outright throws are recorded
+ // as failures by the caller (verify-cli / pipeline runner), which is what trips
+ // the auto-pause.
+ await recordSourceRun(source.key, {
+ ok: true,
+ kind: "pipeline",
+ pages: summary.pages,
+ changed: summary.changed,
+ published: summary.published,
+ needsReview: summary.needsReview,
+ rejected: summary.rejected,
+ errorCount: summary.errors.length,
+ durationMs: Date.now() - startedAt,
});
return summary;
diff --git a/apps/workers/src/verify-cli.ts b/apps/workers/src/verify-cli.ts
new file mode 100644
index 0000000..50b1236
--- /dev/null
+++ b/apps/workers/src/verify-cli.ts
@@ -0,0 +1,81 @@
+import { prisma } from "@spechomes/database";
+import { recomputeFreshness, dedupePass, refreshDue } from "./verify";
+
+/**
+ * Verification-engine CLI — the Stage-3 maintenance commands.
+ *
+ * pnpm --filter @spechomes/workers verify -- freshness
+ * pnpm --filter @spechomes/workers verify -- dedupe # dry-run
+ * pnpm --filter @spechomes/workers verify -- dedupe --apply # write
+ * pnpm --filter @spechomes/workers verify -- refresh-due
+ * pnpm --filter @spechomes/workers verify -- health
+ * pnpm --filter @spechomes/workers verify -- all # freshness + dedupe(dry) + due + health
+ */
+async function health() {
+ const sources = await prisma.sourceRegistry.findMany({
+ select: {
+ key: true, name: true, health: true, active: true,
+ consecutiveFailures: true, lastRunAt: true, lastSuccessAt: true,
+ },
+ orderBy: { key: "asc" },
+ });
+ return sources.map((s) => ({
+ key: s.key,
+ health: s.health,
+ active: s.active,
+ fails: s.consecutiveFailures,
+ lastRun: s.lastRunAt?.toISOString() ?? null,
+ lastOk: s.lastSuccessAt?.toISOString() ?? null,
+ }));
+}
+
+async function main() {
+ const [cmd, ...rest] = process.argv.slice(2).filter((a) => !a.startsWith("--"));
+ const apply = process.argv.includes("--apply");
+ const now = new Date();
+
+ switch (cmd) {
+ case "freshness": {
+ console.log(JSON.stringify(await recomputeFreshness(now), null, 2));
+ break;
+ }
+ case "dedupe": {
+ console.log(JSON.stringify(await dedupePass({ apply }), null, 2));
+ break;
+ }
+ case "refresh-due": {
+ const due = await refreshDue(now);
+ console.log(JSON.stringify({ due: due.map((d) => d.key), count: due.length }, null, 2));
+ break;
+ }
+ case "health": {
+ console.log(JSON.stringify(await health(), null, 2));
+ break;
+ }
+ case "all":
+ case undefined: {
+ const freshness = await recomputeFreshness(now);
+ const dedupe = await dedupePass({ apply: false });
+ const due = await refreshDue(now);
+ console.log(
+ JSON.stringify(
+ { freshness, dedupe, refreshDue: due.map((d) => d.key), sources: await health() },
+ null,
+ 2,
+ ),
+ );
+ break;
+ }
+ default:
+ console.error(`Unknown command "${cmd}". Try: freshness | dedupe [--apply] | refresh-due | health | all`);
+ process.exitCode = 1;
+ }
+ void rest;
+}
+
+main()
+ .catch((error) => {
+ console.error(error);
+ process.exitCode = 1;
+ })
+ .finally(() => prisma.$disconnect());
diff --git a/apps/workers/src/verify.test.ts b/apps/workers/src/verify.test.ts
new file mode 100644
index 0000000..517879e
--- /dev/null
+++ b/apps/workers/src/verify.test.ts
@@ -0,0 +1,68 @@
+import { describe, expect, it } from "vitest";
+import { computeFreshness, labelRank, pickDedupeWinner, HOUR_MS } from "./verify";
+
+const now = new Date("2026-07-22T12:00:00Z");
+const hoursAgo = (h: number) => new Date(now.getTime() - h * HOUR_MS);
+
+// Standard cadence: fresh <=24h, aging <=72h, review <=168h, stale beyond.
+const F = 24, A = 72, S = 168;
+
+describe("computeFreshness", () => {
+ it("null verification time always needs review", () => {
+ expect(computeFreshness(null, F, A, S, now)).toBe("REVIEW_REQUIRED");
+ });
+ it("recent verification is FRESH", () => {
+ expect(computeFreshness(hoursAgo(1), F, A, S, now)).toBe("FRESH");
+ expect(computeFreshness(hoursAgo(24), F, A, S, now)).toBe("FRESH");
+ });
+ it("between fresh and aging intervals is AGING", () => {
+ expect(computeFreshness(hoursAgo(48), F, A, S, now)).toBe("AGING");
+ });
+ it("between aging and stale intervals needs review", () => {
+ expect(computeFreshness(hoursAgo(100), F, A, S, now)).toBe("REVIEW_REQUIRED");
+ });
+ it("past the stale interval is STALE", () => {
+ expect(computeFreshness(hoursAgo(200), F, A, S, now)).toBe("STALE");
+ });
+});
+
+describe("labelRank", () => {
+ it("orders authority high→low, demo lowest", () => {
+ expect(labelRank("VERIFIED_FROM_BUILDER")).toBeGreaterThan(labelRank("BUILDER_WEBSITE"));
+ expect(labelRank("BUILDER_WEBSITE")).toBeGreaterThan(labelRank("BUILDER_SUBMITTED"));
+ expect(labelRank("DEMONSTRATION")).toBeLessThan(labelRank("NO_LONGER_AVAILABLE"));
+ });
+});
+
+describe("pickDedupeWinner", () => {
+ const mk = (id: string, label: Parameters<typeof labelRank>[0], vAgoH: number | null, cAgoH: number) => ({
+ id,
+ verificationLabel: label,
+ lastVerifiedAt: vAgoH === null ? null : hoursAgo(vAgoH),
+ createdAt: hoursAgo(cAgoH),
+ });
+
+ it("prefers the higher-authority label regardless of recency", () => {
+ const winner = pickDedupeWinner([
+ mk("a", "BUILDER_WEBSITE", 1, 1),
+ mk("b", "VERIFIED_FROM_BUILDER", 100, 100),
+ ]);
+ expect(winner.id).toBe("b");
+ });
+
+ it("breaks a label tie by most-recent verification", () => {
+ const winner = pickDedupeWinner([
+ mk("a", "BUILDER_WEBSITE", 50, 10),
+ mk("b", "BUILDER_WEBSITE", 5, 90),
+ ]);
+ expect(winner.id).toBe("b");
+ });
+
+ it("falls back to newest record when verification times match", () => {
+ const winner = pickDedupeWinner([
+ mk("a", "BUILDER_SUBMITTED", null, 200),
+ mk("b", "BUILDER_SUBMITTED", null, 2),
+ ]);
+ expect(winner.id).toBe("b");
+ });
+});
diff --git a/apps/workers/src/verify.ts b/apps/workers/src/verify.ts
new file mode 100644
index 0000000..a326200
--- /dev/null
+++ b/apps/workers/src/verify.ts
@@ -0,0 +1,251 @@
+import { prisma, type Freshness, type VerificationLabel } from "@spechomes/database";
+
+/**
+ * Stage-3 verification engine — the maintenance layer that runs OVER already
+ * published inventory on a schedule. Ingestion (pipeline.ts) gets a home in the
+ * door; this keeps the marketplace honest afterward:
+ * • freshness decay — verified data ages FRESH → AGING → REVIEW → STALE
+ * • source-health auto-pause — repeated failures trip a source to PAUSED
+ * • cross-source dedup — same address from two sources → keep the best
+ *
+ * The pure decision functions (computeFreshness, labelRank, pickDedupeWinner)
+ * carry no DB and are unit-tested directly; the exported async ops wrap them.
+ */
+
+export const HOUR_MS = 3_600_000;
+
+// Default cadence when a home has no registered source.
+export const DEFAULT_INTERVALS = { freshHrs: 24, agingHrs: 72, staleHrs: 168 };
+
+/** Pure: freshness from age vs a source's thresholds. Null verify time = must review. */
+export function computeFreshness(
+ lastVerifiedAt: Date | null,
+ freshHrs: number,
+ agingHrs: number,
+ staleHrs: number,
+ now: Date,
+): Freshness {
+ if (!lastVerifiedAt) return "REVIEW_REQUIRED";
+ const ageHrs = (now.getTime() - lastVerifiedAt.getTime()) / HOUR_MS;
+ if (ageHrs <= freshHrs) return "FRESH";
+ if (ageHrs <= agingHrs) return "AGING";
+ if (ageHrs <= staleHrs) return "REVIEW_REQUIRED";
+ return "STALE";
+}
+
+// Authority ranking — higher wins a dedup tie. DEMONSTRATION never outranks real data.
+const LABEL_RANK: Record<VerificationLabel, number> = {
+ VERIFIED_FROM_BUILDER: 6,
+ BUILDER_WEBSITE: 5,
+ BUILDER_SUBMITTED: 4,
+ NEEDS_REVERIFICATION: 3,
+ NO_LONGER_AVAILABLE: 2,
+ DEMONSTRATION: 1,
+};
+export const labelRank = (l: VerificationLabel) => LABEL_RANK[l] ?? 0;
+
+export interface DedupeCandidate {
+ id: string;
+ verificationLabel: VerificationLabel;
+ lastVerifiedAt: Date | null;
+ createdAt: Date;
+}
+
+/** Pure: from a set of homes at the same address, choose the one record to keep. */
+export function pickDedupeWinner<T extends DedupeCandidate>(group: T[]): T {
+ return [...group].sort((a, b) => {
+ const r = labelRank(b.verificationLabel) - labelRank(a.verificationLabel);
+ if (r !== 0) return r;
+ const av = a.lastVerifiedAt?.getTime() ?? 0;
+ const bv = b.lastVerifiedAt?.getTime() ?? 0;
+ if (bv !== av) return bv - av;
+ return b.createdAt.getTime() - a.createdAt.getTime();
+ })[0]!;
+}
+
+// ── Freshness recompute ────────────────────────────────────────────────────
+
+// Labels that represent an active positive verification — these get demoted to
+// NEEDS_REVERIFICATION once a home goes stale.
+const REVERIFIABLE = new Set<VerificationLabel>([
+ "VERIFIED_FROM_BUILDER",
+ "BUILDER_WEBSITE",
+ "BUILDER_SUBMITTED",
+]);
+
+export async function recomputeFreshness(now = new Date()) {
+ const homes = await prisma.inventoryHome.findMany({
+ where: { status: "PUBLISHED", isDemo: false },
+ select: {
+ id: true,
+ lastVerifiedAt: true,
+ freshness: true,
+ verificationLabel: true,
+ source: { select: { freshIntervalHours: true, agingIntervalHours: true, staleIntervalHours: true } },
+ },
+ });
+
+ const byTarget: Record<Freshness, string[]> = {
+ FRESH: [], AGING: [], REVIEW_REQUIRED: [], STALE: [], INACTIVE: [],
+ };
+ const relabel: string[] = [];
+
+ for (const h of homes) {
+ const fresh = h.source?.freshIntervalHours ?? DEFAULT_INTERVALS.freshHrs;
+ const aging = h.source?.agingIntervalHours ?? DEFAULT_INTERVALS.agingHrs;
+ const stale = h.source?.staleIntervalHours ?? DEFAULT_INTERVALS.staleHrs;
+ const target = computeFreshness(h.lastVerifiedAt, fresh, aging, stale, now);
+ if (target !== h.freshness) byTarget[target].push(h.id);
+ if (target === "STALE" && REVERIFIABLE.has(h.verificationLabel)) relabel.push(h.id);
+ }
+
+ let changed = 0;
+ for (const [target, ids] of Object.entries(byTarget)) {
+ if (ids.length === 0) continue;
+ await prisma.inventoryHome.updateMany({ where: { id: { in: ids } }, data: { freshness: target as Freshness } });
+ changed += ids.length;
+ }
+ if (relabel.length > 0) {
+ await prisma.inventoryHome.updateMany({
+ where: { id: { in: relabel } },
+ data: { verificationLabel: "NEEDS_REVERIFICATION" },
+ });
+ }
+
+ return {
+ scanned: homes.length,
+ changed,
+ relabeled: relabel.length,
+ toFresh: byTarget.FRESH.length,
+ toAging: byTarget.AGING.length,
+ toReview: byTarget.REVIEW_REQUIRED.length,
+ toStale: byTarget.STALE.length,
+ };
+}
+
+// ── Source-health auto-pause ────────────────────────────────────────────────
+
+export interface RunOutcome {
+ ok: boolean;
+ kind?: string;
+ pages?: number;
+ changed?: number;
+ published?: number;
+ needsReview?: number;
+ rejected?: number;
+ errorCount?: number;
+ message?: string;
+ durationMs?: number;
+}
+
+/**
+ * Log a run and roll the source's health forward. A success resets the failure
+ * streak (and recovers a paused/degraded source); a failure increments it and,
+ * once it reaches maxConsecutiveFailures, trips the source to PAUSED + inactive
+ * so runPipeline (which checks `active`) refuses to run it until a human clears it.
+ */
+export async function recordSourceRun(sourceKey: string, outcome: RunOutcome, now = new Date()) {
+ const source = await prisma.sourceRegistry.findUniqueOrThrow({ where: { key: sourceKey } });
+ await prisma.sourceRun.create({
+ data: {
+ sourceId: source.id,
+ kind: outcome.kind ?? "pipeline",
+ ok: outcome.ok,
+ pages: outcome.pages ?? 0,
+ changed: outcome.changed ?? 0,
+ published: outcome.published ?? 0,
+ needsReview: outcome.needsReview ?? 0,
+ rejected: outcome.rejected ?? 0,
+ errorCount: outcome.errorCount ?? 0,
+ message: outcome.message ?? null,
+ durationMs: outcome.durationMs ?? null,
+ startedAt: now,
+ },
+ });
+
+ const failures = outcome.ok ? 0 : source.consecutiveFailures + 1;
+ const tripped = failures >= source.maxConsecutiveFailures;
+ const health = outcome.ok
+ ? outcome.errorCount && outcome.errorCount > 0
+ ? "DEGRADED"
+ : "HEALTHY"
+ : tripped
+ ? "PAUSED"
+ : "FAILING";
+
+ await prisma.sourceRegistry.update({
+ where: { id: source.id },
+ data: {
+ lastRunAt: now,
+ lastSuccessAt: outcome.ok ? now : source.lastSuccessAt,
+ consecutiveFailures: failures,
+ health,
+ // Auto-pause disables the source; a successful run re-enables it.
+ active: outcome.ok ? true : tripped ? false : source.active,
+ },
+ });
+
+ return { key: sourceKey, health, consecutiveFailures: failures, autoPaused: !outcome.ok && tripped };
+}
+
+// ── Refresh scheduling ──────────────────────────────────────────────────────
+
+/** Sources due for a refresh: active, not paused, and past their fresh interval. */
+export async function refreshDue(now = new Date()) {
+ const sources = await prisma.sourceRegistry.findMany({
+ where: { active: true, health: { not: "PAUSED" } },
+ select: { key: true, name: true, lastRunAt: true, freshIntervalHours: true },
+ });
+ return sources.filter((s) => {
+ if (!s.lastRunAt) return true;
+ const ageHrs = (now.getTime() - s.lastRunAt.getTime()) / HOUR_MS;
+ return ageHrs >= s.freshIntervalHours;
+ });
+}
+
+// ── Cross-source dedup ──────────────────────────────────────────────────────
+
+/**
+ * Find published non-demo homes that share a normalized address + zip (the same
+ * physical home surfaced by two sources / crawls). Keep the most authoritative
+ * record; deactivate the rest. Dry-run by default — pass { apply: true } to write.
+ */
+export async function dedupePass(opts: { apply?: boolean } = {}) {
+ const homes = await prisma.inventoryHome.findMany({
+ where: { status: "PUBLISHED", isDemo: false, normalizedAddress: { not: null } },
+ select: {
+ id: true,
+ normalizedAddress: true,
+ zip: true,
+ verificationLabel: true,
+ lastVerifiedAt: true,
+ createdAt: true,
+ },
+ });
+
+ const groups = new Map<string, typeof homes>();
+ for (const h of homes) {
+ const key = `${h.normalizedAddress}|${h.zip}`;
+ const arr = groups.get(key) ?? [];
+ arr.push(h);
+ groups.set(key, arr);
+ }
+
+ const losers: string[] = [];
+ let dupGroups = 0;
+ for (const group of groups.values()) {
+ if (group.length < 2) continue;
+ dupGroups += 1;
+ const winner = pickDedupeWinner(group);
+ for (const h of group) if (h.id !== winner.id) losers.push(h.id);
+ }
+
+ if (opts.apply && losers.length > 0) {
+ await prisma.inventoryHome.updateMany({
+ where: { id: { in: losers } },
+ data: { status: "INACTIVE", freshness: "INACTIVE" },
+ });
+ }
+
+ return { scanned: homes.length, duplicateGroups: dupGroups, duplicates: losers.length, applied: !!opts.apply };
+}
diff --git a/packages/database/prisma/migrations/20260723003309_stage3_verification_engine/migration.sql b/packages/database/prisma/migrations/20260723003309_stage3_verification_engine/migration.sql
new file mode 100644
index 0000000..2a87e2d
--- /dev/null
+++ b/packages/database/prisma/migrations/20260723003309_stage3_verification_engine/migration.sql
@@ -0,0 +1,31 @@
+-- AlterTable
+ALTER TABLE "SourceRegistry" ADD COLUMN "consecutiveFailures" INTEGER NOT NULL DEFAULT 0,
+ADD COLUMN "lastRunAt" TIMESTAMP(3),
+ADD COLUMN "lastSuccessAt" TIMESTAMP(3),
+ADD COLUMN "maxConsecutiveFailures" INTEGER NOT NULL DEFAULT 3,
+ADD COLUMN "staleIntervalHours" INTEGER NOT NULL DEFAULT 168;
+
+-- CreateTable
+CREATE TABLE "SourceRun" (
+ "id" TEXT NOT NULL,
+ "sourceId" TEXT NOT NULL,
+ "kind" TEXT NOT NULL DEFAULT 'pipeline',
+ "ok" BOOLEAN NOT NULL,
+ "pages" INTEGER NOT NULL DEFAULT 0,
+ "changed" INTEGER NOT NULL DEFAULT 0,
+ "published" INTEGER NOT NULL DEFAULT 0,
+ "needsReview" INTEGER NOT NULL DEFAULT 0,
+ "rejected" INTEGER NOT NULL DEFAULT 0,
+ "errorCount" INTEGER NOT NULL DEFAULT 0,
+ "message" TEXT,
+ "durationMs" INTEGER,
+ "startedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+ CONSTRAINT "SourceRun_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE INDEX "SourceRun_sourceId_startedAt_idx" ON "SourceRun"("sourceId", "startedAt");
+
+-- AddForeignKey
+ALTER TABLE "SourceRun" ADD CONSTRAINT "SourceRun_sourceId_fkey" FOREIGN KEY ("sourceId") REFERENCES "SourceRegistry"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma
index 0e8f506..411fd5d 100644
--- a/packages/database/prisma/schema.prisma
+++ b/packages/database/prisma/schema.prisma
@@ -362,8 +362,14 @@ model SourceRegistry {
rateLimitRpm Int?
freshIntervalHours Int @default(24)
agingIntervalHours Int @default(72)
+ staleIntervalHours Int @default(168) // beyond this a home is STALE + needs re-verification
health SourceHealth @default(HEALTHY)
active Boolean @default(true)
+ // Verification-engine run tracking (Stage 3): drives source-health auto-pause.
+ lastRunAt DateTime?
+ lastSuccessAt DateTime?
+ consecutiveFailures Int @default(0)
+ maxConsecutiveFailures Int @default(3) // trip to PAUSED once reached
notes String?
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
@@ -372,6 +378,29 @@ model SourceRegistry {
snapshots RawSnapshot[]
staged StagedRecord[]
homes InventoryHome[]
+ runs SourceRun[]
+}
+
+// Append-only log of every verification-engine / pipeline run per source.
+// The health evaluator reads the recent tail to decide auto-pause / recovery.
+model SourceRun {
+ id String @id @default(cuid())
+ sourceId String
+ kind String @default("pipeline") // "pipeline" | "freshness" | "refresh"
+ ok Boolean
+ pages Int @default(0)
+ changed Int @default(0)
+ published Int @default(0)
+ needsReview Int @default(0)
+ rejected Int @default(0)
+ errorCount Int @default(0)
+ message String?
+ durationMs Int?
+ startedAt DateTime @default(now())
+
+ source SourceRegistry @relation(fields: [sourceId], references: [id])
+
+ @@index([sourceId, startedAt])
}
model RawSnapshot {
← 832a5c0 Design system iter-2: reskin /map explorer panel + /search c
·
back to Homesonspec
·
Contrarian gate fixes: guard auto-pause recovery (manual dis ec6046b →