Wire Mastra populate through self-healing

Author: giaphutran12

.env.example

@@ -14,6 +14,10 @@ OPENROUTER_API_KEY=sk-or-...
 # Used by the backend container to call internal Convex functions.
 CONVEX_SELF_HOSTED_ADMIN_KEY=
 
+# Durable store for self-healing populate recipe manifests.
+# Docker dev overrides this to /app/.bigset/populate-recipes on a named volume.
+POPULATE_RECIPE_STORE_DIR=.bigset/populate-recipes
+
 # PostHog (optional — leave blank to disable analytics entirely in local dev).
 # Get from https://us.posthog.com/project/settings/general.
 NEXT_PUBLIC_POSTHOG_KEY=

.gitignore

@@ -1,5 +1,6 @@
 .DS_Store
 node_modules/
+backend/node_modules
 .env
 .env.local
 Project_BigSet_brief.md
@@ -22,6 +23,7 @@ tmp/
 temp/
 
 .mastra
+.bigset/
 
 # Local tarballs
 *.tgz

backend/.env.example

@@ -1,6 +1,7 @@
 CLIENT_ORIGIN=http://localhost:3500
 CONVEX_URL=http://localhost:3210
 PORT=3501
+POPULATE_RECIPE_STORE_DIR=.bigset/populate-recipes
 
 # Required once the backend starts writing rows via internal Convex mutations.
 # Generate with: docker compose exec convex ./generate_admin_key.sh

backend/src/env.ts

@@ -24,4 +24,9 @@ export const env = {
   CLERK_PUBLISHABLE_KEY: process.env.CLERK_PUBLISHABLE_KEY,
 
   OPENROUTER_API_KEY: process.env.OPENROUTER_API_KEY,
+
+  // Durable recipe manifests for the self-healing populate layer. In Docker
+  // dev this points at a named volume; locally it defaults under the repo.
+  POPULATE_RECIPE_STORE_DIR:
+    process.env.POPULATE_RECIPE_STORE_DIR || ".bigset/populate-recipes",
 };

backend/src/index.ts

@@ -5,10 +5,12 @@ import { env } from "./env.js";
 import clerkAuthPlugin, { requireAuth } from "./clerk-auth.js";
 import { inferSchema } from "./pipeline/schema-inference.js";
 import { datasetContextSchema } from "./pipeline/populate.js";
-import { populateWorkflow } from "./mastra/workflows/populate.js";
+import { ConvexPopulateDatasetRowWriter } from "./pipeline/populate-convex-writer.js";
+import { runSelfHealingPopulate } from "./pipeline/populate-self-healing-runner.js";
 import { convex, api } from "./convex.js";
 
 const fastify = Fastify({ logger: true });
