reverse-engineering-anything/src/sqlite/SqliteDatabaseWorkerExecution.ts

import {
  bindArtifactFileCleanup,
  ArtifactFileCleanupFailure,
  artifactFileCancellationCause,
} from "../artifacts/ArtifactFileOperation.js";
import { randomUUID } from "node:crypto";
import { writeFile } from "node:fs/promises";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { z } from "zod";
import type { AnalysisCapturedOutput } from "../domain/analysisErrorBase.js";
import {
  AnalysisCancelledError,
  AnalysisInputError,
  AnalysisCapabilityUnavailableError,
  AnalysisOutputError,
  AnalysisResourceConstraintError,
} from "../domain/analysisErrorCore.js";
import {
  sqliteDatabaseSchema,
  type InspectSqliteDatabaseInput,
} from "../domain/sqlite/sqliteDatabase.js";
import { readStableArtifact } from "../artifacts/readStableArtifact.js";
import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js";
import type { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
import { runOwnedCommand } from "../process/OwnedCommand.js";
import type {
  ProviderProcessSupervisor,
  ProviderProcessStopResult,
} from "../process/ProviderProcess.js";
import { capturedSqliteOutput } from "./SqliteDatabaseFailures.js";
import { SQLITE_DATABASE_LIMITS } from "./SqliteDatabaseLimits.js";

const operation = "inspect_sqlite_database";
const {
  artifact: _artifactSchema,
  wal: _walSchema,
  worker_timeout_ms: _workerTimeoutSchema,
  ...inspectionShape
} = sqliteDatabaseSchema.shape;
const inspectionSchema = z.strictObject(inspectionShape);
const replySchema = z.union([
  z.strictObject({ ok: z.literal(true), inspection: inspectionSchema }),
  z.strictObject({
    ok: z.literal(false),
    reason: z.enum([
      "format",
      "selection",
      "column-selection",
      "unsupported",
      "unavailable",
      "output-limit",
      "memory",
    ]),
    message: z.string(),
  }),
  z.strictObject({
    ok: z.literal(false),
    reason: z.literal("column-evaluation"),
    message: z.string(),
    readable_columns: z.array(z.string()),
  }),
]);
/** Replaceable launcher retains the shared owned-process supervision boundary. */
export type SqliteWorkerLauncher = NonNullable<
  Parameters<typeof runOwnedCommand>[2]
>["launcher"];

// Each input rejection names the caller argument that requires correction.
const rejectedArgument = {
  format: "path",
  unsupported: "path",
  selection: "table",
  "column-selection": "columns",
} as const;

const rejectedReply = (
  reply: z.output<typeof replySchema> & { ok: false },
  selectedPath: string,
  capturedOutput: AnalysisCapturedOutput,
) => {
  if (reply.reason === "unavailable")
    return new AnalysisCapabilityUnavailableError(
      "sqlite",
      operation,
      reply.message,
      { capturedOutput, userMessage: reply.message },
    );
  if (reply.reason === "column-evaluation") {
    const readable = reply.readable_columns;
    return new AnalysisCapabilityUnavailableError(
      "sqlite",
      operation,
      reply.message,
      {
        capturedOutput,
        userMessage: reply.message,
        remediationAction: `The database is readable; only the listed column values failed. REA's SQLite connection loads no extensions or application functions and disables schema trust. ${
          readable.length === 0
            ? "No other readable column was found"
            : `Select columns whose values were read: columns: ${JSON.stringify(readable)} (CLI: one --column per name)`
        }; omit table and use schema_scope: database to read the stored declarations without evaluating rows. Obtaining the failed values needs a SQLite connection able to evaluate them. Do not repeat the unchanged request.`,
      },
    );
  }
  if (reply.reason === "memory")
    return new AnalysisResourceConstraintError(
      operation,
      "memory",
      reply.message,
      {
        javascript_heap_bytes: 256 * 1024 * 1024,
      },
      { capturedOutput },
    );
  if (reply.reason === "output-limit")
    return new AnalysisResourceConstraintError(
      operation,
      "transport",
      reply.message,
      {
        sqlite_value_or_record_bytes: SQLITE_DATABASE_LIMITS.outputBytes,
        worker_reply_bytes: SQLITE_DATABASE_LIMITS.outputBytes,
      },
      {
        capturedOutput,
        retryable: false,
        remediationAction:
          "The SQLite value/record and worker JSON reply allowances are fixed at 16 MiB; worker_timeout_ms and client deadlines do not raise them. If it preserves the requested facts, select exact columns, reduce row_limit, or use schema_scope: inventory instead of database. This worker refusal leaves no completed Evidence to retrieve. Do not repeat the unchanged request or truncate required database bytes.",
      },
    );
  const argument = rejectedArgument[reply.reason];
  return new AnalysisInputError(operation, { capturedOutput }, [
    {
      path: [argument],
      reason: "invalid_format",
      // The worker sees only the private snapshot, so name the caller's selection here.
      message:
        argument === "path"
          ? `Selected SQLite database is invalid (${reply.message}): ${selectedPath}`
          : reply.message,
    },
  ]);
};

/** Execute only the private snapshot and validate the bounded worker protocol. */
export const executeSqliteDatabaseWorker = async (request: {
  readonly input: InspectSqliteDatabaseInput;
  readonly root: string;
  readonly snapshotPath: string;
  readonly timeoutMs: number;
  readonly environment: Readonly<NodeJS.ProcessEnv>;
  readonly resourceScope: ArtifactResourceScope;
  readonly launcher?: SqliteWorkerLauncher;
  readonly signal?: AbortSignal;
  readonly captured: (output: AnalysisCapturedOutput) => void;
  readonly retainCleanup?: (
    owner: ProviderProcessSupervisor,
    result: Extract<ProviderProcessStopResult, { status: "incomplete" }>,
  ) => void;
}): Promise<z.output<typeof inspectionSchema>> => {
  const requestPath = join(request.root, "request.json");
  const replyPath = join(request.root, "reply.json");
  await writeFile(
    requestPath,
    JSON.stringify({
      snapshot_path: request.snapshotPath,
      reply_path: replyPath,
      input: request.input,
    }),
    { flag: "wx", mode: 0o600 },
  );
  const { NODE_OPTIONS: _ambientNodeOptions, ...environment } =
    request.environment;
  const execution = await runOwnedCommand(
    {
      command: process.execPath,
      arguments: [
        "--max-old-space-size=256",
        fileURLToPath(new URL("./SqliteDatabaseWorker.js", import.meta.url)),
        requestPath,
      ],
      cwd: request.root,
      expectedCommand: null,
      runId: `rea-sqlite-database-${randomUUID()}`,
      hostEnvironment: environment,
    },
    {
      timeoutMs: request.timeoutMs,
      diagnosticBytes: SQLITE_DATABASE_LIMITS.diagnosticBytes,
    },
    {
      ...(request.signal === undefined ? {} : { signal: request.signal }),
      ...(request.launcher === undefined ? {} : { launcher: request.launcher }),
      ...(request.retainCleanup === undefined
        ? {}
        : { retainCleanup: request.retainCleanup }),
    },
  );
  const capturedOutput = capturedSqliteOutput(execution);
  request.captured(capturedOutput);
  let reply: z.output<typeof replySchema>;
  try {
    const file = await readStableArtifact(
      replyPath,
      SQLITE_DATABASE_LIMITS.outputBytes,
      request.signal,
      undefined,
      request.resourceScope,
    );
    reply = replySchema.parse(JSON.parse(file.bytes.toString("utf8")));
  } catch (cause: unknown) {
    const options = {
      cause,
      capturedOutput,
      ...(cause instanceof ArtifactReaderFailure && cause.cleanup !== undefined
        ? { cleanup: cause.cleanup }
        : {}),
    };
    if (
      cause instanceof ArtifactReaderFailure &&
      (cause.reason === "cancelled" ||
        (cause instanceof ArtifactFileCleanupFailure &&
          cause.fileCleanupFailure.outcome.kind === "completed" &&
          request.signal?.aborted))
    ) {
      const error = new AnalysisCancelledError(operation, {
        ...options,
        cause:
          cause.reason === "cancelled"
            ? artifactFileCancellationCause(cause)
            : request.signal?.reason,
      });
      bindArtifactFileCleanup(cause, error);
      throw error;
    }
    const error = new AnalysisOutputError(
      operation,
      `Owned SQLite worker reply is missing or malformed: ${cause instanceof Error ? cause.message : String(cause)}`,
      options,
    );
    bindArtifactFileCleanup(cause, error);
    throw error;
  }
  if (!reply.ok) throw rejectedReply(reply, request.input.path, capturedOutput);
  return reply.inspection;
};