+const populateRowWriter = new ConvexPopulateDatasetRowWriter();
 
 await fastify.register(fastifyCors, {
   origin: env.CLIENT_ORIGIN,
@@ -72,17 +74,37 @@ await fastify.register(async (instance) => {
       if (dataset.ownerId !== authenticatedUserId) {
         return reply.code(403).send({ error: "Not authorized to populate this dataset" });
       }
+      if (!env.CONVEX_ADMIN_KEY) {
+        return reply.code(500).send({
+          error: "Backend is missing the Convex admin key required for row writes.",
+        });
+      }
 
-      const run = await populateWorkflow.createRun();
-      const result = await run.start({ inputData: parsed.data });
-
-      req.log.info({ workflowStatus: result.status, steps: JSON.stringify(result.steps).slice(0, 2000) }, "Populate workflow completed");
+      const result = await runSelfHealingPopulate({
+        context: parsed.data,
+        recipeStoreDirectory: env.POPULATE_RECIPE_STORE_DIR,
+        rowWriter: populateRowWriter,
+        shouldCommitRows: true,
+      });
 
-      if (result.status !== "success") {
-        throw new Error(`Workflow ended with status: ${result.status}`);
+      req.log.info({
+        action: result.action,
+        datasetId: result.datasetId,
+        committedRows: result.committedRows?.insertedRowCount ?? 0,
+        validationIssues: result.validationIssues.slice(0, 5),
+      }, "Self-healing populate completed");
+
+      if (!result.success) {
+        return reply.code(422).send({
+          error: "Self-healing populate failed validation.",
+          result: responseSafePopulateResult(result),
+        });
       }
 
-      return { success: true, result: result.result };
+      return {
+        success: true,
+        result: responseSafePopulateResult(result),
+      };
     } catch (err) {
       const msg = err instanceof Error ? err.message : String(err);
       if (msg.includes("validator") || msg.includes("Invalid")) {
@@ -100,3 +122,20 @@ try {
   fastify.log.error(err);
   process.exit(1);
 }
+
+function responseSafePopulateResult(
+  result: Awaited<ReturnType<typeof runSelfHealingPopulate>>
+) {
+  const diagnosticRun = result.selectedRun ?? result.diagnosticRun;
+  return {
+    action: result.action,
+    datasetId: result.datasetId,
+    success: result.success,
+    committedRows: result.committedRows,
+    rejectionReasons: result.rejectionReasons,
+    validationIssues: result.validationIssues,
+    productionValidation: diagnosticRun?.productionValidation,
+    metrics: diagnosticRun?.metrics,
+    rowCount: diagnosticRun?.rows.length ?? 0,
+  };
+}

backend/src/pipeline/populate-convex-writer.ts

@@ -0,0 +1,71 @@
+import { env } from "../env.js";
+import { convex, internal } from "../convex.js";
+import type {
+  PopulateDatasetRowWriter,
+  PopulateDatasetWriteResult,
+} from "./populate-self-healing-runner.js";
+
+interface ConvexMutationClient {
+  mutation(functionReference: unknown, args: unknown): Promise<unknown>;
+}
+
+export class ConvexPopulateDatasetRowWriter implements PopulateDatasetRowWriter {
+  constructor(
+    private readonly input: {
+      convexClient?: ConvexMutationClient;
+      internalApi?: typeof internal;
+    } = {}
+  ) {}
+
+  async replaceRows(input: Parameters<PopulateDatasetRowWriter["replaceRows"]>[0]):
+    Promise<PopulateDatasetWriteResult> {
+    if (!env.CONVEX_ADMIN_KEY) {
+      throw new Error(
+        "CONVEX_SELF_HOSTED_ADMIN_KEY is required to commit self-healed populate rows."
+      );
+    }
+
+    const convexClient = this.input.convexClient ?? convex;
+    const internalApi = this.input.internalApi ?? internal;
+    const replacement = await convexClient.mutation(
+      internalApi.datasetRows.replaceByDataset,
+      {
+        datasetId: input.datasetId,
+        rows: input.rows.map((row) => ({
+          data: row.cells,
+          sources: row.sourceUrls,
+        })),
+      }
+    );
+
+    return normalizeReplacementResult(replacement, input.rows.length);
+  }
+}
+
+function normalizeReplacementResult(
+  value: unknown,
+  fallbackInsertedRowCount: number
+): PopulateDatasetWriteResult {
+  if (
+    typeof value === "object" &&
+    value !== null &&
+    "insertedRowCount" in value
+  ) {
+    const replacement = value as {
+      clearedRowCount?: unknown;
+      insertedRowCount?: unknown;
+    };
+    return {
+      clearedRowCount: typeof replacement.clearedRowCount === "number"
+        ? replacement.clearedRowCount
+        : undefined,
+      insertedRowCount: typeof replacement.insertedRowCount === "number"
+        ? replacement.insertedRowCount
+        : fallbackInsertedRowCount,
+    };
+  }
+
+  return {
+    insertedRowCount: fallbackInsertedRowCount,
+  };
+}

backend/src/pipeline/populate-self-healing-runner.ts

@@ -0,0 +1,128 @@
+import { join } from "node:path";
+
+import type { DatasetContext } from "./populate.js";
+import {
+  DefaultPopulateRecipeAuthor,
+  FileSystemPopulateRecipeStore,
+  MastraPopulateRecipeRuntime,
+  SelfHealingPopulateRecipeService,
+  type PopulateRecipeAuthor,
+  type PopulateRecipeRunResult,
+  type PopulateRecipeRuntime,
+  type PopulateRecipeStore,
+  type SelfHealingPopulateTickResult,
+} from "./populate-self-healing.js";
+
+export interface PopulateDatasetRowWriter {
+  replaceRows(input: {
+    datasetId: string;
+    rows: PopulateRecipeRunResult["rows"];
+  }): Promise<PopulateDatasetWriteResult>;
+}
+
+export interface PopulateDatasetWriteResult {
+  clearedRowCount?: number;
+  insertedRowCount: number;
+}
+
+export interface RunSelfHealingPopulateInput {
+  context: DatasetContext;
+  store?: PopulateRecipeStore;
+  runtime?: PopulateRecipeRuntime;
+  author?: PopulateRecipeAuthor;
+  rowWriter?: PopulateDatasetRowWriter;
+  shouldCommitRows?: boolean;
+  recipeStoreDirectory?: string;
+}
+
+export interface RunSelfHealingPopulateResult {
+  success: boolean;
+  action: SelfHealingPopulateTickResult["action"];
+  datasetId: string;
+  selectedRun?: PopulateRecipeRunResult;
+  diagnosticRun?: PopulateRecipeRunResult;
+  committedRows?: PopulateDatasetWriteResult;
+  rejectionReasons: string[];
+  validationIssues: string[];
+  tick: SelfHealingPopulateTickResult;
+}
+
+export async function runSelfHealingPopulate(
+  input: RunSelfHealingPopulateInput
+): Promise<RunSelfHealingPopulateResult> {
+  if (input.shouldCommitRows && !input.rowWriter) {
+    throw new Error("rowWriter is required when shouldCommitRows is true.");
+  }
+  const rowWriter = input.rowWriter;
+
+  const store = input.store ?? new FileSystemPopulateRecipeStore(
+    input.recipeStoreDirectory ?? defaultPopulateRecipeStoreDirectory()
+  );
+  const service = new SelfHealingPopulateRecipeService({
+    store,
+    runtime: input.runtime ?? new MastraPopulateRecipeRuntime(),
+    author: input.author ?? new DefaultPopulateRecipeAuthor(),
+  });
+  const tick = await service.tick({
+    datasetId: input.context.datasetId,
+    context: input.context,
+  });
+  const selectedRun = successfulRunForTick(tick);
+  const diagnosticRun = diagnosticRunForTick(tick);
+  let committedRows: PopulateDatasetWriteResult | undefined;
+
+  if (input.shouldCommitRows && selectedRun && rowWriter) {
+    committedRows = await rowWriter.replaceRows({
+      datasetId: input.context.datasetId,
+      rows: selectedRun.rows,
+    });
+  }
+
+  return {
+    success: Boolean(selectedRun),
+    action: tick.action,
+    datasetId: input.context.datasetId,
+    selectedRun,
+    diagnosticRun,
+    committedRows,
+    rejectionReasons: tick.rejectionReasons,
+    validationIssues: validationIssuesForSelfHealingTick(tick),
+    tick,
+  };
+}
+
+export function successfulRunForTick(
+  tick: SelfHealingPopulateTickResult
+): PopulateRecipeRunResult | undefined {
+  if (tick.action === "active_rerun_succeeded") {
+    return tick.activeRun;
+  }
+  if (
+    tick.action === "generated_initial_recipe" ||
+    tick.action === "repaired_active_recipe"
+  ) {
+    return tick.candidateRun;
+  }
+  return undefined;
+}
+
+export function diagnosticRunForTick(
+  tick: SelfHealingPopulateTickResult
+): PopulateRecipeRunResult | undefined {
+  return successfulRunForTick(tick) ?? tick.candidateRun ?? tick.activeRun;
+}
+
+export function validationIssuesForSelfHealingTick(
+  tick: SelfHealingPopulateTickResult
+): string[] {
+  const run = diagnosticRunForTick(tick);
+  return Array.from(new Set([
+    ...(run?.validationIssues ?? []),
+    ...(run?.productionValidation.criticalIssues ?? []),
+    ...tick.rejectionReasons,
+  ]));
+}
+
+function defaultPopulateRecipeStoreDirectory(): string {
+  return join(process.cwd(), ".bigset", "populate-recipes");
+}