libsql-provider.test.ts

/**
 * libSQL provider, end to end (issue #424 Phase 5)
 *
 * Every payload below was captured on 2026-08-27 from BOTH deployments this one
 * type-id reaches: a self-hosted `ghcr.io/tursodatabase/libsql-server` (sqld
 * 0.24.33, SQLite 3.47.0) and a Turso Cloud database in `aws-eu-west-1`.
 * `globalThis.fetch` is REPLACED per test and restored afterwards - `mock.module()`
 * is refused, being process-wide in bun and able to poison sibling files - so the
 * real provider, the real introspection and the real Hrana transport all execute
 * here and only the server is fake.
 *
 * Five measured behaviours drive what is asserted:
 *
 * 1. A FAILED STATEMENT IS AN HTTP 200 with the failure inside `results[]`, so
 *    `response.ok` is never the test.
 * 2. THE TWO DEPLOYMENTS WORD THE SAME REFUSAL DIFFERENTLY - "unsupported
 *    statement: VACUUM" against "SQL not allowed statement: VACUUM" - under one
 *    code, so nothing may key on the wording.
 * 3. `GET /version` IS A SQLD ROUTE TURSO CLOUD DOES NOT HAVE, and a deployment
 *    that publishes no version is not a broken one.
 * 4. AN AUTH FAILURE USES A DIFFERENT ENVELOPE (`{"error": "<string>"}`) and
 *    answers 401 with no token, 400 with a malformed one.
 * 5. `dbstat` ANSWERS ON BOTH, so per-table bytes here are measured - which
 *    `bun:sqlite` cannot do at all.
 */
import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test";
import { assertObjectSurface } from "../../helpers/object-surface-conformance";
import {
  AuthenticationError,
  ConnectionError,
  DatabaseConfigError,
  DatabaseError,
  ExecutionProfileError,
  QueryError,
} from "@/lib/db/errors";
import {
  containerDepth,
  isCountUnavailable,
  isSourcePartUnavailable,
  kindHasColumns,
  sourceBoundTruncationReason,
} from "@/lib/db/object-kinds";
import { LibSQLProvider } from "@/lib/db/providers/sql/libsql";
import { SQLITE_CREATE_TABLE } from "@/lib/db/providers/sql/sqlite-create-table";
import { buildResultExport } from "@/lib/export/result-export";
import { countLibSQLObjects, type LibSQLObjectReader } from "@/lib/db/providers/sql/libsql/objects";
import type {
  DatabaseConnection,
  DatabaseProvider,
  ObjectKindSpec,
  ProviderExecutionContext,
  ReadOnlyStatementBudget,
} from "@/lib/db/types";
import { readFileSync } from "node:fs";
import { createServer } from "node:http";
import type { AddressInfo } from "node:net";
import { join } from "node:path";
import { comparePaths } from "@/lib/db/object-path";
import { MCP_STATEMENT_READ } from "@/lib/db/compatibility";
import { acquireExecutionProfileProvider, clearProviderCache, type ExecutionProfile } from "@/lib/db/factory";
import { readQueryFormsOf } from "@/lib/db/providers/sql/read-query-forms";
import type { McpConnectionContext } from "@/lib/mcp/context";
import { MCP_ENGINE_ERROR_PREFIX, MCP_UNTRUSTED_NOTICE } from "@/lib/mcp/output";
import { runStatement, type McpStatementInput } from "@/lib/mcp/tools/run-statement";
import { mcpStatementArm, type McpStatementArm } from "@/lib/mcp/tools/statement-arms";
import type { ManagedConnection } from "@/lib/seed";
import { RUN_READ_QUERY_ENGINES, RunReadQueryInputSchema, runReadQuery } from "../../helpers/run-read-query";

// ============================================================================
// Harness
// ============================================================================

interface FetchCall {
  url: string;
  body: string | null;
}

const originalFetch = globalThis.fetch;
let calls: FetchCall[] = [];

/** A row set as Hrana encodes one: declared columns, positional typed values. */
type Cell = Record<string, unknown>;

function result(cols: [string, string | null][], rows: Cell[][], extra: Record<string, unknown> = {}): Cell {
  return {
    type: "ok",
    response: {
      type: "execute",
      result: {
        cols: cols.map(([name, decltype]) => ({ name, decltype })),
        rows,
        affected_row_count: 0,
        last_insert_rowid: null,
        replication_index: "1",
        rows_read: rows.length,
        rows_written: 0,
        query_duration_ms: 0.107,
        ...extra,
      },
    },
  };
}

function text(value: string): Cell {
  return { type: "text", value };
}

function int(value: number | string): Cell {
  return { type: "integer", value: String(value) };
}

function failure(message: string, code: string): Cell {
  return { type: "error", error: { message, code } };
}

/**
 * The fixture both deployments answered for, verbatim in shape: two tables, one
 * user index, one foreign key, `dbstat` populated.
 */
function answerFor(sql: string): Cell {
  if (/FROM sqlite_master\s+WHERE type = 'table'\s+AND name NOT LIKE/.test(sql) && /COUNT/.test(sql)) {
    return result([["table_count", null]], [[int(2)]]);
  }
  if (/FROM sqlite_master\s+WHERE type = 'table'/.test(sql)) {
    return result([["name", "TEXT"]], [[text("probe_customers")], [text("probe_orders")]]);
  }
  if (/type = 'index' AND name NOT LIKE/.test(sql)) return result([["index_count", null]], [[int(1)]]);
  if (/SELECT name, tbl_name FROM sqlite_master WHERE type = 'index'/.test(sql)) {
    return result(
      [
        ["name", "TEXT"],
        ["tbl_name", "TEXT"],
      ],
      [[text("idx_country"), text("probe_customers")]],
    );
  }
  if (/FROM dbstat/.test(sql)) {
    return result(
      [
        ["name", "TEXT"],
        ["bytes", null],
      ],
      [
        [text("probe_customers"), int(8192)],
        [text("idx_country"), int(4096)],
        [text("probe_orders"), int(270336)],
      ],
    );
  }
  if (/COUNT\(\*\) AS row_count FROM "probe_customers"/.test(sql)) return result([["row_count", null]], [[int(3)]]);
  if (/COUNT\(\*\) AS row_count FROM "probe_orders"/.test(sql)) return result([["row_count", null]], [[int(2000)]]);
  if (/pragma_table_info\('probe_customers'\)/.test(sql)) {
    return result(
      [
        ["cid", null],
        ["name", null],
        ["type", null],
        ["notnull", null],
        ["dflt_value", null],
        ["pk", null],
      ],
      [
        [int(0), text("id"), text("INTEGER"), int(1), { type: "null" }, int(1)],
        [int(1), text("country"), text("TEXT"), int(0), { type: "null" }, int(0)],
      ],
    );
  }
  if (/pragma_table_info\('probe_orders'\)/.test(sql)) {
    return result(
      [
        ["cid", null],
        ["name", null],
        ["type", null],
        ["notnull", null],
        ["dflt_value", null],
        ["pk", null],
      ],
      [[int(0), text("id"), text("INTEGER"), int(0), { type: "null" }, int(1)]],
    );
  }
  if (/pragma_index_list\('probe_customers'\)/.test(sql)) {
    return result(
      [
        ["seq", null],
        ["name", null],
        ["unique", null],
        ["origin", null],
      ],
      [
        [int(0), text("idx_country"), int(1), text("c")],
        [int(1), text("sqlite_autoindex_probe_customers_1"), int(1), text("pk")],
      ],
    );
  }
  if (/pragma_index_info\('idx_country'\)/.test(sql)) {
    return result(
      [
        ["seqno", null],
        ["cid", null],
        ["name", null],
      ],
      [[int(0), int(1), text("country")]],
    );
  }
  if (/pragma_foreign_key_list\('probe_orders'\)/.test(sql)) {
    return result(
      [
        ["id", null],
        ["seq", null],
        ["table", null],
        ["from", null],
        ["to", null],
      ],
      [[int(0), int(0), text("probe_customers"), text("customer_id"), text("id")]],
    );
  }
  if (/pragma_page_count/.test(sql)) return result([["size_bytes", null]], [[int(282624)]]);
  if (/sqlite_version\(\)/.test(sql)) return result([["version", null]], [[text("3.47.0")]]);
  if (/PRAGMA integrity_check/.test(sql)) return result([["integrity_check", null]], [[text("ok")]]);
  if (/PRAGMA journal_mode/.test(sql)) return result([["journal_mode", null]], [[text("wal")]]);
  if (/^SELECT 1$/.test(sql)) return result([["1", null]], [[int(1)]]);

  return result([], []);
}

/**
 * What the server does with one pipeline: an answer per statement, then a close.
 *
 * The bound arguments are handed over as well as the statement, because the object
 * surface binds the object NAME rather than embedding it (`pragma_table_xinfo(?, ?)`
 * is one statement for every object), so a server that saw only the SQL could not
 * answer per object - and could not tell a name bind from a container one, which is
 * the whole point of the two-level test below.
 */
type Server = (sql: string, args: unknown[]) => Cell;

let server: Server = answerFor;
let versionRoute: () => Response = () => new Response("sqld 0.24.33 (f8fb14f3 2026-08-11)", { status: 200 });

function installFetch(): void {
  globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => {
    const url = String(input);
    const body = init?.body === undefined ? null : String(init.body);
    calls.push({ url, body });

    if (url.endsWith("/version")) return Promise.resolve(versionRoute());

    const requests = (JSON.parse(body ?? "{}") as { requests?: { type: string; stmt?: Statement }[] }).requests;
    const results = (requests ?? [])
      .filter((request) => request.type === "execute")
      .map((request) => server(String(request.stmt?.sql), boundArgs(request.stmt)));

    return Promise.resolve(
      new Response(JSON.stringify({ baton: null, base_url: null, results: [...results, { type: "ok" }] }), {
        status: 200,
        headers: { "content-type": "application/json" },
      }),
    );
  }) as unknown as typeof fetch;
}

function connection(overrides: Partial<DatabaseConnection> = {}): DatabaseConnection {
  return {
    id: "libsql-probe",
    name: "libSQL probe",
    type: "libsql",
    host: "127.0.0.1",
    port: 18081,
    createdAt: new Date("2026-08-27T00:00:00.000Z"),
    ...overrides,
  };
}

async function connected(overrides: Partial<DatabaseConnection> = {}): Promise<LibSQLProvider> {
  const provider = new LibSQLProvider(connection(overrides));
  await provider.connect();
  return provider;
}

/** One statement as it left this process: what was sent, and what was bound into it. */
interface Statement {
  sql?: string;
  args?: { value?: unknown }[];
}

/** The values bound into one statement, in order, with the wire encoding unwrapped. */
function boundArgs(statement: Statement | undefined): unknown[] {
  return (statement?.args ?? []).map((arg) => arg.value);
}

/** Every statement the provider sent, with its bindings, in order. */
function sentCalls(): { sql: string; args: unknown[] }[] {
  return calls
    .filter((call) => call.body !== null)
    .flatMap((call) => {
      const requests = (JSON.parse(call.body as string) as { requests?: { stmt?: Statement }[] }).requests ?? [];
      return requests
        .filter((request) => request.stmt !== undefined)
        .map((request) => ({ sql: String(request.stmt?.sql), args: boundArgs(request.stmt) }));
    });
}

/** Every statement the provider sent, in order. */
function sentStatements(): string[] {
  return sentCalls().map((call) => call.sql);
}

beforeEach(() => {
  calls = [];
  server = answerFor;
  versionRoute = () => new Response("sqld 0.24.33 (f8fb14f3 2026-08-11)", { status: 200 });
  installFetch();
});

afterEach(() => {
  globalThis.fetch = originalFetch;
});

// ============================================================================
// Configuration and connection
// ============================================================================

describe("LibSQLProvider configuration", () => {
  test("refuses a connection with neither a host nor a URL", () => {
    expect(() => new LibSQLProvider(connection({ host: undefined }))).toThrow(DatabaseConfigError);
  });

  test("resolves a libsql:// URL into host, port and token", async () => {
    const provider = await connected({
      host: undefined,
      port: undefined,
      connectionString: "libsql://libredb-probe-424-cevheri.aws-eu-west-1.turso.io?authToken=jwt-123",
    });

    // Port 443 is the https default, which the URL leaves out of its serialization.
    expect(calls[0]?.url).toBe("https://libredb-probe-424-cevheri.aws-eu-west-1.turso.io/v2/pipeline");
    await provider.disconnect();
  });

  test("connects with the cheapest statement there is, not a health route", async () => {
    await connected();

    expect(sentStatements()).toEqual(["SELECT 1"]);
  });

  test("connecting twice reuses the transport rather than probing again", async () => {
    const provider = await connected();
    await provider.connect();

    expect(sentStatements()).toEqual(["SELECT 1"]);
    await provider.disconnect();
  });

  test("reports a missing token as an authentication failure, not a connection one", async () => {
    // 401, and the envelope is `{"error": "<string>"}` rather than the statement
    // shape - captured from Turso Cloud with no Authorization header.
    globalThis.fetch = (() =>
      Promise.resolve(
        new Response(
          JSON.stringify({ error: "Unauthorized: `unauthorized access attempt on database: empty JWT token`" }),
          {
            status: 401,
          },
        ),
      )) as unknown as typeof fetch;

    await expect(new LibSQLProvider(connection()).connect()).rejects.toBeInstanceOf(AuthenticationError);
  });

  test("reports a malformed token as an authentication failure even though the status is 400", async () => {
    globalThis.fetch = (() =>
      Promise.resolve(
        new Response(JSON.stringify({ error: "JWT error: InvalidToken" }), { status: 400 }),
      )) as unknown as typeof fetch;

    await expect(new LibSQLProvider(connection({ password: "notatoken" })).connect()).rejects.toBeInstanceOf(
      AuthenticationError,
    );
  });

  test("reports an unreachable server as a connection failure naming host and port", async () => {
    globalThis.fetch = (() => Promise.reject(new Error("connect ECONNREFUSED"))) as unknown as typeof fetch;

    const failed = new LibSQLProvider(connection()).connect();

    await expect(failed).rejects.toBeInstanceOf(ConnectionError);
    await expect(failed).rejects.toThrow(/connect ECONNREFUSED/);
  });
});

/**
 * Which credential a connection sends, through the real provider: `connect()` builds the transport from the
 * resolved connection, so this is where a `user` either reaches the header or is lost on the way.
 *
 * sqld's legacy HTTP Basic auth (`SQLD_HTTP_AUTH="basic:<base64(user:password)>"`) pairs a user with a password,
 * so a connection that names a user sends the pair as Basic. A `libsql://` URL authenticates with the token it
 * carries, so a `user` beside it is ignored and the URL path keeps sending a bearer token.
 */
describe("LibSQLProvider credentials", () => {
  /**
   * The Authorization header of every request from here on, in order, null where none was sent. It wraps the
   * fake server `installFetch` put in place rather than replacing it, so the provider still gets real answers,
   * and `afterEach` puts the real `fetch` back underneath both.
   */
  function authorizationHeaders(): (string | null)[] {
    const seen: (string | null)[] = [];
    const answering = globalThis.fetch;
    globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => {
      seen.push(new Headers(init?.headers).get("authorization"));
      return answering(input, init);
    }) as unknown as typeof fetch;
    return seen;
  }

  test("sends a user and password as HTTP Basic on the connect probe and on every query after it", async () => {
    const seen = authorizationHeaders();
    const provider = await connected({ user: "libsql", password: "libsql-pass" });
    await provider.query("SELECT 1");

    expect(seen).toEqual(["Basic bGlic3FsOmxpYnNxbC1wYXNz", "Basic bGlic3FsOmxpYnNxbC1wYXNz"]);
    await provider.disconnect();
  });

  test("keeps the bearer token when the connection names no user", async () => {
    const seen = authorizationHeaders();
    const provider = await connected({ password: "tok-123" });

    expect(seen).toEqual(["Bearer tok-123"]);
    await provider.disconnect();
  });

  test("keeps a libsql:// URL's token as the bearer credential when the connection also names a user", async () => {
    const seen = authorizationHeaders();
    const provider = await connected({
      host: undefined,
      port: undefined,
      user: "ignored",
      connectionString: "libsql://libredb-probe-424-cevheri.aws-eu-west-1.turso.io?authToken=jwt-123",
    });

    expect(seen).toEqual(["Bearer jwt-123"]);
    await provider.disconnect();
  });

  test("keeps a libsql:// URL's own password as the bearer credential, its user name ignored", async () => {
    const seen = authorizationHeaders();
    const provider = await connected({
      host: undefined,
      port: undefined,
      user: "also-ignored",
      connectionString: "libsql://url-user:p%40ss@db.turso.io",
    });

    expect(seen).toEqual(["Bearer p@ss"]);
    await provider.disconnect();
  });

  test("drops the user of a connection whose connection string does not parse, so it sends no Basic header", async () => {
    // The string is ignored and the host and port fields are used (docs/BACKLOG.md D232), but the user is dropped
    // as on every connection-string path, so the password still goes out as a bearer token.
    const seen = authorizationHeaders();
    const provider = await connected({ user: "dropped", password: "tok-456", connectionString: "not a url" });

    expect(seen).toEqual(["Bearer tok-456"]);
    await provider.disconnect();
  });

  test("reports a Basic credential the server refused as an authentication failure", async () => {
    // The body sqld's source builds for a refused Basic credential, AuthError::BasicRejected inside
    // Error::AuthError, answered 401 (`libsql-server/src/error.rs` and `src/auth/errors.rs` at
    // libsql-server-v0.24.32). Read from the source rather than captured live; the provider keys on the status.
    globalThis.fetch = (() =>
      Promise.resolve(
        new Response(
          JSON.stringify({ error: "Unauthorized: `The `Basic` HTTP authentication credentials were rejected`" }),
          { status: 401 },
        ),
      )) as unknown as typeof fetch;

    await expect(
      new LibSQLProvider(connection({ user: "libsql", password: "wrong" })).connect(),
    ).rejects.toBeInstanceOf(AuthenticationError);
  });
});

// ============================================================================
// Capabilities
// ============================================================================

describe("LibSQLProvider capabilities", () => {
  test("offers only the two maintenance operations the server accepts", () => {
    const capabilities = new LibSQLProvider(connection()).getCapabilities();

    expect(capabilities.maintenanceOperations).toEqual(["reindex", "check"]);
    // Measured on BOTH deployments: each of these is refused by the server's own
    // statement allowlist, so a control for it could only ever fail.
    expect(capabilities.maintenanceOperations).not.toContain("vacuum");
    expect(capabilities.maintenanceOperations).not.toContain("analyze");
  });

  test("reads EXPLAIN the way SQLite does, and takes a pasted URL", () => {
    const capabilities = new LibSQLProvider(connection()).getCapabilities();

    expect(capabilities.explainFormat).toBe("sqlite-queryplan");
    expect(capabilities.supportsConnectionString).toBe(true);
    expect(capabilities.defaultPort).toBe(8080);
  });

  test("pages with LIMIT n OFFSET m, the grammar it shares with SQLite", () => {
    // #816: declared true because `prepareQuery` really applies a positive offset.
    // Measured across every type-id in tests/unit/db/result-pagination-capability.test.ts.
    const provider = new LibSQLProvider(connection());
    expect(provider.getCapabilities().supportsResultPagination).toBe(true);

    const pageTwo = provider.prepareQuery("SELECT * FROM t", { limit: 50, offset: 50 });
    expect(pageTwo.query).toBe("SELECT * FROM t LIMIT 50 OFFSET 50");
    expect(pageTwo.wasLimited).toBe(true);
  });

  test("declares SQLite's create-table grammar, the same object, because it embeds SQLite", () => {
    // #1822: shared by import, never by a type-id branch; docs/providers/sqlite.md, "Create Table form".
    expect(new LibSQLProvider(connection()).getCapabilities().createTable).toBe(SQLITE_CREATE_TABLE);
  });

  test("declares no transaction, because the stream closes with each statement", () => {
    expect(new LibSQLProvider(connection()).getCapabilities().supportsTransactions).toBe(false);
  });

  test("tells the reader there is no statement history to enable", () => {
    expect(new LibSQLProvider(connection()).getLabels().slowQueriesEmptyState).toBe(
      "libSQL keeps no statistics about finished statements, so there is nothing to enable.",
    );
  });
});

// ============================================================================
// Queries
// ============================================================================

describe("LibSQLProvider query", () => {
  test("returns rows, declared fields and the types the engine declared", async () => {
    const provider = await connected();

    const answer = await provider.query("SELECT name, type FROM sqlite_master WHERE type = 'table'");

    expect(answer.rows).toEqual([{ name: "probe_customers" }, { name: "probe_orders" }]);
    expect(answer.fields).toEqual(["name"]);
    expect(answer.columnTypes).toEqual({ name: "TEXT" });
    await provider.disconnect();
  });

  test("a repeated or empty column name keeps its own value and type", async () => {
    // The shape sqld answers for `SELECT o.id, c.id, 1 AS "" ...`: two columns declared `id`
    // and one declared with an empty name. Keyed by name, the second id replaced the first.
    server = () =>
      result(
        [
          ["id", "INTEGER"],
          ["id", "INTEGER"],
          ["", null],
        ],
        [[int(100), int(1), int(7)]],
      );
    const provider = await connected();

    const answer = await provider.query('SELECT o.id, c.id, 1 AS "" FROM probe_orders o JOIN probe_customers c');

    expect(answer.fields).toEqual(["id", "id (2)", "(No column name)"]);
    expect(answer.rows).toEqual([{ id: 100, "id (2)": 1, "(No column name)": 7 }]);
    expect(answer.columnTypes).toEqual({ id: "INTEGER", "id (2)": "INTEGER" });
    await provider.disconnect();
  });

  test("omits columnTypes entirely when the engine declared none", async () => {
    const provider = await connected();

    const answer = await provider.query("SELECT sqlite_version() AS version");

    expect(answer.rows).toEqual([{ version: "3.47.0" }]);
    expect(answer.columnTypes).toBeUndefined();
    await provider.disconnect();
  });

  test("counts a write by what the engine says it changed", async () => {
    server = () => result([], [], { affected_row_count: 3, last_insert_rowid: "12" });
    const provider = await connected();

    const answer = await provider.query("DELETE FROM probe_orders WHERE id < 4");

    expect(answer.rowCount).toBe(3);
    expect(answer.rows).toEqual([]);
    await provider.disconnect();
  });

  test("binds positional parameters as the protocol requires", async () => {
    const provider = await connected();

    await provider.query("SELECT * FROM probe_customers WHERE country = ?", ["tr"]);

    const sent = JSON.parse(calls[1]?.body ?? "{}") as { requests: { stmt?: { args?: unknown[] } }[] };
    expect(sent.requests[0]?.stmt?.args).toEqual([{ type: "text", value: "tr" }]);
    await provider.disconnect();
  });

  test("hands a past-2^53 key back as its exact digits and binds those digits back as the integer", async () => {
    // The whole inline-grid round trip through the real provider: read the key out
    // of a row, then send that very value back in the WHERE clause. Measured against
    // sqld 0.24.33 on 2026-09-18: bound as TEXT this matched 0 rows on a column with
    // no affinity (BLOB or undeclared), so the editor reported nothing changed;
    // bound as an integer it matches the one row, on every column declaration.
    server = () => result([["id", null]], [[int("9007199254740993")]]);
    const provider = await connected();

    const read = await provider.query('SELECT id FROM "probe_orders"');
    const key = read.rows[0]?.id;
    expect(key).toBe("9007199254740993");

    server = () => result([], [], { affected_row_count: 1 });
    await provider.query('UPDATE "probe_orders" SET note = ? WHERE id = ?', ["duzenlendi", key]);

    const sent = JSON.parse(calls.at(-1)?.body ?? "{}") as { requests: { stmt?: { args?: unknown[] } }[] };
    expect(sent.requests[0]?.stmt?.args).toEqual([
      { type: "text", value: "duzenlendi" },
      { type: "integer", value: "9007199254740993" },
    ]);
    await provider.disconnect();
  });

  test("hands a BLOB to the wire as bytes, and the SQL export writes it back as X'..'", async () => {
    // Measured against sqld 0.24.33 on 2026-10-04: `x'DEADBEEF00FF'` answers
    // `{"type":"blob","base64":"3q2+7wD/"}` and `x''` answers base64 "". Before the
    // transport decoded to a Buffer, the grid received `{"0":222,...}` and the SQL export
    // wrote that object as quoted text.
    server = () =>
      result(
        [
          ["bin", "BLOB"],
          ["empty", "BLOB"],
        ],
        [
          [
            { type: "blob", base64: "3q2+7wD/" },
            { type: "blob", base64: "" },
          ],
        ],
      );
    const provider = await connected();

    const read = await provider.query("SELECT bin, empty FROM t_types");
    const wire = JSON.parse(JSON.stringify(read.rows)) as Record<string, unknown>[];
    expect(wire).toEqual([
      { bin: { type: "Buffer", data: [0xde, 0xad, 0xbe, 0xef, 0x00, 0xff] }, empty: { type: "Buffer", data: [] } },
    ]);
    const exported = buildResultExport("sql-insert", {
      rows: wire,
      fields: read.fields,
      tabName: "t",
      dialect: "libsql",
    });
    expect(exported.content).toBe(`INSERT INTO t ("bin", "empty") VALUES (X'deadbeef00ff', X'');`);
    await provider.disconnect();
  });

  test("still binds an ordinary key that happens to be all digits as text", async () => {
    const provider = await connected();

    await provider.query('UPDATE "probe_customers" SET country = ? WHERE code = ?', ["tr", "007"]);

    const sent = JSON.parse(calls.at(-1)?.body ?? "{}") as { requests: { stmt?: { args?: unknown[] } }[] };
    expect(sent.requests[0]?.stmt?.args).toEqual([
      { type: "text", value: "tr" },
      { type: "text", value: "007" },
    ]);
    await provider.disconnect();
  });

  test("surfaces SQLite's own wording for a statement the engine rejected", async () => {
    const provider = await connected();
    server = () => failure("SQLite error: no such table: nope", "SQLITE_UNKNOWN");

    const failed = provider.query("SELECT * FROM nope");

    await expect(failed).rejects.toBeInstanceOf(QueryError);
    await expect(failed).rejects.toThrow("SQLite error: no such table: nope");
    await provider.disconnect();
  });

  test("surfaces a refusal the same way whichever deployment worded it", async () => {
    // sqld says "unsupported statement"; Turso Cloud says "SQL not allowed
    // statement". Both are SQL_PARSE_ERROR and both must reach the user verbatim.
    for (const message of [
      "SQL string could not be parsed: unsupported statement: VACUUM",
      "SQL not allowed statement: VACUUM",
    ]) {
      server = answerFor;
      const provider = await connected();
      server = () => failure(message, "SQL_PARSE_ERROR");

      await expect(provider.query("VACUUM")).rejects.toThrow(message);
      await provider.disconnect();
    }
  });
});

// ============================================================================
// Schema
// ============================================================================

// ============================================================================
// Monitoring
// ============================================================================

describe("LibSQLProvider monitoring", () => {
  test("names the server version and the SQLite it embeds", async () => {
    const provider = await connected();

    const overview = await provider.getOverview();

    expect(overview.version).toBe("sqld 0.24.33 (f8fb14f3 2026-08-11) (SQLite 3.47.0)");
    expect(overview.databaseSize).toBe("276 KB");
    expect(overview.tableCount).toBe(2);
    expect(overview.indexCount).toBe(1);
    expect(overview.maxConnections).toBe(0);
    await provider.disconnect();
  });

  test("shows the SQLite version alone on Turso Cloud, where /version does not exist", async () => {
    versionRoute = () => new Response(JSON.stringify({ error: 'route not found: ["version"]' }), { status: 404 });
    const provider = await connected();

    expect((await provider.getOverview()).version).toBe("SQLite 3.47.0");
    await provider.disconnect();
  });

  test("reports the integrity check and journal mode, and no invented cache ratio", async () => {
    const provider = await connected();

    const health = await provider.getHealth();

    expect(health.databaseSize).toBe("276 KB");
    expect(health.cacheHitRatio).toBe("N/A");
    expect(health.slowQueries.map((entry) => entry.query)).toEqual(["Integrity: OK", "Journal Mode: wal"]);
    expect(health.activeSessions).toEqual([]);
    await provider.disconnect();
  });

  test("counts zero deadlocks as a fact about the engine and measures nothing else", async () => {
    const provider = await connected();

    expect(await provider.getPerformanceMetrics()).toEqual({ deadlocks: 0 });
    expect(await provider.getSlowQueries()).toEqual([]);
    expect(await provider.getActiveSessions()).toEqual([]);
    await provider.disconnect();
  });

  test("splits measured pages between tables and their indexes", async () => {
    const provider = await connected();

    const stats = await provider.getTableStats();

    expect(stats[0]).toMatchObject({ tableName: "probe_customers", tableSizeBytes: 8192, indexSizeBytes: 4096 });
    expect(stats[1]).toMatchObject({ tableName: "probe_orders", rowCount: 2000, totalSizeBytes: 270336 });
    await provider.disconnect();
  });

  test("reports the user index with its columns and measured bytes", async () => {
    const provider = await connected();

    expect(await provider.getIndexStats()).toEqual([
      {
        schemaName: "main",
        tableName: "probe_customers",
        indexName: "idx_country",
        columns: ["country"],
        isUnique: true,
        isPrimary: false,
        indexSize: "4 KB",
        indexSizeBytes: 4096,
        scans: 0,
      },
    ]);
    await provider.disconnect();
  });

  test("reports the one database as storage, from its own page counters", async () => {
    const provider = await connected();

    expect(await provider.getStorageStats()).toEqual([{ name: "main", size: "276 KB", sizeBytes: 282624 }]);
    await provider.disconnect();
  });

  test("a token that expires mid-session fails every monitoring read and the integrity check as authentication", async () => {
    const provider = await connected();
    globalThis.fetch = (() =>
      Promise.resolve(
        new Response(JSON.stringify({ error: "JWT error: ExpiredSignature" }), { status: 401 }),
      )) as unknown as typeof fetch;

    const reads: [string, () => Promise<unknown>][] = [
      ["getHealth", () => provider.getHealth()],
      ["getOverview", () => provider.getOverview()],
      ["getTableStats", () => provider.getTableStats()],
      ["getIndexStats", () => provider.getIndexStats()],
      ["getStorageStats", () => provider.getStorageStats()],
      ["runMaintenance", () => provider.runMaintenance("check")],
    ];
    for (const [name, read] of reads) {
      const refusal = await read().catch((error: unknown) => error);
      expect([name, refusal instanceof AuthenticationError]).toEqual([name, true]);
      expect((refusal as Error).message).toContain("ExpiredSignature");
    }
  });
});

// ============================================================================
// Maintenance
// ============================================================================

describe("LibSQLProvider runMaintenance", () => {
  test("runs a bare REINDEX, and a targeted one against the named table", async () => {
    const provider = await connected();
    calls = [];

    expect(await provider.runMaintenance("reindex")).toMatchObject({ success: true });
    expect(await provider.runMaintenance("reindex", "probe_customers")).toMatchObject({ success: true });
    // #772: a container is ignored - the connection resolves names against its one
    // attached database, exactly as sqlite.ts does.
    expect(await provider.runMaintenance("reindex", "probe_customers", "main")).toMatchObject({ success: true });

    expect(sentStatements()).toEqual(["REINDEX", 'REINDEX "probe_customers"', 'REINDEX "probe_customers"']);
    await provider.disconnect();
  });

  test("reads the integrity check's ANSWER rather than its status", async () => {
    const provider = await connected();

    expect(await provider.runMaintenance("check")).toMatchObject({ success: true, message: "ok" });
    await provider.disconnect();
  });

  test("reports a corrupt database as a failed check, even though the statement succeeded", async () => {
    server = (sql) =>
      /integrity_check/.test(sql)
        ? result([["integrity_check", null]], [[text("*** in database main ***")]])
        : answerFor(sql);
    const provider = await connected();

    expect(await provider.runMaintenance("check")).toMatchObject({
      success: false,
      message: "*** in database main ***",
    });
    await provider.disconnect();
  });

  test("refuses VACUUM here rather than relaying the server's refusal", async () => {
    const provider = await connected();
    calls = [];

    const refused = provider.runMaintenance("vacuum");

    await expect(refused).rejects.toThrow(/do not accept VACUUM/);
    // Nothing was sent: the refusal is ours, so the user is not told about a
    // statement they never asked for.
    expect(sentStatements()).toEqual([]);
    await provider.disconnect();
  });

  test("refuses ANALYZE and OPTIMIZE for the same measured reason", async () => {
    const provider = await connected();

    await expect(provider.runMaintenance("analyze")).rejects.toThrow(/do not accept ANALYZE/);
    await expect(provider.runMaintenance("optimize")).rejects.toThrow(/do not accept OPTIMIZE/);
    await provider.disconnect();
  });
});

// ============================================================================
// Object surface (#789, task 13)
// ----------------------------------------------------------------------------
// The fixture below is a CATALOG rather than a set of canned answers, and that is
// the whole design. Every row was captured on 2026-09-11 from a live
// `ghcr.io/tursodatabase/libsql-server:v0.24.33` (sqld 0.24.33 `40a151bd`, SQLite
// 3.45.1) built by the DDL in `docs/providers/libsql.md`, in the engine's own
// order, and the fake server applies only the predicates the statement it receives
// actually spells. So dropping a predicate from the provider does not fall through
// to an unmatched regex and a canned row set: it widens the population the way it
// would against the real server, which is what makes the mutations below die for
// the right reason.
//
// The two facts that separate this engine from `sqlite.ts` are both measured and
// both drive assertions here:
//
// - sqld REFUSES `VACUUM`, `ANALYZE`, `PRAGMA query_only`, `ATTACH DATABASE` and
//   `CREATE TEMP TABLE`, but ACCEPTS `CREATE VIEW temp.<name>` - so a `temp` object
//   shadowing a `main` one is reachable and the schema restriction is load-bearing.
// - `CREATE FUNCTION ... LANGUAGE wasm` is refused by sqld's own parser
//   ("syntax error around L1:16: `FUNCTION`"), so NO routine kind is declared.
// ============================================================================

/** One row of `PRAGMA table_list`, verbatim and in the engine's own order. */
const TABLE_LIST_ROWS: readonly { schema: string; name: string; type: string }[] = [
  { schema: "main", name: "notes_content", type: "shadow" },
  { schema: "main", name: "legacy_ref", type: "table" },
  { schema: "main", name: "legacy", type: "table" },
  { schema: "main", name: "order_summary", type: "view" },
  { schema: "main", name: "sqlite_schema", type: "table" },
  { schema: "main", name: "badges", type: "table" },
  { schema: "main", name: "orders", type: "table" },
  { schema: "main", name: "shipments", type: "table" },
  { schema: "main", name: "regions", type: "table" },
  { schema: "main", name: "notes_config", type: "shadow" },
  { schema: "main", name: "sqlite_sequence", type: "table" },
  { schema: "main", name: "archive", type: "table" },
  { schema: "main", name: "notes_idx", type: "shadow" },
  { schema: "main", name: "customers", type: "table" },
  { schema: "main", name: "sqliteXledger", type: "table" },
  { schema: "main", name: "notes_docsize", type: "shadow" },
  { schema: "main", name: "notes", type: "virtual" },
  { schema: "main", name: "notes_data", type: "shadow" },
  { schema: "temp", name: "sqlite_temp_schema", type: "table" },
];

/**
 * The same catalog with a `temp` VIEW live, which sqld really does accept.
 *
 * `CREATE TEMP VIEW` is refused by sqld's parser and `CREATE VIEW temp.orders` is
 * not, measured. The row is here so the schema restriction can be mutated: without
 * `t.schema = ?` the listing answers `orders` TWICE, two objects with one path,
 * which is the uniqueness the tree addresses rows by.
 */
const TABLE_LIST_WITH_TEMP: readonly { schema: string; name: string; type: string }[] = [
  ...TABLE_LIST_ROWS,
  { schema: "temp", name: "orders", type: "view" },
];

/**
 * The same catalog plus the two names that separate the ENGINE's order from this process's.
 *
 * Measured on a live sqld 0.24.33: `ORDER BY t.name` answers `U+E000` before `U+1F600`,
 * because SQLite's BINARY collation is the UTF-8 BYTE order (`EE 80 80` below
 * `F0 9F 98 80`), while a JavaScript sort compares UTF-16 code units and puts the surrogate
 * `0xD83D` first. So a bounded read's MEMBERSHIP is the server's and the ORDER of the
 * answer is ours, and neither can be dropped without the other showing it.
 */
const TABLE_LIST_WITH_EXOTIC: readonly { schema: string; name: string; type: string }[] = [
  ...TABLE_LIST_ROWS,
  { schema: "main", name: "\uE000", type: "table" },
  { schema: "main", name: "\u{1F600}", type: "table" },
];

/**
 * One row of `sqlite_schema`, verbatim and in the engine's own order.
 *
 * RE-CAPTURED 2026-09-13 against a self-hosted sqld 0.24.33 (40a151bd 2025-12-19) started
 * from `docker/sqlite-init/02-libsql-object-fixture.sql`, which is where that DDL lives now:
 * it was a fenced block in `docs/providers/libsql.md` and nowhere else, so nobody could
 * re-run the measurement it stood for (D53, standing ruling 5i). The row ORDER is the
 * earlier capture's and is kept; what the re-capture adds is the `badges` TRIGGER below and
 * the definition text in `SQLITE_SCHEMA_SQL`.
 */
const SQLITE_SCHEMA_ROWS: readonly { type: string; name: string; tbl_name: string }[] = [
  { type: "table", name: "customers", tbl_name: "customers" },
  { type: "table", name: "sqlite_sequence", tbl_name: "sqlite_sequence" },
  { type: "table", name: "orders", tbl_name: "orders" },
  { type: "table", name: "regions", tbl_name: "regions" },
  { type: "table", name: "archive", tbl_name: "archive" },
  { type: "table", name: "sqliteXledger", tbl_name: "sqliteXledger" },
  { type: "table", name: "notes", tbl_name: "notes" },
  { type: "table", name: "notes_data", tbl_name: "notes_data" },
  { type: "table", name: "notes_idx", tbl_name: "notes_idx" },
  { type: "table", name: "notes_content", tbl_name: "notes_content" },
  { type: "table", name: "notes_docsize", tbl_name: "notes_docsize" },
  { type: "table", name: "notes_config", tbl_name: "notes_config" },
  { type: "view", name: "order_summary", tbl_name: "order_summary" },
  { type: "index", name: "idx_orders_customer", tbl_name: "orders" },
  { type: "index", name: "idx_orders_placed", tbl_name: "orders" },
  { type: "index", name: "idx_customers_name", tbl_name: "customers" },
  { type: "trigger", name: "orders_stamp", tbl_name: "orders" },
  { type: "trigger", name: "order_summary_guard", tbl_name: "order_summary" },
  // A TRIGGER whose name is ALSO a TABLE's, which sqld accepts (measured, and the file
  // engine accepts it too). It is here for the Source read: `SELECT sql FROM sqlite_schema
  // WHERE name = 'badges'` answers TWO rows and the table's comes first, so a read taking
  // its `type` from what the name matched rather than from the KIND hands a reader the
  // table's DDL under a trigger's address. Every other name in this catalog resolves to one
  // row whatever the type filter does.
  { type: "trigger", name: "badges", tbl_name: "badges" },
  { type: "table", name: "legacy", tbl_name: "legacy" },
  { type: "table", name: "legacy_ref", tbl_name: "legacy_ref" },
  { type: "table", name: "shipments", tbl_name: "shipments" },
  { type: "table", name: "badges", tbl_name: "badges" },
  // The implicit index behind `code TEXT UNIQUE` on a ROWID table, and it IS a row of
  // `sqlite_schema` rather than an invisible structure. Measured: a WITHOUT ROWID table
  // (`regions`) produces an autoindex that `pragma_index_list` publishes and
  // `sqlite_schema` does NOT, which is exactly the shape that hid this row from an
  // earlier fixture and made the reserved-name predicate look untestable here.
  { type: "index", name: "sqlite_autoindex_badges_1", tbl_name: "badges" },
];

/**
 * `sqlite_schema.sql` per row, keyed `"<type>/<name>"`, from the SAME live capture (#789).
 *
 * Beside the array rather than on it, so the rows above keep the comments that say what each
 * one is for and this file keeps one row per line. `sqliteSchemaSql()` is the only reader and
 * it THROWS by name for a row this map does not carry, so a row added to the catalog without
 * its definition text fails loudly instead of reading as a definition the server refused.
 *
 * NULL on exactly one entry, `index/sqlite_autoindex_badges_1`. That is the only shape in
 * which this engine answers a row with no definition, and the listings exclude every
 * `sqlite_`-prefixed name, so no path the tree produces can address it: the Source read has
 * no reachable refusal here, which docs/providers/libsql.md states as a CANNOT.
 *
 * Captured 2026-09-13 against a self-hosted sqld 0.24.33 (40a151bd 2025-12-19) started from
 * `docker/sqlite-init/02-libsql-object-fixture.sql`.
 */
const SQLITE_SCHEMA_SQL: Readonly<Record<string, string | null>> = {
  "table/customers":
    "CREATE TABLE customers (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, country TEXT DEFAULT 'TR')",
  "table/sqlite_sequence": "CREATE TABLE sqlite_sequence(name,seq)",
  "table/orders":
    "CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER NOT NULL REFERENCES customers,\n                     total REAL NOT NULL, tax REAL GENERATED ALWAYS AS (total * 0.2) VIRTUAL, placed_at TEXT)",
  "table/regions":
    "CREATE TABLE regions (region TEXT NOT NULL, year INTEGER NOT NULL, revenue REAL,\n                      PRIMARY KEY (region, year)) WITHOUT ROWID",
  "table/archive": "CREATE TABLE archive (id INTEGER PRIMARY KEY, body TEXT) STRICT",
  "table/sqliteXledger": "CREATE TABLE sqliteXledger (id INTEGER PRIMARY KEY, note TEXT)",
  "table/legacy": "CREATE TABLE legacy (note TEXT)",
  "table/legacy_ref": "CREATE TABLE legacy_ref (id INTEGER PRIMARY KEY, note TEXT REFERENCES legacy)",
  "table/shipments":
    "CREATE TABLE shipments (id INTEGER PRIMARY KEY, order_id INTEGER REFERENCES orders(id), carrier TEXT)",
  "table/badges": "CREATE TABLE badges (id INTEGER PRIMARY KEY, code TEXT UNIQUE, label TEXT)",
  "index/sqlite_autoindex_badges_1": null,
  "table/notes": "CREATE VIRTUAL TABLE notes USING fts5(title, body)",
  "table/notes_data": "CREATE TABLE 'notes_data'(id INTEGER PRIMARY KEY, block BLOB)",
  "table/notes_idx": "CREATE TABLE 'notes_idx'(segid, term, pgno, PRIMARY KEY(segid, term)) WITHOUT ROWID",
  "table/notes_content": "CREATE TABLE 'notes_content'(id INTEGER PRIMARY KEY, c0, c1)",
  "table/notes_docsize": "CREATE TABLE 'notes_docsize'(id INTEGER PRIMARY KEY, sz BLOB)",
  "table/notes_config": "CREATE TABLE 'notes_config'(k PRIMARY KEY, v) WITHOUT ROWID",
  "view/order_summary":
    "CREATE VIEW order_summary AS SELECT c.name, o.total FROM orders o JOIN customers c ON c.id = o.customer_id",
  "index/idx_orders_customer": "CREATE INDEX idx_orders_customer ON orders(customer_id)",
  "index/idx_orders_placed": "CREATE INDEX idx_orders_placed ON orders(date(placed_at))",
  "index/idx_customers_name": "CREATE UNIQUE INDEX idx_customers_name ON customers(name)",
  "trigger/orders_stamp":
    "CREATE TRIGGER orders_stamp AFTER INSERT ON orders\n  BEGIN UPDATE orders SET placed_at = datetime('now') WHERE id = NEW.id; END",
  "trigger/order_summary_guard":
    "CREATE TRIGGER order_summary_guard INSTEAD OF INSERT ON order_summary\n  BEGIN SELECT RAISE(ABORT, 'read only'); END",
  "trigger/badges": "CREATE TRIGGER badges AFTER INSERT ON badges BEGIN SELECT 1; END",
};

/** One row's stored definition, or a throw naming the row the capture forgot. */
function sqliteSchemaSql(type: string, name: string): string | null {
  const key = `${type}/${name}`;
  if (!Object.hasOwn(SQLITE_SCHEMA_SQL, key)) {
    throw new Error(`the sqlite_schema capture carries no sql for ${key}`);
  }
  return SQLITE_SCHEMA_SQL[key];
}

/**
 * The same, for a row an assertion expects to HAVE a definition.
 *
 * A NULL reaching a text assertion would be compared against a refusal and could only pass by
 * accident, so it raises naming the row instead. Exactly one row of this capture is NULL and
 * it has its own test.
 */
function definedSchemaSql(type: string, name: string): string {
  const definition = sqliteSchemaSql(type, name);
  if (definition === null) throw new Error(`${type}/${name} has a NULL definition and cannot be read back as text`);
  return definition;
}

/**
 * `pragma_table_xinfo(name, 'main')` per object, exactly as the engine publishes it:
 * name, declared type, `notnull`, default, `pk` rank and `hidden`.
 *
 * `hidden` is carried rather than pre-applied, because two separate rules read it and both
 * are mutable. `hidden = 1` is a virtual table module's OWN interface columns, which the
 * statement excludes; `hidden = 2` is a VIRTUAL generated column and `hidden = 3` a STORED
 * one, which `pragma_table_info` drops entirely and `pragma_table_xinfo` publishes.
 * Measured: `table_info('orders')` answers four columns where `table_xinfo` answers five.
 */
interface ColumnRow {
  name: string;
  type: string;
  notnull: number;
  dflt: string | null;
  pk: number;
  hidden: number;
}

function column(name: string, type: string, notnull: number, dflt: string | null, pk: number, hidden = 0): ColumnRow {
  return { name, type, notnull, dflt, pk, hidden };
}

const COLUMNS: Readonly<Record<string, ColumnRow[]>> = {
  orders: [
    column("id", "INTEGER", 0, null, 1),
    column("customer_id", "INTEGER", 1, null, 0),
    column("total", "REAL", 1, null, 0),
    // GENERATED ALWAYS AS (total * 0.2) VIRTUAL, so `hidden` is 2.
    column("tax", "REAL", 0, null, 0, 2),
    column("placed_at", "TEXT", 0, null, 0),
  ],
  // PRIMARY KEY (region, year): `pk` is a 1-based RANK, so `= 1` demotes `year`.
  regions: [
    column("region", "TEXT", 1, null, 1),
    column("year", "INTEGER", 1, null, 2),
    column("revenue", "REAL", 0, null, 0),
  ],
  customers: [
    column("id", "INTEGER", 0, null, 1),
    column("name", "TEXT", 1, null, 0),
    column("country", "TEXT", 0, "'TR'", 0),
  ],
  // The FTS5 table. Its declared columns carry the EMPTY STRING as their type, and the
  // module's own interface columns are `notes` and `rank` at `hidden = 1`.
  notes: [
    column("title", "", 0, null, 0),
    column("body", "", 0, null, 0),
    column("notes", "", 0, null, 0, 1),
    column("rank", "", 0, null, 0, 1),
  ],
  order_summary: [column("name", "TEXT", 0, null, 0), column("total", "REAL", 0, null, 0)],
  legacy: [column("note", "TEXT", 0, null, 0)],
  legacy_ref: [column("id", "INTEGER", 0, null, 1), column("note", "TEXT", 0, null, 0)],
  // #1029: one column per shape of catalog default. `dflt_value` is the expression AS
  // WRITTEN, identical to SQLite's `PRAGMA table_info` on every row. Served only through
  // `DEFAULTS_CATALOG`, so no FIXTURE-backed listing or count sees this table.
  column_defaults: [
    column("id", "INTEGER", 0, null, 1),
    column("def_null_string", "TEXT", 0, "'NULL'", 0),
    column("def_text", "TEXT", 0, "'abc'", 0),
    column("def_empty", "TEXT", 0, "''", 0),
    column("def_quote", "TEXT", 0, "'it''s'", 0),
    column("def_backslash", "TEXT", 0, "'a\\b'", 0),
    column("def_number", "INTEGER", 0, "42", 0),
    column("def_expression", "TEXT", 0, "CURRENT_TIMESTAMP", 0),
  ],
  archive: [column("id", "INTEGER", 0, null, 1), column("body", "TEXT", 0, null, 0)],
  sqliteXledger: [column("id", "INTEGER", 0, null, 1), column("note", "TEXT", 0, null, 0)],
  shipments: [
    column("id", "INTEGER", 0, null, 1),
    column("order_id", "INTEGER", 0, null, 0),
    column("carrier", "TEXT", 0, null, 0),
  ],
  // Two names whose UTF-8 byte order is the OPPOSITE of their UTF-16 code-unit order, so
  // the engine's cut and this process's sort can be told apart (#789).
  "\uE000": [column("x", "INTEGER", 0, null, 0)],
  "\u{1F600}": [column("x", "INTEGER", 0, null, 0)],
  badges: [
    column("id", "INTEGER", 0, null, 1),
    column("code", "TEXT", 0, null, 0),
    column("label", "TEXT", 0, null, 0),
  ],
};

/** `pragma_index_list(name, 'main')` per object, including the implicit ones. */
const INDEX_LIST: Readonly<Record<string, [string, number][]>> = {
  orders: [
    ["idx_orders_placed", 0],
    ["idx_orders_customer", 0],
  ],
  customers: [["idx_customers_name", 1]],
  // WITHOUT ROWID with a composite primary key, so the engine made an index nobody
  // declared and nobody can drop. The `sqlite_` predicate is what removes it.
  regions: [["sqlite_autoindex_regions_1", 1]],
  // `code TEXT UNIQUE` on a ROWID table. Same exclusion, and unlike the one above this
  // index is ALSO a row of `sqlite_schema`, so it reaches the Indexes folder too.
  badges: [["sqlite_autoindex_badges_1", 1]],
};

/** `pragma_index_info(name, 'main')`, in `seqno` order. A null name is an EXPRESSION key. */
const INDEX_COLUMNS: Readonly<Record<string, (string | null)[]>> = {
  idx_orders_customer: ["customer_id"],
  // CREATE INDEX idx_orders_placed ON orders(date(placed_at)): `name` is NULL, `cid` is -2.
  idx_orders_placed: [null],
  idx_customers_name: ["name"],
  sqlite_autoindex_regions_1: ["region", "year"],
  sqlite_autoindex_badges_1: ["code"],
};

/** `pragma_foreign_key_list(name, 'main')`. A null `to` means the parent's primary key. */
const FOREIGN_KEYS: Readonly<Record<string, [number, number, string, string, string | null][]>> = {
  // REFERENCES customers, with no column list.
  orders: [[0, 0, "customers", "customer_id", null]],
  // REFERENCES legacy, whose parent declares no primary key at all.
  legacy_ref: [[0, 0, "legacy", "note", null]],
  // REFERENCES orders(id), which NAMES its column - so nothing needs the parent's key.
  shipments: [[0, 0, "orders", "order_id", "id"]],
};

/** Every kind, and how many of it `main` holds. Counted by hand off the fixture DDL. */
const EXPECTED_COUNTS = { table: 10, view: 1, index: 3, trigger: 3 } as const;

/** True when the statement carries the reserved-name predicate, escaped or not. */
function reservedPredicate(sql: string): "escaped" | "wildcard" | "none" {
  if (/NOT LIKE 'sqlite\\_%' ESCAPE '\\'/.test(sql)) return "escaped";
  if (/NOT LIKE 'sqlite_%'/.test(sql)) return "wildcard";
  return "none";
}

/** The engine's own LIKE, for the one pattern these statements use. */
function keepsName(sql: string, name: string): boolean {
  const predicate = reservedPredicate(sql);
  if (predicate === "none") return true;
  // `_` is LIKE's single-character wildcard, so the UNESCAPED pattern also matches
  // `sqliteXledger`. That is the whole reason ESCAPE is in the statement.
  return predicate === "escaped" ? !name.startsWith("sqlite_") : !/^sqlite.{1}/.test(name);
}

/** The `type IN (...)` or `type = '...'` set the statement actually spells. */
function selectedTypes(sql: string): string[] | null {
  const list = /t\.type IN \(([^)]*)\)/.exec(sql);
  if (list) return [...list[1].matchAll(/'([^']*)'/g)].map((match) => match[1]);
  const single = /t\.type = '([^']*)'/.exec(sql);
  return single ? [single[1]] : null;
}

/**
 * `PRAGMA table_list` as the engine answers it, narrowed by the predicates the
 * statement spells and by nothing else.
 *
 * The schema filter applies only when `t.schema = ?` is present, so a provider that
 * stopped restricting to `main` sees `temp` here exactly as it would see it live.
 */
function tableListRows(sql: string, args: unknown[], rows: typeof TABLE_LIST_ROWS): string[][] {
  const types = selectedTypes(sql);
  const schema = /t\.schema = \?/.test(sql) ? args[0] : null;
  return rows
    .filter((row) => schema === null || row.schema === schema)
    .filter((row) => types === null || types.includes(row.type))
    .filter((row) => keepsName(sql, row.name))
    .map((row) => (/AS kind/.test(sql) ? [row.type === "view" ? "view" : "table"] : [row.name]));
}

/** `sqlite_schema` as the engine answers it, narrowed the same way. */
function sqliteSchemaRows(sql: string, rows: typeof SQLITE_SCHEMA_ROWS): typeof SQLITE_SCHEMA_ROWS {
  const list = /s\.type IN \(([^)]*)\)/.exec(sql);
  const single = /s\.type = '([^']*)'/.exec(sql);
  const types = list ? [...list[1].matchAll(/'([^']*)'/g)].map((match) => match[1]) : single ? [single[1]] : null;
  return rows.filter((row) => types === null || types.includes(row.type)).filter((row) => keepsName(sql, row.name));
}

/** A `{ kind, n }` result built the way the GROUP BY does. */
function groupedCounts(sql: string, args: unknown[], catalog: Catalog): Cell {
  const kinds = [
    ...tableListRows(sql, args, catalog.tableList).map((row) => row[0]),
    ...sqliteSchemaRows(sql, catalog.sqliteSchema).map((row) => row.type),
  ];
  const tally = new Map<string, number>();
  for (const kind of kinds) tally.set(kind, (tally.get(kind) ?? 0) + 1);
  return result(
    [
      ["kind", "TEXT"],
      ["n", null],
    ],
    [...tally].map(([kind, n]) => [text(kind), int(n)]),
  );
}

/** Which rows of a keyed fixture one statement asks for, or none at all. */
function lookup<T>(table: Readonly<Record<string, T[]>>, key: unknown): T[] {
  return table[String(key)] ?? [];
}

/** The two catalogs one object-surface test runs against. */
interface Catalog {
  tableList: typeof TABLE_LIST_ROWS;
  sqliteSchema: typeof SQLITE_SCHEMA_ROWS;
}

const FIXTURE: Catalog = { tableList: TABLE_LIST_ROWS, sqliteSchema: SQLITE_SCHEMA_ROWS };

/**
 * Two names in the order SQLite's BINARY collation puts them, which is the UTF-8 BYTE
 * order and NOT JavaScript's UTF-16 code-unit order. Measured on a live sqld 0.24.33.
 */
function byteOrder(left: string, right: string): number {
  return Buffer.compare(Buffer.from(left, "utf8"), Buffer.from(right, "utf8"));
}

/** The name a flat-surface statement embeds as a LITERAL rather than binding. */
function literalArg(sql: string): string | undefined {
  return /pragma_\w+\('([^']*)'\)/.exec(sql)?.[1];
}

/**
 * The parent key column an unnamed foreign key resolves to, as the correlated subquery in
 * `OBJECT_FOREIGN_KEYS_SQL` resolves it: the parent's key column whose 1-based `pk` rank is
 * `seq + 1`. `legacy` declares no primary key at all, so it answers null.
 */
function parentKey(parent: string, seq: number): string | null {
  const keys = lookup(COLUMNS, parent)
    .filter((row) => row.pk > 0)
    .sort((left, right) => left.pk - right.pk);
  return keys[seq]?.name ?? null;
}

/**
 * The five statements of the BULK column read, answered off the same fixture the single
 * read answers off (#789).
 *
 * Written as one arm rather than five, because all five carry the same target CTE and the
 * point of the fake is that they cut the SAME set: the target is derived here once and the
 * four detail statements are answered against it. What this arm cannot see is a rewrite of
 * the SQL itself (standing ruling 5b), which is why the whole method was also run against a
 * live `ghcr.io/tursodatabase/libsql-server:v0.24.33` - see docs/providers/libsql.md.
 */
function bulkRows(sql: string, args: unknown[], catalog: Catalog): Cell | null {
  if (!/WITH described AS/.test(sql)) return null;
  const types = selectedTypes(sql) ?? [];
  // `?` are positional and the type list is a LITERAL list, the same spelling the folder's
  // own listing carries, so the only binds before the LIMIT are the schema.
  const limit = /LIMIT \?/.test(sql) ? Number(args[1]) : undefined;
  const ordered = catalog.tableList
    .filter((row) => row.schema === args[0])
    .filter((row) => types.includes(row.type))
    .filter((row) => keepsName(sql, row.name))
    .map((row) => row.name);
  // `ORDER BY t.name` is APPLIED here rather than assumed, and under the engine's own
  // collation. Measured against a live sqld 0.24.33: the sort is BINARY, the UTF-8 BYTE
  // order, so `U+E000` (EE 80 80) comes back before `U+1F600` (F0 9F 98 80) - the opposite
  // of a JavaScript sort, which compares UTF-16 code units and puts the surrogate 0xD83D
  // first. Without applying it here, dropping the clause from the statement could not be
  // mutated, and the two orders could not be told apart at all.
  const targets = (/ORDER BY t\.name/.test(sql) ? [...ordered].sort(byteOrder) : ordered).slice(0, limit);

  const cols = (names: string[]): [string, string | null][] => names.map((name) => [name, null]);
  const rows = (names: string[], values: (Cell | null)[][]): Cell =>
    result(
      cols(names),
      values.map((row) => row.map((cell) => cell ?? { type: "null" })),
    );

  if (/JOIN pragma_table_xinfo/.test(sql)) {
    return rows(
      ["object_name", "name", "type", "notnull", "dflt_value", "pk"],
      targets.flatMap((object) =>
        lookup(COLUMNS, object)
          .filter((row) => !/hidden <> 1/.test(sql) || row.hidden !== 1)
          .map((row) => [
            text(object),
            text(row.name),
            text(row.type),
            int(row.notnull),
            row.dflt === null ? null : text(row.dflt),
            int(row.pk),
          ]),
      ),
    );
  }
  if (/JOIN pragma_index_info/.test(sql)) {
    return rows(
      ["object_name", "index_name", "name"],
      targets.flatMap((object) =>
        lookup(INDEX_LIST, object)
          .filter(([name]) => keepsName(sql, name))
          .flatMap(([index]) =>
            lookup(INDEX_COLUMNS, index).map((column) => [
              text(object),
              text(index),
              column === null ? null : text(column),
            ]),
          ),
      ),
    );
  }
  if (/JOIN pragma_index_list/.test(sql)) {
    return rows(
      ["object_name", "name", "unique"],
      targets.flatMap((object) =>
        (/ORDER BY d\.object_name, i\.name/.test(sql)
          ? [...lookup(INDEX_LIST, object)].sort(([left], [right]) => byteOrder(left, right))
          : lookup(INDEX_LIST, object)
        )
          .filter(([name]) => keepsName(sql, name))
          .map(([name, unique]) => [text(object), text(name), int(unique)]),
      ),
    );
  }
  if (/JOIN pragma_foreign_key_list/.test(sql)) {
    return rows(
      ["object_name", "id", "seq", "table", "from", "to", "parent_key"],
      targets.flatMap((object) =>
        lookup(FOREIGN_KEYS, object).map(([id, seq, parent, from, to]) => {
          const key = parentKey(parent, seq);
          return [
            text(object),
            int(id),
            int(seq),
            text(parent),
            text(from),
            to === null ? null : text(to),
            key === null ? null : text(key),
          ];
        }),
      ),
    );
  }
  return rows(
    ["object_name"],
    targets.map((object) => [text(object)]),
  );
}

/**
 * The FLAT reading, answered off the SAME fixture the object reading answers off (#789).
 *
 * Without it `getSchema()` here answers the two `probe_*` tables of the recorded
 * connection payloads, which share no name with any object this fixture publishes - so the
 * conformance helper's join guard could only report the two readings as two populations.
 * A fixture that puts one object in front of both surfaces is the whole point of that
 * guard, and answering the flat read from a different catalog is the thing it exists to
 * catch.
 *
 * Two differences from the object reading are the ENGINE's and are kept rather than
 * smoothed away: the flat read is over `sqlite_master`, so it sees the FTS5 shadow tables
 * as ordinary tables, and its `NOT LIKE 'sqlite_%'` carries no `ESCAPE`, so it also drops
 * `sqliteXledger`, a name a user can really have. Both are why the object surface reads
 * `PRAGMA table_list` and escapes its wildcard.
 */
function flatRows(sql: string, catalog: Catalog): Cell | null {
  if (/FROM sqlite_master/.test(sql) && /type = 'table'/.test(sql) && /SELECT name/.test(sql)) {
    const names = catalog.sqliteSchema
      .filter((row) => row.type === "table")
      .map((row) => row.name)
      .filter((name) => !/^sqlite.{1}/.test(name))
      .sort();
    return result(
      [["name", "TEXT"]],
      names.map((name) => [text(name)]),
    );
  }
  if (/^SELECT COUNT\(\*\) AS row_count/.test(sql.trim())) {
    return result([["row_count", null]], [[int(0)]]);
  }
  const name = literalArg(sql);
  if (name === undefined) return null;
  if (/pragma_table_info/.test(sql)) {
    return result(
      [
        ["cid", null],
        ["name", null],
        ["type", null],
        ["notnull", null],
        ["dflt_value", null],
        ["pk", null],
      ],
      lookup(COLUMNS, name)
        // `table_info` and not `table_xinfo`: the flat surface loses a generated column,
        // which is the defect the object surface's `table_xinfo` does not have.
        .filter((row) => row.hidden === 0)
        .map((row, cid) => [
          int(cid),
          text(row.name),
          text(row.type),
          int(row.notnull),
          row.dflt === null ? { type: "null" } : text(row.dflt),
          int(row.pk),
        ]),
    );
  }
  if (/pragma_index_list/.test(sql)) {
    return result(
      [
        ["seq", null],
        ["name", null],
        ["unique", null],
        ["origin", null],
      ],
      lookup(INDEX_LIST, name).map(([index, unique], seq) => [int(seq), text(index), int(unique), text("c")]),
    );
  }
  if (/pragma_index_info/.test(sql)) {
    return result(
      [
        ["seqno", null],
        ["cid", null],
        ["name", null],
      ],
      lookup(INDEX_COLUMNS, name).map((column, seqno) => [
        int(seqno),
        int(seqno),
        column === null ? { type: "null" } : text(column),
      ]),
    );
  }
  if (/pragma_foreign_key_list/.test(sql)) {
    return result(
      [
        ["id", null],
        ["seq", null],
        ["table", null],
        ["from", null],
        ["to", null],
      ],
      lookup(FOREIGN_KEYS, name).map(([id, seq, parent, from, to]) => [
        int(id),
        int(seq),
        text(parent),
        text(from),
        to === null ? { type: "null" } : text(to),
      ]),
    );
  }
  return null;
}

/** The catalog every object-surface test runs against. */
function objectServer(catalog: Catalog = FIXTURE): Server {
  return (sql, args) => {
    const bulk = bulkRows(sql, args, catalog);
    if (bulk !== null) return bulk;
    if (/GROUP BY kind/.test(sql)) return groupedCounts(sql, args, catalog);
    if (/pragma_table_list/.test(sql)) {
      return result(
        [["name", null]],
        tableListRows(sql, args, catalog.tableList).map((row) => [text(row[0])]),
      );
    }
    // BEFORE the generic `sqlite_schema` branch, because the Source read selects from the
    // same catalog and a listing-shaped answer would win. The predicates are read off the
    // STATEMENT and bound by the order they appear in it, so dropping either one widens the
    // population here exactly as it would against the real server: with no `s.type = ?` the
    // name `badges` answers the TABLE row as well as the trigger's, and the table's comes
    // first, which is the live behaviour measured on sqld 0.24.33.
    if (/s\.sql AS sql/.test(sql)) {
      const bound = Object.fromEntries(
        [...sql.matchAll(/s\.(type|name) = \?/g)].map((match, index) => [match[1], String(args[index])]),
      );
      const selected = catalog.sqliteSchema.filter(
        (row) =>
          (bound.type === undefined || row.type === bound.type) &&
          (bound.name === undefined || row.name === bound.name),
      );
      return result(
        [["sql", "TEXT"]],
        selected.map((row) => {
          const definition = sqliteSchemaSql(row.type, row.name);
          return [definition === null ? { type: "null" } : text(definition)];
        }),
      );
    }
    if (/FROM sqlite_schema AS s/.test(sql)) {
      const selected = sqliteSchemaRows(sql, catalog.sqliteSchema);
      if (/s\.tbl_name AS parent/.test(sql)) {
        return result(
          [
            ["name", "TEXT"],
            ["parent", "TEXT"],
          ],
          selected.map((row) => [text(row.name), text(row.tbl_name)]),
        );
      }
      return result(
        [["name", "TEXT"]],
        selected.map((row) => [text(row.name)]),
      );
    }
    if (/pragma_foreign_key_list/.test(sql)) {
      // `args[1]`, not `args[0]`: the statement's first placeholder belongs to the
      // parent-key subquery, which SQLite numbers first because the select list is written
      // before the FROM. A fake reading `args[0]` would answer the SCHEMA's foreign keys,
      // which is nothing, and every foreign key assertion would quietly go vacuous.
      const object = args[1];
      return result(
        [
          ["id", null],
          ["seq", null],
          ["table", null],
          ["from", null],
          ["to", null],
          ["parent_key", null],
        ],
        lookup(FOREIGN_KEYS, object).map(([id, seq, parent, from, to]) => {
          const key = parentKey(parent, seq);
          return [
            int(id),
            int(seq),
            text(parent),
            text(from),
            to === null ? { type: "null" } : text(to),
            key === null ? { type: "null" } : text(key),
          ];
        }),
      );
    }
    // BEFORE the `pragma_table_x?info` branch, and that order is load-bearing: the foreign
    // key statement resolves the parent's key with a correlated `pragma_table_info`, so a
    // column-shaped answer would win and every foreign key would read as empty.
    if (/pragma_table_x?info/.test(sql)) {
      // `table_info` publishes only the ordinary columns and `table_xinfo` publishes the
      // hidden ones too; then the statement's own `hidden <> 1` narrows what it asked for.
      const published = lookup(COLUMNS, args[0]).filter((row) => /pragma_table_xinfo/.test(sql) || row.hidden === 0);
      const selected = published.filter((row) => !/hidden <> 1/.test(sql) || row.hidden !== 1);
      if (/pk > 0/.test(sql)) {
        const keys = selected.filter((row) => row.pk > 0).sort((left, right) => left.pk - right.pk);
        return result(
          [["name", null]],
          keys.map((row) => [text(row.name)]),
        );
      }
      return result(
        [
          ["name", null],
          ["type", null],
          ["notnull", null],
          ["dflt_value", null],
          ["pk", null],
        ],
        selected.map((row) => [
          text(row.name),
          text(row.type),
          int(row.notnull),
          row.dflt === null ? { type: "null" } : text(row.dflt),
          int(row.pk),
        ]),
      );
    }
    if (/pragma_index_list/.test(sql)) {
      // `INDEX_LIST` is captured in the ENGINE's own order, which is reverse creation
      // order, and the statement's `ORDER BY name` is applied here rather than assumed:
      // dropping it from the statement has to change what this fake answers, or the clause
      // that keeps the single and the bulk read agreeing could not be mutated (#789).
      const listed = lookup(INDEX_LIST, args[0]).filter(([name]) => keepsName(sql, name));
      const ordered = /ORDER BY name/.test(sql)
        ? [...listed].sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))
        : listed;
      return result(
        [
          ["name", null],
          ["unique", null],
        ],
        ordered.map(([name, unique]) => [text(name), int(unique)]),
      );
    }
    if (/pragma_index_info/.test(sql)) {
      return result(
        [["name", null]],
        lookup(INDEX_COLUMNS, args[0]).map((name) => [name === null ? { type: "null" } : text(name)]),
      );
    }
    const flat = flatRows(sql, catalog);
    if (flat !== null) return flat;
    return answerFor(sql);
  };
}

/** A connected provider whose server answers the object catalog above. */
async function connectedWithObjects(catalog: Catalog = FIXTURE): Promise<LibSQLProvider> {
  server = objectServer(catalog);
  const provider = new LibSQLProvider(connection());
  await provider.connect();
  calls = [];
  return provider;
}

describe("LibSQLProvider object surface (#789)", () => {
  let objects: LibSQLProvider;

  afterEach(async () => {
    if (objects?.isConnected()) await objects.disconnect();
  });

  test("declares the four kinds libSQL has, at zero container levels, and no routine kind", async () => {
    objects = await connectedWithObjects();
    const capabilities = objects.getCapabilities();
    const kinds = capabilities.objectKinds ?? [];

    expect(kinds.map((kind) => kind.id)).toEqual(["table", "view", "index", "trigger"]);
    expect(kinds.find((kind) => kind.id === "table")?.role).toBe("relation");
    expect(kinds.find((kind) => kind.id === "table")?.acceptsRowWrites).toBe(true);
    expect(kinds.find((kind) => kind.id === "view")?.role).toBe("relation");
    expect(kinds.find((kind) => kind.id === "view")?.acceptsRowWrites).toBeUndefined();
    expect(kinds.find((kind) => kind.id === "index")?.role).toBe("config");
    expect(kinds.find((kind) => kind.id === "trigger")?.role).toBe("attached");
    expect(kinds.find((kind) => kind.id === "trigger")?.attachedTo).toBe("table");
    // NO `function` kind, which the brief made conditional on the server accepting
    // `CREATE FUNCTION ... LANGUAGE wasm`. Measured on the image `database-compose.yml`
    // pins: sqld's own parser refuses it ("syntax error around L1:16: `FUNCTION`"),
    // `libsql_wasm_func_table` does not exist and `sqld --help` carries no wasm flag.
    // A declared kind that always counts zero is a folder for something the engine
    // cannot do; an undeclared one is the honest answer.
    for (const absent of ["function", "procedure", "routine", "sequence", "package", "event"]) {
      expect(kinds.find((kind) => kind.id === absent)).toBeUndefined();
    }
    expect(capabilities.containerLevels).toEqual([]);
    expect(containerDepth(capabilities)).toBe(0);
  });

  test("satisfies the object-surface conformance contract", async () => {
    objects = await connectedWithObjects();

    await assertObjectSurface(objects, {
      containers: [],
      kinds: { ...EXPECTED_COUNTS },
      sampleObject: { path: ["orders"], kind: "table" },
      // No container level, so the authored path is the bare name. `emptyKinds` is absent
      // because every source-bearing kind this engine declares is counted above zero.
      absentSource: { path: ["no_such_table"], kind: "table" },
    });
  });

  test("listContainers answers an empty array, which is an answer and not a refusal", async () => {
    objects = await connectedWithObjects();

    await expect(objects.listContainers()).resolves.toEqual([]);
    // An answer, not a declaration read: it is not answerable off a provider that
    // never connected, and it costs no round trip when it is.
    expect(calls).toEqual([]);
    await expect(new LibSQLProvider(connection()).listContainers()).rejects.toThrow();
  });

  // --------------------------------------------------------------------------
  // What the counts and the listings enumerate
  // --------------------------------------------------------------------------

  test("a shadow table is not an object, and a virtual table is", async () => {
    objects = await connectedWithObjects();

    // The control, measured live: `sqlite_schema` types an FTS5 table's five shadow
    // tables `table`, so a naive scan answers 16 where the engine holds 10.
    const naive = SQLITE_SCHEMA_ROWS.filter((row) => row.type === "table");
    expect(naive).toHaveLength(16);
    expect(naive.map((row) => row.name)).toContain("notes_data");

    const tables = await objects.listObjects([], "table");

    expect(tables.map((table) => table.name)).toEqual([
      "archive",
      "badges",
      "customers",
      "legacy",
      "legacy_ref",
      "notes",
      "orders",
      "regions",
      "shipments",
      "sqliteXledger",
    ]);
    expect((await objects.countObjects([])).table).toEqual({ count: EXPECTED_COUNTS.table });
  });

  test("every value PRAGMA table_list can answer has a rule, and the fixture holds all four", async () => {
    // Standing ruling 5a: the vocabulary is SQLite's documented four values, not a
    // `SELECT DISTINCT` over whatever this fixture happens to hold. The fixture is
    // built to contain all four so the rule is exercised rather than asserted, and the
    // live catalog this was captured from answers exactly these four.
    objects = await connectedWithObjects();

    expect([...new Set(TABLE_LIST_ROWS.map((row) => row.type))].sort()).toEqual(["shadow", "table", "view", "virtual"]);

    const names = (await objects.listObjects([], "table")).map((object) => object.name);
    const views = (await objects.listObjects([], "view")).map((object) => object.name);

    // `virtual` is a table, `shadow` is nothing, `view` is a view, `table` is a table.
    expect(names).toContain("notes");
    expect(names).toContain("orders");
    expect(names.filter((name) => name.startsWith("notes_"))).toEqual([]);
    expect(views).toEqual(["order_summary"]);
  });

  test("the names libSQL reserves for itself are not objects, and ESCAPE is what makes that true", async () => {
    objects = await connectedWithObjects();

    const tables = (await objects.listObjects([], "table")).map((object) => object.name);

    // Present: a user CAN create this name, and the unescaped `LIKE 'sqlite_%'` would
    // swallow it because `_` is LIKE's single-character wildcard.
    expect(tables).toContain("sqliteXledger");
    // Absent: `sqlite_sequence` appears the moment a table declares AUTOINCREMENT, and
    // `sqlite_schema` is a row of `PRAGMA table_list` on every database there is.
    expect(tables).not.toContain("sqlite_sequence");
    expect(tables).not.toContain("sqlite_schema");
  });

  test("an index the engine created for itself is not an object, and it IS a row of sqlite_schema", async () => {
    // The reserved-name predicate on the INDEX listing, pinned by BEHAVIOUR.
    //
    // An earlier version of this suite claimed it could not be: the fixture's only
    // implicit index belonged to `regions`, a WITHOUT ROWID table, whose autoindex
    // `pragma_index_list` publishes and `sqlite_schema` does NOT. That one shape is the
    // exception. Measured on sqld 0.24.33, `code TEXT UNIQUE` on an ordinary ROWID table
    // puts `sqlite_autoindex_badges_1` straight into `sqlite_schema` as an `index` row,
    // so the predicate removes a real row here and a test can see it.
    objects = await connectedWithObjects();

    // The control, and the non-vacuity guard: the catalog really does hold the row this
    // test exists to exclude, so the assertions below are not about an empty population.
    const catalogIndexes = SQLITE_SCHEMA_ROWS.filter((row) => row.type === "index");
    expect(catalogIndexes.map((row) => row.name)).toContain("sqlite_autoindex_badges_1");

    const listed = await objects.listObjects([], "index");
    const counts = await objects.countObjects([]);

    expect(listed.map((index) => index.name)).toEqual([
      "idx_customers_name",
      "idx_orders_customer",
      "idx_orders_placed",
    ]);
    // The badge counts what the folder lists, one row fewer than the catalog holds.
    expect(counts.index).toEqual({ count: catalogIndexes.length - 1 });
    // And the same predicate applies inside the object's own detail, so the two surfaces
    // cannot disagree about what an index is.
    expect(await objects.describeObject(["badges"], "table")).toMatchObject({ indexes: [] });
  });

  test("a temp object shadowing a main one never reaches the tree, which sqld really can produce", async () => {
    // sqld refuses `CREATE TEMP VIEW` and accepts `CREATE VIEW temp.orders`, measured.
    // Without `t.schema = ?` the listing then answers `orders` twice under one path.
    objects = await connectedWithObjects({ ...FIXTURE, tableList: TABLE_LIST_WITH_TEMP });

    const views = await objects.listObjects([], "view");
    const tables = await objects.listObjects([], "table");

    expect(views.map((view) => view.path)).toEqual([["order_summary"]]);
    expect(tables.filter((table) => table.name === "orders")).toHaveLength(1);
    // And the COUNT is restricted by the same predicate, or the badge says 2 while the
    // folder holds 1 - a badge disagreeing with its own folder (standing ruling 5f).
    expect((await objects.countObjects([])).view).toEqual({ count: 1 });
    // And the bind is `main`, on the statement rather than in it.
    expect(sentCalls().every((call) => !/pragma_table_list/.test(call.sql) || call.args[0] === "main")).toBe(true);
  });

  test("the listing contains exactly what the count counted, for every declared kind", async () => {
    // Standing ruling 5f. A badge comes from `countObjects` and a folder from
    // `listObjects`, and the two read different statements, so nothing but a test
    // keeps them enumerating the same population.
    objects = await connectedWithObjects();
    const counts = await objects.countObjects([]);

    for (const [kind, expected] of Object.entries(EXPECTED_COUNTS)) {
      const listed = await objects.listObjects([], kind);
      expect(counts[kind]).toEqual({ count: expected });
      expect(listed).toHaveLength(expected);
    }
  });

  test("a trigger nests under the object it fires on, view or table", async () => {
    objects = await connectedWithObjects();

    const triggers = await objects.listObjects([], "trigger");

    // `attachedTo: "table"` names the kind a trigger usually hangs off; sqld accepts an
    // INSTEAD OF trigger on a VIEW and `sqlite_schema.tbl_name` then names the view.
    // Both the count and the listing carry it rather than one of them dropping it.
    expect(triggers.map((trigger) => trigger.path)).toEqual([
      // The name collision the Source read needs: sqld accepts a trigger whose name is
      // also a table's, which no other kind here can be.
      ["badges", "badges"],
      ["order_summary", "order_summary_guard"],
      ["orders", "orders_stamp"],
    ]);
    expect(triggers.every((trigger) => trigger.kind === "trigger")).toBe(true);
  });

  test("a listing is sorted by path here, which the engine's own order is not", async () => {
    objects = await connectedWithObjects();

    // Measured: `PRAGMA table_list` answers in PAGE order rather than name order. The exact
    // sequence is a property of one build - the same DDL replayed into a fresh container
    // produced a different one - so what is pinned here is that the fixture's captured
    // order is NOT sorted, which is what makes the assertion below non-vacuous.
    const catalogOrder = TABLE_LIST_ROWS.filter((row) => row.type === "table").map((row) => row.name);
    expect(catalogOrder).not.toEqual([...catalogOrder].sort());

    const tables = await objects.listObjects([], "table");

    expect(tables.map((table) => table.name)).toEqual([...tables.map((table) => table.name)].sort());
    expect(tables[0]?.name).toBe("archive");
  });

  // --------------------------------------------------------------------------
  // describeObject
  // --------------------------------------------------------------------------

  test("describes a table with its generated column and its composite primary key", async () => {
    objects = await connectedWithObjects();

    const orders = await objects.describeObject(["orders"], "table");
    const regions = await objects.describeObject(["regions"], "table");

    expect(orders.path).toEqual(["orders"]);
    // `tax` is GENERATED ALWAYS AS ... VIRTUAL. `pragma_table_info` drops it and
    // `table_xinfo` publishes it, which is why the object surface reads the latter.
    expect(orders.columns.map((column) => column.name)).toEqual(["id", "customer_id", "total", "tax", "placed_at"]);
    expect(orders.columns.find((column) => column.name === "customer_id")?.nullable).toBe(false);
    expect(orders.columns.find((column) => column.name === "placed_at")?.nullable).toBe(true);
    // `pk` is a 1-based RANK, not a flag: `= 1` would report `year` as ordinary.
    expect(regions.columns.filter((column) => column.isPrimary).map((column) => column.name)).toEqual([
      "region",
      "year",
    ]);
  });

  test("a default value is the value the column defaults to, and an absent one is absent", async () => {
    objects = await connectedWithObjects();

    const customers = await objects.describeObject(["customers"], "table");

    // The catalog reports `'TR'`; the value is `TR`, and the text is kept as the expression (#1029).
    const country = customers.columns.find((column) => column.name === "country");
    expect(country?.defaultValue).toBe("TR");
    expect(country?.defaultExpression).toBe("'TR'");
    const name = customers.columns.find((column) => column.name === "name");
    expect(name?.defaultValue).toBeUndefined();
    expect(name?.defaultExpression).toBeUndefined();
  });

  test("describes a view, which has columns and neither indexes nor foreign keys", async () => {
    objects = await connectedWithObjects();

    const detail = await objects.describeObject(["order_summary"], "view");

    expect(detail.columns.map((column) => column.name)).toEqual(["name", "total"]);
    expect(detail.indexes).toEqual([]);
    expect(detail.foreignKeys).toEqual([]);
  });

  test("a virtual table describes its declared columns, with the type the engine actually gave", async () => {
    objects = await connectedWithObjects();

    const detail = await objects.describeObject(["notes"], "table");

    expect(detail.columns.map((column) => column.name)).toEqual(["title", "body"]);
    // The EMPTY STRING, which is what an FTS5 column's declared type is. `getSchema()`
    // writes "TEXT" there, which is a guess about affinity rather than an answer.
    expect(detail.columns.map((column) => column.type)).toEqual(["", ""]);
  });

  test("a table's indexes exclude the implicit ones, and an expression key has no column name", async () => {
    objects = await connectedWithObjects();

    const orders = await objects.describeObject(["orders"], "table");
    const regions = await objects.describeObject(["regions"], "table");

    // BY NAME, which is this read's own `ORDER BY` and not `pragma_index_list`'s: measured
    // on sqld 0.24.33, that pragma answers in reverse creation order, so it would put
    // `idx_orders_placed` first. The bulk read has to order by the object to group its
    // rows, and two orders over one table's indexes is a disagreement between the two
    // surfaces (#789).
    expect(orders.indexes).toEqual([
      { name: "idx_orders_customer", columns: ["customer_id"], unique: false },
      { name: "idx_orders_placed", columns: [], unique: false },
    ]);
    // `sqlite_autoindex_regions_1` serves the composite primary key of a WITHOUT ROWID
    // table. Nobody declared it and nobody can drop it, and the Indexes folder excludes
    // it by the same predicate, so the two surfaces cannot disagree.
    expect(regions.indexes).toEqual([]);
  });

  test("a composite index lists every key column, in seqno order, on the single and the bulk read", async () => {
    objects = await connectedWithObjects();
    const base = objectServer();
    // CREATE INDEX idx_orders_customer ON orders(customer_id, placed_at): two rows of one index.
    server = (sql, args) => {
      if (/JOIN pragma_index_info/.test(sql)) {
        const answer = base(sql, args) as { response: { result: { rows: (Cell | null)[][] } } };
        const rows = answer.response.result.rows;
        const at = rows.findIndex((row) => (row[1] as Cell).value === "idx_orders_customer");
        rows.splice(at + 1, 0, [text("orders"), text("idx_orders_customer"), text("placed_at")]);
        return answer as Cell;
      }
      if (/pragma_index_info/.test(sql) && args[0] === "idx_orders_customer") {
        return result([["name", null]], [[text("customer_id")], [text("placed_at")]]);
      }
      return base(sql, args);
    };
    const expected = [
      { name: "idx_orders_customer", columns: ["customer_id", "placed_at"], unique: false },
      { name: "idx_orders_placed", columns: [], unique: false },
    ];

    expect((await objects.describeObject(["orders"], "table")).indexes).toEqual(expected);
    const bulk = await objects.describeObjects([], "table");
    expect(bulk.details.find((detail) => detail.path.join(".") === "orders")?.indexes).toEqual(expected);
  });

  test("a foreign key that names no column resolves to the parent's primary key", async () => {
    objects = await connectedWithObjects();

    const detail = await objects.describeObject(["orders"], "table");

    // `REFERENCES customers` answers `to = NULL`, which SQLite reads as the parent's
    // PRIMARY KEY. `ForeignKeySchema.referencedColumn` is a string, so the alternative
    // is a null in a typed string field.
    expect(detail.foreignKeys).toEqual([
      { columnName: "customer_id", referencedTable: "customers", referencedColumn: "id" },
    ]);
  });

  test("a foreign key whose parent has no primary key carries no referenced column", async () => {
    objects = await connectedWithObjects();

    const detail = await objects.describeObject(["legacy_ref"], "table");

    // A parent with no primary key is a schema SQLite accepts and rejects only on
    // INSERT, so there is nothing to name and the field is empty rather than invented.
    expect(detail.foreignKeys).toEqual([{ columnName: "note", referencedTable: "legacy", referencedColumn: "" }]);
  });

  test("no foreign key costs a parent read of its own, whether it names its column or not", async () => {
    // The other half of the resolution rule, and the half a fixture of all-implicit keys
    // cannot see: `REFERENCES orders(id)` answers `to = 'id'`, so nothing has to be
    // resolved at all. Neither half costs a request any more - the parent's key is a
    // correlated subquery inside the foreign key statement (#789), which is what lets the
    // bulk read describe a whole folder without a statement per distinct parent.
    objects = await connectedWithObjects();

    const detail = await objects.describeObject(["shipments"], "table");

    expect(detail.foreignKeys).toEqual([{ columnName: "order_id", referencedTable: "orders", referencedColumn: "id" }]);
    // `shipments` declares no index either, so the second batch is empty and costs no
    // The pragma matched is the LAST `FROM pragma_*` in each statement, because the foreign
    // key statement also names `pragma_table_info` inside its parent-key subquery, and the
    // subquery is written before the outer FROM.
    expect(sentCalls().map((call) => [...call.sql.matchAll(/FROM pragma_(\w+)/g)].pop()?.[1])).toEqual([
      "table_xinfo",
      "index_list",
      "foreign_key_list",
    ]);
  });

  test("hasColumns is declared on the two relation kinds, and describeObject backs the declaration", async () => {
    // The declaration is a CLIENT GATE (#789): the tree draws a twisty only where a kind
    // declares `hasColumns`, so a kind declaring it whose describe answers nothing opens
    // on nothing, and a kind answering columns without declaring it hides them behind a
    // leaf row. Both directions are asserted against this engine's own answer rather
    // than against the role, which happens to coincide here and does not elsewhere:
    // `describeLibSQLObject` in src/lib/db/providers/sql/libsql/objects.ts gates on
    // `spec.role !== "relation"`, while a MariaDB sequence is declared `config` and still
    // has columns.
    objects = await connectedWithObjects();
    const kinds = objects.getCapabilities().objectKinds ?? [];

    expect(kinds.filter((kind) => kindHasColumns(kind)).map((kind) => kind.id)).toEqual(["table", "view"]);
    // ABSENT rather than `false`, which is what the field's docblock asks of a provider
    // that has no columns for a kind: absent and false read the same and one writer is
    // enough.
    expect(kinds.find((kind) => kind.id === "index")?.hasColumns).toBeUndefined();
    expect(kinds.find((kind) => kind.id === "trigger")?.hasColumns).toBeUndefined();

    const view = await objects.describeObject(["order_summary"], "view");

    expect(view.columns.length).toBeGreaterThan(0);
    for (const column of view.columns) {
      expect(typeof column.name).toBe("string");
      expect(column.name.trim()).not.toBe("");
      expect(typeof column.type).toBe("string");
      expect(column.type.trim()).not.toBe("");
    }
  });

  test("a kind that declares no hasColumns answers no columns at all", async () => {
    objects = await connectedWithObjects();
    const index = (objects.getCapabilities().objectKinds ?? []).find((kind) => kind.id === "index");

    expect(kindHasColumns(index)).toBe(false);
    expect((await objects.describeObject(["idx_orders_customer"], "index")).columns).toEqual([]);
  });

  test("an index and a trigger describe as three empty arrays, without a round trip", async () => {
    objects = await connectedWithObjects();

    const index = await objects.describeObject(["idx_orders_customer"], "index");
    const trigger = await objects.describeObject(["orders", "orders_stamp"], "trigger");

    expect(index).toEqual({ path: ["idx_orders_customer"], columns: [], indexes: [], foreignKeys: [] });
    expect(trigger).toEqual({ path: ["orders", "orders_stamp"], columns: [], indexes: [], foreignKeys: [] });
    // A true fact about those kinds rather than a failed read: nothing was sent.
    expect(calls).toEqual([]);
  });

  test("a relation is described in two round trips, not one per index and parent", async () => {
    // The difference from the SQLite provider, and the reason it is not a copy: there
    // every read is a call into a file handle, and here every read is a request across
    // a network. One batch asks for the columns, the index list and the foreign keys;
    // one more asks for every index's columns.
    objects = await connectedWithObjects();

    await objects.describeObject(["orders"], "table");

    expect(calls).toHaveLength(2);
    expect(sentCalls().map((call) => call.args)).toEqual([
      ["orders", "main"],
      ["orders", "main"],
      // THREE binds, and the schema first: the foreign key statement's first placeholder
      // belongs to the parent-key subquery, because SQLite numbers placeholders by where
      // they appear in the text and the select list is written before the FROM. Passing
      // the two-value array here is what sqld answers "Arguments do not match SQL
      // parameters: value for parameter 3 not found" to - measured against a live server,
      // and invisible to any fake that does not count binds (#789).
      ["main", "orders", "main"],
      ["idx_orders_customer", "main"],
      ["idx_orders_placed", "main"],
    ]);
  });

  test("a catalog row with no name is refused rather than addressed by an empty segment", async () => {
    // `readText(...) ?? ""` is the obvious spelling and it MASKS: the row would become an
    // object whose last path segment is the empty string, which the tree would draw, let a
    // user click, and then describe as nothing. Every one of these catalog columns is NOT
    // NULL, so this is unreachable on the engine - which is precisely why a silent default
    // would never be noticed if it ever stopped being unreachable.
    objects = await connectedWithObjects();
    server = () => result([["name", "TEXT"]], [[{ type: "null" }]]);

    const refusal = await objects.listObjects([], "index").then(
      () => undefined,
      (error: unknown) => error,
    );

    expect(refusal).toBeInstanceOf(DatabaseError);
    expect((refusal as DatabaseError).message).toBe("libSQL answered a index row with no name");
    // Naming the statement, so whoever reads it knows which of the four listings produced
    // the row rather than being told only that something was nameless.
    expect((refusal as DatabaseError).query).toContain("s.type = 'index'");
  });

  test("a trigger row with no parent is refused rather than promoted to the top level", async () => {
    // The parent is half of an attached kind's ADDRESS, so losing it is not a cosmetic
    // gap: the trigger would come back as `["orders_stamp"]`, a top-level row sitting
    // beside the tables. Which kinds require one is read off `attachedTo` rather than off
    // whether the column is present, so the three unattached listings are unaffected.
    objects = await connectedWithObjects();
    server = () =>
      result(
        [
          ["name", "TEXT"],
          ["parent", "TEXT"],
        ],
        [[text("orders_stamp"), { type: "null" }]],
      );

    await expect(objects.listObjects([], "trigger")).rejects.toThrow(/libSQL answered a trigger parent with no name/);
    // The control: the same null in the column the other three kinds do not select is not
    // an error, because absent is what "this kind does not nest" looks like.
    await expect(objects.listObjects([], "index")).resolves.toEqual([
      { path: ["orders_stamp"], name: "orders_stamp", kind: "index" },
    ]);
  });

  test("an object that is not there is a failed read and says so", async () => {
    objects = await connectedWithObjects();

    // Zero columns cannot be a real answer: SQLite refuses `CREATE TABLE t()`, so every
    // table and every view has at least one column.
    await expect(objects.describeObject(["nowhere"], "table")).rejects.toThrow(/No libSQL table named nowhere/);
  });

  // --------------------------------------------------------------------------
  // Declarations, shapes and refusals
  // --------------------------------------------------------------------------

  test("a kind the database holds none of still draws its folder, badged zero", async () => {
    // A database with tables and nothing else: no view, no index, no trigger.
    objects = await connectedWithObjects({
      tableList: TABLE_LIST_ROWS.filter((row) => row.type !== "view"),
      sqliteSchema: SQLITE_SCHEMA_ROWS.filter((row) => row.type === "table"),
    });

    const counts = await objects.countObjects([]);

    // Every declared kind is seeded at zero before a row overwrites it. Building the
    // record from the GROUP BY rows alone would leave the kind out entirely, and an
    // absent kind means something else and stronger: the engine has no such concept.
    expect(counts).toEqual({
      table: { count: EXPECTED_COUNTS.table },
      view: { count: 0 },
      index: { count: 0 },
      trigger: { count: 0 },
    });
  });

  test("a kind libSQL does not declare is refused by name, in both methods", async () => {
    objects = await connectedWithObjects();

    await expect(objects.listObjects([], "procedure")).rejects.toThrow(/libSQL declares no object kind "procedure"/);
    await expect(objects.describeObject(["x"], "procedure")).rejects.toThrow(
      /libSQL declares no object kind "procedure"/,
    );
  });

  test("a kind that is declared but has no listing statement says so, not that it is undeclared", async () => {
    // Two guards, two different facts, and the message must not confuse them. The
    // DECLARATION decides whether a kind exists; whether this file can list it is a gap in
    // this file. Reporting the second as the first would send whoever reads it looking at
    // `objectKinds` for a kind that is right there.
    objects = await connectedWithObjects();
    const real = objects.getCapabilities();
    spyOn(objects, "getCapabilities").mockReturnValue({
      ...real,
      objectKinds: [...(real.objectKinds ?? []), { id: "sequence", role: "config", label: "S", labelPlural: "S" }],
    });

    await expect(objects.listObjects([], "sequence")).rejects.toThrow(
      /libSQL declares the kind "sequence" but has no statement that lists it/,
    );
    // The control: a kind nothing declares still reports the other sentence.
    await expect(objects.listObjects([], "procedure")).rejects.toThrow(/declares no object kind "procedure"/);
  });

  test("a container path of another engine's shape is refused rather than read", async () => {
    objects = await connectedWithObjects();

    // Awaited one at a time rather than built into an array first: a rejected promise
    // that waits a tick for its handler is an unhandled rejection under bun.
    await expect(objects.countObjects(["main"])).rejects.toThrow(
      /A libSQL container path is empty, received \["main"\]/,
    );
    await expect(objects.listObjects(["main"], "table")).rejects.toThrow(
      /A libSQL container path is empty, received \["main"\]/,
    );
    await expect(objects.describeObject(["main", "orders"], "table")).rejects.toThrow(
      /A libSQL "table" path is \[name\], received \["main","orders"\]/,
    );
    await expect(objects.describeObject(["orders_stamp"], "trigger")).rejects.toThrow(
      /A libSQL "trigger" path is \[table, name\], received \["orders_stamp"\]/,
    );
    // A caller mistake is not something the engine refused, so nothing was sent.
    expect(calls).toEqual([]);
  });

  test("every population the object surface reads applies the SAME two restrictions", async () => {
    // Standing ruling 5f from the statement side, and it is a SHAPE assertion on purpose.
    // It covers exactly the two places no data this engine can produce reaches, and every
    // other restriction is pinned by behaviour in the tests above:
    //
    // - `t.schema = ?` on the TABLE listing. sqld refuses `CREATE TEMP TABLE` and
    //   `CREATE TABLE temp.x` outright and accepts only `CREATE VIEW temp.x`, whose row
    //   is typed `view` and so can never enter a `type IN ('table','virtual')` listing.
    //   The counts arm and the view listing are driven behaviourally by the temp test.
    // - the reserved-name predicate on the TRIGGER listing. Measured on sqld 0.24.33:
    //   `CREATE TRIGGER sqlite_guard ...` is refused, "object name reserved for internal
    //   use", and the engine creates no trigger of its own, so no `sqlite`-prefixed
    //   trigger row can exist. The INDEX listing is NOT in this category and is pinned
    //   behaviourally above, because an implicit `sqlite_autoindex_*` on a ROWID table IS
    //   a row of `sqlite_schema`.
    //
    // Both stay, because a badge and its folder must enumerate one population however the
    // engine happens to be shaped today, and this is what keeps them from drifting apart.
    objects = await connectedWithObjects();
    const declared = (objects.getCapabilities().objectKinds ?? []).map((kind) => kind.id);

    await objects.countObjects([]);
    for (const kind of declared) await objects.listObjects([], kind);

    const statements = sentCalls();
    // Derived from the declaration rather than pinned to a number, so a new kind cannot
    // slip past by making this loop shorter. One statement per kind, plus the counts.
    expect(statements).toHaveLength(declared.length + 1);
    for (const statement of statements) {
      // Counted per POPULATION rather than merely "present", because the counts statement
      // reads two catalogs in one `UNION ALL`: asserting containment would let one of its
      // two arms lose the predicate while the other still carried it.
      const populations = statement.sql.match(/FROM (?:pragma_table_list|sqlite_schema)\b/g) ?? [];
      const reserved = statement.sql.match(/NOT LIKE 'sqlite\\_%' ESCAPE '\\'/g) ?? [];
      expect(reserved).toHaveLength(populations.length);
      // Only `pragma_table_list` spans schemas; `sqlite_schema` unqualified is already
      // `main`. So the schema restriction is required exactly where it can apply, and the
      // bind count follows from that rather than from a typed number.
      const spanning = statement.sql.match(/FROM pragma_table_list\b/g) ?? [];
      const restricted = statement.sql.match(/t\.schema = \?/g) ?? [];
      expect(restricted).toHaveLength(spanning.length);
      expect(statement.args).toEqual(spanning.map(() => "main"));
    }
  });

  test("the container depth, the object path and the name bind are DERIVED, which a two-level declaration shows", async () => {
    // Standing ruling 5g. On a ZERO-container engine this test is the only thing that
    // can tell a derivation from a literal: `container.length !== 0` and `path[0]` for
    // the object name are behaviour-identical to the derived forms at depth 0, which is
    // why the same defect shipped three times before it was named. Driven all the way
    // to a BOUND VALUE, never stopping at the refusal.
    objects = await connectedWithObjects();
    const real = objects.getCapabilities();
    spyOn(objects, "getCapabilities").mockReturnValue({
      ...real,
      containerLevels: [
        { id: "catalog", label: "Catalog", labelPlural: "Catalogs" },
        { id: "schema", label: "Database", labelPlural: "Databases" },
      ],
    });

    // The refusal half: the depth comes from `containerDepth()`, so the EMPTY container
    // path - the only one this engine really accepts - is now wrong.
    await expect(objects.countObjects([])).rejects.toThrow(
      /A libSQL container path is \[catalog, database\], received \[\]/,
    );
    await expect(objects.listObjects([], "table")).rejects.toThrow(
      /A libSQL container path is \[catalog, database\], received \[\]/,
    );
    await expect(objects.describeObject(["orders"], "table")).rejects.toThrow(
      /A libSQL "table" path is \[catalog, database, name\], received \["orders"\]/,
    );

    // The bound-value half, which a refusal-only test cannot see.
    expect(await objects.countObjects(["cat", "sch"])).toEqual({
      table: { count: EXPECTED_COUNTS.table },
      view: { count: EXPECTED_COUNTS.view },
      index: { count: EXPECTED_COUNTS.index },
      trigger: { count: EXPECTED_COUNTS.trigger },
    });

    const tables = await objects.listObjects(["cat", "sch"], "table");
    expect(tables.map((table) => table.path)).toContainEqual(["cat", "sch", "orders"]);
    const triggers = await objects.listObjects(["cat", "sch"], "trigger");
    expect(triggers.map((trigger) => trigger.path)).toEqual([
      ["cat", "sch", "badges", "badges"],
      ["cat", "sch", "order_summary", "order_summary_guard"],
      ["cat", "sch", "orders", "orders_stamp"],
    ]);

    // AND THE BIND. `path[path.length - 1]` is `orders` at this depth while `path[0]` is
    // the CATALOG: both are depth-identical at 0, so only this declaration tells them
    // apart. Binding either literal would ask for an object called `cat` or `sch`.
    calls = [];
    const detail = await objects.describeObject(["cat", "sch", "orders"], "table");
    expect(detail.path).toEqual(["cat", "sch", "orders"]);
    expect(detail.columns.map((column) => column.name)).toEqual(["id", "customer_id", "total", "tax", "placed_at"]);
    // The foreign key statement binds the SCHEMA first, so its `args[0]` is `main`; every
    // other read binds the object's own name, which is `orders` and never `cat` or `sch`.
    expect(sentCalls().map((call) => call.args[0])).toEqual([
      "orders",
      "orders",
      "main",
      "idx_orders_customer",
      "idx_orders_placed",
    ]);
  });

  // --------------------------------------------------------------------------
  // A read the engine refuses
  // --------------------------------------------------------------------------

  test("a refused count carries the server's own sentence, for every kind at once", async () => {
    objects = await connectedWithObjects();
    // A failed statement is an HTTP 200 carrying the engine's words, which is the whole
    // reason this provider never reads `response.ok` as the verdict.
    server = () => failure("SQLite error: no such table: pragma_table_list", "SQLITE_UNKNOWN");

    const counts = await objects.countObjects([]);

    expect(counts).toEqual({
      table: { unavailable: "SQLite error: no such table: pragma_table_list" },
      view: { unavailable: "SQLite error: no such table: pragma_table_list" },
      index: { unavailable: "SQLite error: no such table: pragma_table_list" },
      trigger: { unavailable: "SQLite error: no such table: pragma_table_list" },
    });
    // Verbatim, with no product prefix in front of the server's words.
    for (const count of Object.values(counts)) {
      expect(isCountUnavailable(count) ? count.unavailable : "").not.toContain("libSQL");
    }
  });

  test("a count refused because the server is unreachable says that instead of zero", async () => {
    objects = await connectedWithObjects();
    globalThis.fetch = (() => Promise.reject(new Error("connect ECONNREFUSED"))) as unknown as typeof fetch;

    const counts = await objects.countObjects([]);

    expect(isCountUnavailable(counts.table) ? counts.table.unavailable : "").toContain("connect ECONNREFUSED");
  });

  test("a refused LISTING raises against the statement the server received", async () => {
    objects = await connectedWithObjects();
    server = (sql, args) =>
      /pragma_table_list/.test(sql)
        ? failure("SQLite error: no such table: pragma_table_list", "SQLITE_UNKNOWN")
        : objectServer()(sql, args);

    const refusal = await objects.listObjects([], "table").then(
      () => undefined,
      (error: unknown) => error,
    );

    expect(refusal).toBeInstanceOf(DatabaseError);
    expect((refusal as DatabaseError).message).toContain("no such table: pragma_table_list");
    expect((refusal as DatabaseError).provider).toBe("libsql");
    expect((refusal as DatabaseError).query).toContain("pragma_table_list");
    // The kinds that read `sqlite_schema` are untouched, which is the control.
    await expect(objects.listObjects([], "index")).resolves.toHaveLength(EXPECTED_COUNTS.index);
  });

  test("a refused DETAIL read raises against the statement, per statement of the batch", async () => {
    objects = await connectedWithObjects();
    server = (sql, args) =>
      /pragma_index_list/.test(sql)
        ? failure("SQLite error: disk I/O error", "SQLITE_IOERR")
        : objectServer()(sql, args);

    const refusal = await objects.describeObject(["orders"], "table").then(
      () => undefined,
      (error: unknown) => error,
    );

    expect(refusal).toBeInstanceOf(DatabaseError);
    expect((refusal as DatabaseError).message).toContain("disk I/O error");
    expect((refusal as DatabaseError).query).toContain("pragma_index_list");
  });

  test("an expired token during a listing stays an authentication failure", async () => {
    // The status carries the distinction and the wording does not, which is why the
    // object surface maps through the provider's own mapping rather than wrapping
    // every failure as a query error.
    objects = await connectedWithObjects();
    globalThis.fetch = (() =>
      Promise.resolve(
        new Response(JSON.stringify({ error: "JWT error: ExpiredSignature" }), { status: 401 }),
      )) as unknown as typeof fetch;

    await expect(objects.listObjects([], "table")).rejects.toBeInstanceOf(AuthenticationError);
  });

  test("a catalog row whose kind names a prototype member draws no folder", async () => {
    // `Object.hasOwn` and not `in`: `"toString" in counts` is true on any object
    // literal, so the `in` spelling writes a folder for a kind nothing declared.
    objects = await connectedWithObjects();
    server = () =>
      result(
        [
          ["kind", "TEXT"],
          ["n", null],
        ],
        [
          [text("table"), int(3)],
          [text("toString"), int(9)],
          [text("constructor"), int(9)],
          [text("__proto__"), int(9)],
        ],
      );

    const counts = await objects.countObjects([]);

    // The control, so this is not a test of an empty read.
    expect(counts.table).toEqual({ count: 3 });
    expect(Object.keys(counts).sort()).toEqual(["index", "table", "trigger", "view"]);
  });
});

/**
 * The bulk column read (#789).
 *
 * The fake below replays the SAME fixture the single read replays, so every "the two
 * readings agree" assertion compares two answers the provider produced. What it cannot see
 * is a rewrite of the SQL itself (standing ruling 5b), which is why the whole method was
 * also run against a live `ghcr.io/tursodatabase/libsql-server:v0.24.33` - the measurements
 * and what that run caught are in docs/providers/libsql.md.
 */
/**
 * The Source read, against the catalog the live server built (#789 Phase 2).
 *
 * Every definition asserted here is the text sqld 0.24.33 answered for that row, captured
 * from `docker/sqlite-init/02-libsql-object-fixture.sql` and carried in `SQLITE_SCHEMA_SQL`.
 * The fake applies the predicates the statement actually spells, so dropping one widens the
 * population here the way it would against the real server.
 *
 * NOTHING IN THIS BLOCK KEYS ON REFUSAL WORDING, and that is deliberate rather than an
 * oversight: sqld and Turso Cloud word the identical refusal differently (section 3.2 of the
 * provider doc), so a test that pinned either sentence would pass on one deployment and fail
 * on the other. The one sentence asserted below is OURS, for a NULL definition the engine
 * says nothing at all about.
 */
describe("LibSQLProvider object source (#789)", () => {
  let objects: LibSQLProvider;

  afterEach(async () => {
    if (objects?.isConnected()) await objects.disconnect();
  });

  test("declares source on exactly the kinds that have a definition text", async () => {
    objects = await connectedWithObjects();
    const kinds = objects.getCapabilities().objectKinds ?? [];
    const declared = kinds
      .filter((kind) => kind.hasSource === true)
      .map((kind) => [kind.id, kind.sourceLanguage] as const)
      .sort();

    expect(declared).toEqual([
      ["index", "sql"],
      ["table", "sql"],
      ["trigger", "sql"],
      ["view", "sql"],
    ]);
    // The other direction, so a kind added later cannot quietly gain a Source tab. libSQL is
    // one of the engines where this list is EMPTY, and the assertion is still what fails the
    // day a fifth kind is declared without a decision about its source.
    expect(
      kinds
        .filter((kind) => kind.hasSource !== true)
        .map((kind) => kind.id)
        .sort(),
    ).toEqual([]);
  });

  test("reads the definition of a table and says what the text is", async () => {
    objects = await connectedWithObjects();

    const document = await objects.readObjectSource!(["orders"], "table");

    expect(document.path).toEqual(["orders"]);
    expect(document.kind).toBe("table");
    expect(document.parts).toHaveLength(1);
    const [part] = document.parts;
    expect(isSourcePartUnavailable(part)).toBe(false);
    if (isSourcePartUnavailable(part)) throw new Error("narrowing");
    expect(part.id).toBe("definition");
    expect(part.label).toBe("Definition");
    expect(part.text).toBe(definedSchemaSql("table", "orders"));
    expect(part.text).toContain("tax REAL GENERATED ALWAYS AS (total * 0.2) VIRTUAL");
    expect(part.language).toBe("sql");
    // The author's own bytes: the second line of the fixture statement comes back with its
    // twenty-one-space continuation indent, which a regeneration would destroy.
    expect(part.form).toBe("complete");
    expect(part.origin).toBe("stored");
    expect(part.text).toContain("\n                     total REAL NOT NULL,");
    expect(part.truncated).toBeUndefined();
  });

  /**
   * The per-kind READ-THE-TEXT pin, population taken from the DECLARATION (#789).
   *
   * Recipe rule 6: a whole-statement pin is necessary and NOT sufficient. A wrong reply
   * column reads as `undefined`, the provider correctly turns that into a refusal, and a
   * refusal passes the conformance walk, the statement pin and every count and length
   * assertion there is. Only comparing the TEXT, per object, can see it.
   *
   * Three guards against a vacuous arm: a declared source-bearing kind whose listing is
   * empty THROWS by name, a listed object whose capture carries no definition THROWS by name
   * through `sqliteSchemaSql`, and the total read is compared against the total listed.
   */
  test("every object of every source-bearing kind reads back the text the live capture holds", async () => {
    objects = await connectedWithObjects();
    const sourceKinds = (objects.getCapabilities().objectKinds ?? []).filter((kind) => kind.hasSource === true);
    if (sourceKinds.length === 0) throw new Error("no kind declares hasSource, so this test reads nothing");

    let read = 0;
    let listedTotal = 0;
    for (const spec of sourceKinds) {
      const listed = await objects.listObjects([], spec.id);
      if (listed.length === 0) throw new Error(`the fixture holds no ${spec.id}, so its source read is unexercised`);
      listedTotal += listed.length;
      for (const object of listed) {
        const document = await objects.readObjectSource!(object.path, spec.id);
        const [part] = document.parts;
        if (isSourcePartUnavailable(part)) {
          throw new Error(`${spec.id}/${object.name} answered the refusal "${part.unavailable}"`);
        }
        expect(part.text).toBe(definedSchemaSql(spec.id, object.name));
        read += 1;
      }
    }
    expect(read).toBe(listedTotal);
    expect(read).toBe(EXPECTED_COUNTS.table + EXPECTED_COUNTS.view + EXPECTED_COUNTS.index + EXPECTED_COUNTS.trigger);
  });

  test("the statement is one text, with the type BOUND and taken from the KIND", async () => {
    objects = await connectedWithObjects();

    await objects.readObjectSource!(["orders"], "table");

    const sent = calls.map((call) => JSON.parse(call.body ?? "{}").requests[0].stmt);
    // The whole statement as a LITERAL, never the module's own constant: importing it would
    // move both sides of the assertion together and pin nothing (recipe rule 6).
    expect(sent).toEqual([
      {
        sql: `
      SELECT s.sql AS sql
        FROM sqlite_schema AS s
       WHERE s.type = ?
         AND s.name = ?
    `,
        args: [
          { type: "text", value: "table" },
          { type: "text", value: "orders" },
        ],
      },
    ]);
  });

  /**
   * The one object this fixture holds that can tell a KIND-derived type from a name match.
   *
   * `badges` is a table AND a trigger, which sqld accepts. `SELECT sql FROM sqlite_schema
   * WHERE name = 'badges'` answers TWO rows with the table's first, measured live, so a read
   * that resolved the type from the name would hand a reader the table's DDL under the
   * trigger's address.
   */
  test("a trigger sharing a name with a table reads the TRIGGER, not the table", async () => {
    objects = await connectedWithObjects();

    const [triggerPart] = (await objects.readObjectSource!(["badges", "badges"], "trigger")).parts;
    const [tablePart] = (await objects.readObjectSource!(["badges"], "table")).parts;
    if (isSourcePartUnavailable(triggerPart) || isSourcePartUnavailable(tablePart)) throw new Error("narrowing");

    expect(triggerPart.text).toBe(definedSchemaSql("trigger", "badges"));
    expect(tablePart.text).toBe(definedSchemaSql("table", "badges"));
    expect(triggerPart.text).not.toBe(tablePart.text);
  });

  test("an object that is not there RAISES, naming the segment, and never answers a refusal", async () => {
    objects = await connectedWithObjects();

    await expect(objects.readObjectSource!(["no_such_table"], "table")).rejects.toThrow(
      /No libSQL table named no_such_table/,
    );
    await expect(objects.readObjectSource!(["orders", "orders"], "trigger")).rejects.toThrow(
      /No libSQL trigger named orders/,
    );
  });

  test("a kind this engine does not declare, and a path of the wrong shape, both raise", async () => {
    objects = await connectedWithObjects();

    await expect(objects.readObjectSource!(["x"], "procedure")).rejects.toThrow(
      /libSQL declares no object kind "procedure"/,
    );
    await expect(objects.readObjectSource!(["orders_stamp"], "trigger")).rejects.toThrow(
      /A libSQL "trigger" path is \[table, name\], received \["orders_stamp"\]/,
    );
  });

  /**
   * The refusal this engine CANNOT produce, driven anyway.
   *
   * `sqlite_schema.sql` is NULL for exactly one row of this catalog, the autoindex behind
   * `code TEXT UNIQUE`, and the listings exclude every `sqlite_`-prefixed name, so no path
   * the tree offers addresses it. The provider doc states that as a CANNOT. The arm exists
   * anyway, because an empty definition must never reach an editor as a definition, and this
   * is what the suite would say if the read ever started producing one.
   *
   * The path is authored rather than listed, which is the exception the control below covers:
   * the same catalog answers the readable rows above, so a refusal here is the NULL and not
   * a broken connection.
   */
  test("the one NULL definition in the catalog becomes a refusal part, not an empty text", async () => {
    objects = await connectedWithObjects();

    const [part] = (await objects.readObjectSource!(["sqlite_autoindex_badges_1"], "index")).parts;

    expect(isSourcePartUnavailable(part)).toBe(true);
    if (!isSourcePartUnavailable(part)) throw new Error("narrowing");
    expect(part.id).toBe("definition");
    // OUR sentence, because the engine supplies none for a NULL, and nothing in this block
    // keys on sqld's or Turso's own wording for anything.
    expect(part.unavailable).toContain("sqlite_schema.sql is NULL");
    // The CAUSE, asserted here and nowhere else: a stored NULL is the only one of the three
    // blank shapes for which "an index it created for itself" is a true reason.
    expect(part.unavailable).toContain("an index it created for itself");
    expect(part.unavailable.trim().length).toBeGreaterThan(20);
  });

  /**
   * The other half of the blank arm, which no row of this catalog can produce.
   *
   * The NULL case above is a real row; a whitespace-only definition is not a state this
   * engine reaches at all, and it is driven here because "unreachable" is the reason to
   * write the arm rather than a reason to leave it unmeasured. Without this test, deleting
   * the `.trim() === ""` half of the guard failed NOTHING in this suite (measured, 95 pass
   * 0 fail), and an editor would have opened over three spaces presented as a definition.
   */
  test("a whitespace-only definition is refused too, and is NOT reported as the engine's NULL", async () => {
    objects = await connectedWithObjects();
    server = (sql) => (/s\.sql AS sql/.test(sql) ? result([["sql", "TEXT"]], [[text("   \n  ")]]) : failure("x", "y"));

    const [part] = (await objects.readObjectSource!(["orders"], "table")).parts;

    expect(isSourcePartUnavailable(part)).toBe(true);
    if (!isSourcePartUnavailable(part)) throw new Error("narrowing");
    expect(part.unavailable).toContain("no non-whitespace character");
    // The row is there and so is the column, so a sentence blaming an index the engine made
    // for itself would be false of this shape.
    expect(part.unavailable).not.toContain("an index it created for itself");
    expect(part.unavailable).not.toContain("is NULL");
  });

  /**
   * Recipe rule 6's own failure mode, on the transport that hides it best.
   *
   * A reply whose definition COLUMN is spelled differently carries no `sql` key at all, reads
   * as `undefined`, and becomes a refusal that passes the conformance walk, the statement pin
   * and every count assertion. It is refused here, which is the only correct answer, and the
   * SENTENCE has to send a reader to the statement: this is a defect in the read, not a fact
   * about the object, and the stored-NULL reason would be false as well as misleading.
   */
  test("a reply carrying no sql column at all becomes a refusal that names the READ, not the object", async () => {
    objects = await connectedWithObjects();
    server = (sql) =>
      /s\.sql AS sql/.test(sql)
        ? result([["definition", "TEXT"]], [[text("CREATE TABLE orders (a)")]])
        : failure("x", "y");

    const [part] = (await objects.readObjectSource!(["orders"], "table")).parts;

    expect(isSourcePartUnavailable(part)).toBe(true);
    if (!isSourcePartUnavailable(part)) throw new Error("narrowing");
    expect(part.unavailable).toContain("no sqlite_schema.sql column at all");
    expect(part.unavailable).not.toContain("an index it created for itself");
  });

  test("a server failure RAISES through the provider's own mapping rather than becoming a refusal", async () => {
    objects = await connectedWithObjects();
    server = () => failure("SQLite error: no such table: sqlite_schema", "SQLITE_UNKNOWN");

    await expect(objects.readObjectSource!(["orders"], "table")).rejects.toThrow(/no such table: sqlite_schema/);
  });

  test("the caller's bound cuts the text and says so, and an exact answer is never marked", async () => {
    objects = await connectedWithObjects();
    const [whole] = (await objects.readObjectSource!(["orders"], "table")).parts;
    if (isSourcePartUnavailable(whole)) throw new Error("narrowing");

    const [cut] = (await objects.readObjectSource!(["orders"], "table", 20)).parts;
    if (isSourcePartUnavailable(cut)) throw new Error("narrowing");

    expect(whole.text.length).toBeGreaterThan(20);
    expect(cut.text).toBe(whole.text.slice(0, 20));
    expect(cut.truncated).toEqual({ limit: 20, reason: sourceBoundTruncationReason(20) });

    const [uncut] = (await objects.readObjectSource!(["orders"], "table", whole.text.length)).parts;
    if (isSourcePartUnavailable(uncut)) throw new Error("narrowing");
    expect(uncut.truncated).toBeUndefined();
  });

  /**
   * The three guards a correct DECLARATION cannot reach, driven through the declaration.
   *
   * A kind that is declared and not source-bearing, a kind that declares source and forgets
   * the Monaco id that renders it, and a kind that declares source with no catalog type
   * behind it. None is reachable with the four kinds libSQL really declares.
   */
  test("a declared kind that is not source-bearing, or is under-declared, raises by name", async () => {
    objects = await connectedWithObjects();
    const real = objects.getCapabilities();
    const withKind = (extra: ObjectKindSpec) =>
      spyOn(objects, "getCapabilities").mockReturnValue({ ...real, objectKinds: [...(real.objectKinds ?? []), extra] });
    const base = { id: "synonym", role: "config" as const, label: "Synonym", labelPlural: "Synonyms" };

    for (const [extra, pattern] of [
      [base, /libSQL publishes no definition text for the kind "synonym"/],
      [{ ...base, hasSource: true }, /declares readable source for the kind "synonym" and no sourceLanguage/],
      [
        { ...base, hasSource: true, sourceLanguage: "sql" },
        /declares readable source for the kind "synonym" but has no catalog type that reads it/,
      ],
    ] as const) {
      const spy = withKind(extra);
      try {
        await expect(objects.readObjectSource!(["x"], "synonym")).rejects.toThrow(pattern);
      } finally {
        spy.mockRestore();
      }
    }
  });

  /**
   * Standing ruling 5g, on a zero-level engine, driven to the BOUND VALUE (#789).
   *
   * The declaration is swapped for a two-level one and a three-segment path is driven all the
   * way to the binds. Two mutations die here and nowhere else in this suite: a shape check
   * written `path.length !== 1` refuses this path, and a name bind written `path[0]` binds
   * "cat" instead of "orders".
   */
  test("derives the object name and the path shape from the DECLARATION, not from a position", async () => {
    objects = await connectedWithObjects();
    const spy = spyOn(objects, "getCapabilities").mockReturnValue({
      ...objects.getCapabilities(),
      containerLevels: [
        { id: "catalog", label: "Database", labelPlural: "Databases" },
        { id: "schema", label: "Schema", labelPlural: "Schemas" },
      ],
    });
    try {
      const document = await objects.readObjectSource!(["cat", "sch", "orders"], "table");

      const sent = calls.map((call) => JSON.parse(call.body ?? "{}").requests[0].stmt.args);
      expect(sent).toEqual([
        [
          { type: "text", value: "table" },
          { type: "text", value: "orders" },
        ],
      ]);
      expect(document.path).toEqual(["cat", "sch", "orders"]);
    } finally {
      spy.mockRestore();
    }
  });
});

describe("LibSQLProvider bulk column read (#789)", () => {
  let objects: LibSQLProvider;

  test("describes every table in ONE round trip, and each detail is what the single read answers", async () => {
    objects = await connectedWithObjects();

    const batch = await objects.describeObjects([], "table");

    expect(batch.truncated).toBeUndefined();
    expect(batch.details.map((detail) => detail.path)).toEqual([
      ["archive"],
      ["badges"],
      ["customers"],
      ["legacy"],
      ["legacy_ref"],
      ["notes"],
      ["orders"],
      ["regions"],
      ["shipments"],
      ["sqliteXledger"],
    ]);
    // ONE request for ten objects. This is the engine-specific half of the decision: the
    // five statements are independent of each other, so they go in one batch, where the
    // SQLite provider issues five calls into a file handle.
    expect(calls).toHaveLength(1);

    // One mapper serves both reads, so a divergence here is the bulk read spelling a
    // column, an index or a foreign key differently from the single read of the same table.
    for (const detail of batch.details) {
      expect(detail).toEqual(await objects.describeObject(detail.path, "table"));
    }
  });

  test("every described path is one listObjects produced", async () => {
    objects = await connectedWithObjects();

    const listed = await objects.listObjects([], "table");
    const batch = await objects.describeObjects([], "table");

    expect(batch.details.map((detail) => detail.path)).toEqual(listed.map((object) => object.path));
  });

  test("the five statements bind what their placeholders ask for, and the foreign key read takes three", async () => {
    // The arity pin, and it is not theoretical: a two-value array against the foreign key
    // statement's three placeholders is what sqld answers "Arguments do not match SQL
    // parameters: value for parameter 3 not found" to. Measured against a live server, and
    // invisible to this fake, which dispatches on statement text and never counted binds.
    objects = await connectedWithObjects();

    await objects.describeObjects([], "table", 4);

    const sent = sentCalls();
    expect(sent).toHaveLength(5);
    for (const call of sent) {
      const placeholders = (call.sql.match(/\?/g) ?? []).length;
      expect(call.args).toHaveLength(placeholders);
    }
    // The schema, the bound `limit + 1`, and one schema per pragma the statement joins.
    // Hrana encodes an integer as a decimal STRING on the wire, which is why the 5 comes
    // back quoted here and never in a provider assertion.
    expect(sent.map((call) => call.args)).toEqual([
      ["main", "5"],
      ["main", "5", "main"],
      ["main", "5", "main"],
      ["main", "5", "main", "main"],
      ["main", "5", "main", "main"],
    ]);
  });

  test("a view is described from the same statements a table is", async () => {
    objects = await connectedWithObjects();

    const batch = await objects.describeObjects([], "view");

    expect(batch.details.map((detail) => detail.path)).toEqual([["order_summary"]]);
    expect(batch.details[0]!.columns.map((column) => column.name)).toEqual(["name", "total"]);
    expect(batch.details[0]!.indexes).toEqual([]);
    expect(batch.details[0]!.foreignKeys).toEqual([]);
  });

  test("an index and a trigger answer an empty batch without touching the network", async () => {
    objects = await connectedWithObjects();

    await expect(objects.describeObjects([], "index")).resolves.toEqual({ details: [] });
    await expect(objects.describeObjects([], "trigger")).resolves.toEqual({ details: [] });

    // Not "no rows came back": nothing was sent. Neither kind has columns on this engine -
    // `pragma_table_xinfo` answers zero rows for either name - and that is a fact about the
    // KIND, answered from the declaration.
    expect(calls).toEqual([]);
    // The control: a kind that DOES have columns still reaches the server.
    await objects.describeObjects([], "view");
    expect(calls.length).toBeGreaterThan(0);
  });

  test("an undeclared kind raises, naming the engine and the kind", async () => {
    objects = await connectedWithObjects();

    await expect(objects.describeObjects([], "sequence")).rejects.toThrow(/libSQL declares no object kind "sequence"/);
  });

  test("a container path of the wrong shape raises through the declaration, not a literal depth", async () => {
    objects = await connectedWithObjects();

    await expect(objects.describeObjects(["main"], "table")).rejects.toThrow(/A libSQL container path is empty/);
  });

  test("a two-level declaration is accepted and reaches the produced path", async () => {
    // Standing ruling 5g, driven to a VALUE and not to a refusal. libSQL declares no
    // container level, so `container.length !== 0` and `[name]` are behaviour-identical on
    // the real server and only a differently shaped declaration tells them apart.
    objects = await connectedWithObjects();
    spyOn(objects, "getCapabilities").mockReturnValue({
      ...objects.getCapabilities(),
      containerLevels: [
        { id: "catalog", label: "Catalog", labelPlural: "Catalogs" },
        { id: "schema", label: "Database", labelPlural: "Databases" },
      ],
    });

    const batch = await objects.describeObjects(["cat", "sch"], "view");

    expect(batch.details.map((detail) => detail.path)).toEqual([["cat", "sch", "order_summary"]]);
    await expect(objects.describeObjects([], "view")).rejects.toThrow(
      /A libSQL container path is \[catalog, database\]/,
    );
  });

  test("a limit that is not a positive whole number raises rather than clamping", async () => {
    objects = await connectedWithObjects();

    for (const limit of [0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY]) {
      await expect(objects.describeObjects([], "table", limit)).rejects.toThrow(
        /A libSQL bulk column read limit must be a positive whole number/,
      );
    }
    // Nothing was sent for any of them: the guard is before the batch.
    expect(calls).toEqual([]);
  });

  test("a bound that bites reports the caller's own limit, and one that does not never reports", async () => {
    objects = await connectedWithObjects();

    const bounded = await objects.describeObjects([], "table", 3);
    expect(bounded.details.map((detail) => detail.path)).toEqual([["archive"], ["badges"], ["customers"]]);
    expect(bounded.truncated?.limit).toBe(3);
    expect(bounded.truncated?.reason.length).toBeGreaterThan(0);
    // The bound cuts OBJECTS and never columns: every detail it kept is complete.
    expect(bounded.details[2]!.columns.map((column) => column.name)).toEqual(["id", "name", "country"]);

    // Exactly as many as the folder holds. `limit + 1` is what the statement carries, so a
    // saturated read is told from an exact one with no second count.
    const exact = await objects.describeObjects([], "table", EXPECTED_COUNTS.table);
    expect(exact.details).toHaveLength(EXPECTED_COUNTS.table);
    expect(exact.truncated).toBeUndefined();

    const unbounded = await objects.describeObjects([], "table");
    expect(unbounded.details).toHaveLength(EXPECTED_COUNTS.table);
    expect(unbounded.truncated).toBeUndefined();
  });

  test("the engine cuts under BINARY and the answer is sorted by path, and those are two different orders", async () => {
    // The one probe that separates the two, and it is a real disagreement rather than a
    // contrived one - see TABLE_LIST_WITH_EXOTIC for the live measurement behind the
    // fixture's order.
    objects = await connectedWithObjects({ tableList: TABLE_LIST_WITH_EXOTIC, sqliteSchema: SQLITE_SCHEMA_ROWS });

    const batch = await objects.describeObjects([], "table");

    // Last two by path, in UTF-16 order: the emoji before U+E000.
    expect(batch.details.map((detail) => detail.path).slice(-2)).toEqual([["\u{1F600}"], ["\uE000"]]);
    // And the cut keeps the engine's last, which is the other one: a limit of 11 over the
    // twelve the catalog now holds drops the emoji, not U+E000.
    const bounded = await objects.describeObjects([], "table", 11);
    expect(bounded.details.map((detail) => detail.path)).toContainEqual(["\uE000"]);
    expect(bounded.details.map((detail) => detail.path)).not.toContainEqual(["\u{1F600}"]);
    expect(bounded.truncated?.limit).toBe(11);
  });

  test("a generated column survives the bulk read, as it does the single one", async () => {
    objects = await connectedWithObjects();

    const batch = await objects.describeObjects([], "table");
    const orders = batch.details.find((detail) => detail.path[0] === "orders")!;

    // `pragma_table_xinfo` and not `pragma_table_info`, which DROPS a generated column.
    // `readSchema()` still reads `table_info` and still loses it: the bulk read inherits the
    // object model's catalog and not the flat surface's.
    expect(orders.columns.map((column) => column.name)).toContain("tax");
  });

  test("an expression index publishes no fabricated column name, and an implicit one is not published at all", async () => {
    objects = await connectedWithObjects();

    const batch = await objects.describeObjects([], "table");
    const byName = new Map(batch.details.map((detail) => [detail.path[0], detail]));

    expect(byName.get("orders")!.indexes).toEqual([
      { name: "idx_orders_customer", columns: ["customer_id"], unique: false },
      // `idx_orders_placed` keys `date(placed_at)`, whose `index_info` row carries a null
      // name. The index still appears, with an empty column list.
      { name: "idx_orders_placed", columns: [], unique: false },
    ]);
    // `regions` is WITHOUT ROWID with a composite primary key, so the engine made
    // `sqlite_autoindex_regions_1`, which the same `sqlite_` predicate removes here as in
    // the Indexes folder.
    expect(byName.get("regions")!.indexes).toEqual([]);
  });

  test("a foreign key resolves its parent's key, and a parent with none carries no column", async () => {
    objects = await connectedWithObjects();

    const batch = await objects.describeObjects([], "table");
    const byName = new Map(batch.details.map((detail) => [detail.path[0], detail]));

    // `REFERENCES customers`, no column list: the correlated subquery answers the parent's
    // primary key.
    expect(byName.get("orders")!.foreignKeys).toEqual([
      { columnName: "customer_id", referencedTable: "customers", referencedColumn: "id" },
    ]);
    // `REFERENCES orders(id)`, which names its column, so nothing had to be resolved.
    expect(byName.get("shipments")!.foreignKeys).toEqual([
      { columnName: "order_id", referencedTable: "orders", referencedColumn: "id" },
    ]);
    // `REFERENCES legacy`, whose parent declares no primary key at all. There is nothing to
    // name and the field is empty rather than invented.
    expect(byName.get("legacy_ref")!.foreignKeys).toEqual([
      { columnName: "note", referencedTable: "legacy", referencedColumn: "" },
    ]);
  });

  test("a statement the server refused raises with the server's own sentence", async () => {
    objects = await connectedWithObjects();
    server = (sql) =>
      /JOIN pragma_index_info/.test(sql)
        ? failure("no such table: pragma_index_info", "SQLITE_UNKNOWN")
        : objectServer()(sql, []);

    // One statement of the batch failing is the whole read failing: a batch whose index
    // columns are missing would otherwise answer every index with no columns, which reads
    // as a fact about the tables.
    await expect(objects.describeObjects([], "table")).rejects.toThrow(/no such table: pragma_index_info/);
  });
});

describe("countLibSQLObjects through the transport seam", () => {
  /**
   * The Hrana transport wraps EVERY failure in an Error before it leaves the file, so the
   * non-Error arm of the refusal message is not reachable through it. It is reachable
   * through the seam, which is the point of the seam: `LibSQLTransport` has a documented
   * second implementation in mind (WebSocket, the embedded engine, `@libsql/client`), and
   * a folder that reported `undefined` as the reason it has no number would be worse than
   * the sentence itself.
   */
  const reader = (thrown: unknown): LibSQLObjectReader => ({
    transport: {
      kind: "hrana-http",
      execute: () => Promise.reject(thrown),
      executeBatch: () => Promise.resolve([]),
      serverVersion: () => Promise.resolve(null),
      close: () => Promise.resolve(),
    },
    capabilities: new LibSQLProvider(connection()).getCapabilities(),
    mapError: (error: unknown) => (error instanceof Error ? error : new QueryError(String(error), "libsql")),
  });

  test("a transport that throws something that is not an Error still names a reason", async () => {
    const counts = await countLibSQLObjects(reader("a transport that threw a string"), []);

    expect(counts.table).toEqual({ unavailable: "a transport that threw a string" });
    // The control: an Error's own sentence is carried verbatim rather than stringified,
    // which is what `String(new Error("x"))` would turn into "Error: x".
    const fromError = await countLibSQLObjects(reader(new Error("no such table: pragma_table_list")), []);
    expect(fromError.table).toEqual({ unavailable: "no such table: pragma_table_list" });
  });
});

describe("comparePaths (libsql)", () => {
  // Exported and unit-tested directly, because the two cases that separate this from
  // `JSON.stringify` cannot arise inside ONE kind on an engine whose paths of a kind
  // are all the same length. Standing ruling 5h: written the settled way, and NOT
  // hoisted here - Task 28 replaces the copies with one definition.
  test("orders by segments, so a prefix sorts above what nests under it", () => {
    expect(comparePaths(["orders"], ["orders", "orders_stamp"])).toBeLessThan(0);
    expect(comparePaths(["orders", "orders_stamp"], ["orders"])).toBeGreaterThan(0);
    expect(comparePaths(["orders"], ["orders"])).toBe(0);
  });

  test("orders by code point, so an escaped name is not reordered by its escape", () => {
    expect(comparePaths(['a"b'], ["a\\b"])).toBeLessThan(0);
    expect(comparePaths(["a", "z"], ["b", "a"])).toBeLessThan(0);
  });
});

// ============================================================================
// endOpenQueryTransaction(): the absence is declared, not accidental (D75)
// ============================================================================

describe("endOpenQueryTransaction()", () => {
  /** The only three answers D75 accepts from a provider that does not implement the surface. */
  const ABSENCES = [
    "the engine has no transaction to leave open",
    "the driver cannot be asked",
    "nobody has measured it yet",
  ] as const;

  test("is not implemented, and the doc names WHICH absence that is", () => {
    const provider: DatabaseProvider = new LibSQLProvider(connection());

    expect(provider.endOpenQueryTransaction).toBeUndefined();

    // A boundary nobody wrote down becomes a fallback the next reader trusts, so the
    // absence has to be readable in the doc as well as in the type. Exactly one of the
    // three: "one of these two" is not an answer, and a doc that names none has not
    // declared anything.
    const doc = readFileSync(join(import.meta.dir, "../../../docs/providers/libsql.md"), "utf8");
    expect(doc).toContain("endOpenQueryTransaction");

    // The set that DOES implement the surface is read from the type, never enumerated
    // here: a closed list repeated across the provider docs went stale the day a further
    // provider implemented the surface, which is exactly what happened during D75.
    expect(doc).not.toContain("`postgres`, `sqlite` and `duckdb`");
    expect(ABSENCES.filter((absence) => doc.includes(absence))).toEqual([
      "the engine has no transaction to leave open",
    ]);
  });
});

/**
 * Column defaults as libSQL's catalog reports them (#1029).
 *
 * The provider reads `pragma_table_xinfo`, whose `dflt_value` is the expression AS WRITTEN,
 * identical to SQLite's: a string default arrives quoted with SQL standard doubling, a
 * number or an expression arrives bare. `defaultValue` is the VALUE the column defaults to;
 * `defaultExpression` keeps the text as the engine wrote it.
 */
describe("LibSQLProvider column defaults (#1029)", () => {
  const DEFAULTS_CATALOG: Catalog = {
    tableList: [...TABLE_LIST_ROWS, { schema: "main", name: "column_defaults", type: "table" }],
    sqliteSchema: [...SQLITE_SCHEMA_ROWS, { type: "table", name: "column_defaults", tbl_name: "column_defaults" }],
  };

  const EXPECTED: Record<string, { defaultValue?: string; defaultExpression?: string }> = {
    id: {},
    def_null_string: { defaultValue: "NULL", defaultExpression: "'NULL'" },
    def_text: { defaultValue: "abc", defaultExpression: "'abc'" },
    def_empty: { defaultValue: "", defaultExpression: "''" },
    def_quote: { defaultValue: "it's", defaultExpression: "'it''s'" },
    def_backslash: { defaultValue: "a\\b", defaultExpression: "'a\\b'" },
    def_number: { defaultValue: "42", defaultExpression: "42" },
    def_expression: { defaultValue: "CURRENT_TIMESTAMP", defaultExpression: "CURRENT_TIMESTAMP" },
  };

  const defaultsOf = (columns: readonly { name: string; defaultValue?: string; defaultExpression?: string }[]) =>
    Object.fromEntries(
      columns.map((column) => [
        column.name,
        {
          ...(column.defaultValue === undefined ? {} : { defaultValue: column.defaultValue }),
          ...(column.defaultExpression === undefined ? {} : { defaultExpression: column.defaultExpression }),
        },
      ]),
    );

  let objects: LibSQLProvider;

  afterEach(async () => {
    if (objects?.isConnected()) await objects.disconnect();
  });

  test("a string default is reported as its value, with the catalog text kept alongside", async () => {
    objects = await connectedWithObjects(DEFAULTS_CATALOG);

    const single = await objects.describeObject(["column_defaults"], "table");
    expect(defaultsOf(single.columns)).toEqual(EXPECTED);

    // The bulk read goes through the same mapper; asserting it too is what catches a fix
    // applied to one read and not the other, the mistake #795 had to correct.
    const batch = await objects.describeObjects([], "table");
    const bulk = batch.details.find((detail) => detail.path[0] === "column_defaults")!;
    expect(defaultsOf(bulk.columns)).toEqual(EXPECTED);
  });
});

// ============================================================================
// The agent read-only profile
// ============================================================================

describe("LibSQLProvider agent read-only profile", () => {
  const AGENT: ProviderExecutionContext = { readOnly: true, sendsStatements: true };
  const BUDGET: ReadOnlyStatementBudget = { statementTimeoutMs: 5_000, maxResultRows: 10, maxResultBytes: 4_096 };
  const NAME = '"libredb_agent_probe_[0-9a-f]{32}"';
  const PROBES = [
    new RegExp(`^DROP TABLE IF EXISTS ${NAME}$`),
    new RegExp(`^DELETE FROM ${NAME} WHERE 0$`),
    new RegExp(`^INSERT INTO ${NAME} SELECT 1 WHERE 0$`),
  ];
  /** A token shaped like a JWT, built here so no literal token sits in the source; the signature is not one. */
  const fakeToken = (claims: unknown): string => {
    const part = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url");
    return `${part({ alg: "EdDSA", typ: "JWT" })}.${part(claims)}.not-a-real-signature`;
  };
  const READ_ONLY_TOKEN = fakeToken({ a: "ro" });
  /** The body sqld answers a read-only token's write with, for the whole request (measured on sqld 0.24.33). */
  const FORBIDDEN = {
    error: "Authorization forbidden: Current session doesn't not have Write permission to namespace default",
  };

  /**
   * A server whose token is read-only: a request carrying a write is answered 403 as a whole, everything else as the
   * fixture answers it. `write` decides what a write statement is, the way sqld classifies one before it runs.
   */
  function readOnlyToken(
    write: (sql: string) => boolean = (sql) => /^\s*(DROP|INSERT|DELETE|UPDATE|CREATE)\b/i.test(sql),
  ): void {
    const answering = globalThis.fetch;
    globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => {
      const body = init?.body === undefined ? "" : String(init.body);
      const requests = (JSON.parse(body || "{}") as { requests?: { stmt?: { sql?: string } }[] }).requests ?? [];
      if (requests.some((request) => write(String(request.stmt?.sql ?? "")))) {
        calls.push({ url: String(input), body });
        return Promise.resolve(new Response(JSON.stringify(FORBIDDEN), { status: 403 }));
      }
      return answering(input, init);
    }) as unknown as typeof fetch;
  }

  async function opened(
    overrides: Partial<DatabaseConnection> = {},
    execution: ProviderExecutionContext = AGENT,
  ): Promise<LibSQLProvider> {
    const provider = new LibSQLProvider(connection({ password: READ_ONLY_TOKEN, ...overrides }), {}, execution);
    await provider.connect();
    return provider;
  }

  async function refusal(run: () => Promise<unknown>): Promise<Error> {
    try {
      await run();
    } catch (error) {
      return error as Error;
    }
    throw new Error("expected a refusal");
  }

  test("declares the token form for the agent's own credential", () => {
    expect(new LibSQLProvider(connection()).getCapabilities().agentCredential).toBe("token");
  });

  test("opens under the profile when every probe is refused for the token, after the connect probe", async () => {
    readOnlyToken();
    const provider = await opened();

    const sent = sentStatements();
    expect(sent).toHaveLength(4);
    expect(sent[0]).toBe("SELECT 1");
    PROBES.forEach((probe, index) => expect(sent[index + 1]).toMatch(probe));
    expect(provider.isConnected()).toBe(true);
    await provider.disconnect();
  });

  test("refuses a server that lets a read-only token write, and stays closed", async () => {
    // No 403 anywhere: the fixture runs the probe as it runs every statement, as a server that checks no token does.
    const provider = new LibSQLProvider(connection({ password: READ_ONLY_TOKEN }), {}, AGENT);

    const error = await refusal(() => provider.connect());

    expect(error).toBeInstanceOf(ExecutionProfileError);
    expect((error as ExecutionProfileError).reasonCode).toBe("PROFILE_SERVER_UNENFORCED");
    expect(error.message).toContain("so the server does not hold the token to them");
    expect(provider.isConnected()).toBe(false);
    await expect(provider.queryReadOnly("SELECT 1", BUDGET)).rejects.toThrow();
  });

  test("refuses a user name and password, and a connection with no token, before sending anything", async () => {
    for (const overrides of [{ user: "admin", password: "not-a-real-password" }, { password: undefined }]) {
      calls = [];
      const provider = new LibSQLProvider(connection(overrides), {}, AGENT);

      expect(await refusal(() => provider.connect())).toBeInstanceOf(ExecutionProfileError);
      expect(calls).toEqual([]);
    }
  });

  test("a connection string's token is the agent's credential, its user ignored as on every URL", async () => {
    readOnlyToken();
    const provider = await opened({
      host: undefined,
      port: undefined,
      user: "ignored",
      password: undefined,
      connectionString: `libsql://db.example.invalid?authToken=${READ_ONLY_TOKEN}`,
    });

    expect(provider.isConnected()).toBe(true);
    await provider.disconnect();
  });

  test("refuses a token that can write as one, before sending anything with it", async () => {
    readOnlyToken();
    for (const password of [fakeToken({ a: "rw" }), fakeToken({ iat: 1 }), "not-a-real-token"]) {
      calls = [];
      const provider = new LibSQLProvider(connection({ password }), {}, AGENT);

      const error = await refusal(() => provider.connect());

      expect((error as ExecutionProfileError).reasonCode).toBe("PROFILE_PRIVILEGES_TOO_BROAD");
      expect(error.message).toBe(
        'The libSQL token is not a read-only token (a JWT whose claims carry "a": "ro"), so it can write, and the agent read-only profile is refused: use a read-only token',
      );
      expect(provider.isConnected()).toBe(false);
      expect(calls).toEqual([]);
    }
  });

  test("refuses a read-only token that carries another claim, before sending anything with it", async () => {
    readOnlyToken();
    for (const password of [fakeToken({ a: "ro", p: { rw: { ns: ["default"] } } }), fakeToken({ a: "ro", perm: [] })]) {
      calls = [];
      const provider = new LibSQLProvider(connection({ password }), {}, AGENT);

      const error = await refusal(() => provider.connect());

      expect((error as ExecutionProfileError).reasonCode).toBe("PROFILE_PRIVILEGES_TOO_BROAD");
      expect(error.message).toContain("so it can write, and the agent read-only profile is refused");
      expect(calls).toEqual([]);
    }
  });

  /**
   * A local server answering as Turso Cloud does (measured): every request 200, and a write sent with a read-only token
   * refused in its own step, code BLOCKED, while a read answers. `refusal` is the step's message. No token reaches
   * Turso: the server is on 127.0.0.1 and the token is a fake one.
   */
  async function tursoShapedServer(refusalMessage: string): Promise<{ readonly port: number; close(): Promise<void> }> {
    const server = createServer((request, response) => {
      let body = "";
      request.on("data", (chunk: Buffer) => {
        body += chunk.toString("utf8");
      });
      request.on("end", () => {
        const requests = (JSON.parse(body || "{}") as { requests?: { type: string; stmt?: { sql?: string } }[] })
          .requests;
        const results = (requests ?? []).map((step) => {
          if (step.type !== "execute") return { type: "ok", response: { type: step.type } };
          const sql = String(step.stmt?.sql ?? "");
          if (/^\s*(DROP|INSERT|DELETE|UPDATE|CREATE)\b/i.test(sql)) {
            return { type: "error", error: { message: refusalMessage, code: "BLOCKED" } };
          }
          return result([["1", null]], [[int(1)]]);
        });
        response.writeHead(200, { "content-type": "application/json" });
        response.end(JSON.stringify({ baton: null, base_url: null, results }));
      });
    });
    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", () => resolve()));
    const port = (server.address() as AddressInfo).port;
    return { port, close: () => new Promise<void>((resolve) => server.close(() => resolve())) };
  }

  test("opens on a Turso Cloud shaped server that refuses every probe in-band for the token", async () => {
    globalThis.fetch = originalFetch;
    const turso = await tursoShapedServer(
      "Operation was blocked: SQL write operations are forbidden (current session doesn't have write permission)",
    );
    try {
      const provider = new LibSQLProvider(
        connection({ port: turso.port, password: fakeToken({ a: "ro", iat: 1, id: "x", kid: "x", rid: "x" }) }),
        {},
        AGENT,
      );
      await provider.connect();

      expect(provider.isConnected()).toBe(true);
      expect((await provider.queryReadOnly("SELECT 1", BUDGET)).rowCount).toBe(1);
      const error = await refusal(() => provider.queryReadOnly("DELETE FROM probe_orders", BUDGET));
      expect(error).toBeInstanceOf(QueryError);
      expect(error.message).toContain(
        "SQL write operations are forbidden (current session doesn't have write permission)",
      );
      await provider.disconnect();
    } finally {
      await turso.close();
    }
  });

  test("refuses Turso's answer for a database whose writes are blocked, which a token that can write gets", async () => {
    globalThis.fetch = originalFetch;
    const turso = await tursoShapedServer(
      "Operation was blocked: SQL write operations are forbidden (writes are blocked, do you need to upgrade your plan?)",
    );
    try {
      const provider = new LibSQLProvider(connection({ port: turso.port, password: READ_ONLY_TOKEN }), {}, AGENT);

      const error = await refusal(() => provider.connect());

      expect((error as ExecutionProfileError).reasonCode).toBe("PROFILE_UNSUPPORTED_TARGET");
      expect(provider.isConnected()).toBe(false);
    } finally {
      await turso.close();
    }
  });

  test("refuses a 403 whose body is not the server's authorization refusal", async () => {
    globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => {
      const body = init?.body === undefined ? "" : String(init.body);
      if (body.includes("libredb_agent_probe_")) {
        return Promise.resolve(new Response("<html>403 Forbidden</html>", { status: 403 }));
      }
      return Promise.resolve(
        new Response(
          JSON.stringify({ baton: null, base_url: null, results: [result([["1", null]], [[int(1)]]), { type: "ok" }] }),
          {
            status: 200,
          },
        ),
      );
    }) as unknown as typeof fetch;
    const provider = new LibSQLProvider(connection({ password: READ_ONLY_TOKEN }), {}, AGENT);

    const error = await refusal(() => provider.connect());

    expect((error as ExecutionProfileError).reasonCode).toBe("PROFILE_UNSUPPORTED_TARGET");
  });

  test("an editor handle opens unproven and queryReadOnly refuses on it before sending anything", async () => {
    const provider = await opened({}, {});
    expect(sentStatements()).toEqual(["SELECT 1"]);
    calls = [];

    await expect(provider.queryReadOnly("SELECT 1", BUDGET)).rejects.toThrow(
      "Read-only execution requires a provider opened under the agent read-only profile",
    );
    expect(calls).toEqual([]);
    await provider.disconnect();
  });

  test("the operations profile sends no statement of the model's, so it opens unproven and has no read path", async () => {
    const provider = await opened(
      { user: "admin", password: "not-a-real-password" },
      { readOnly: true, sendsStatements: false },
    );
    expect(sentStatements()).toEqual(["SELECT 1"]);

    await expect(provider.queryReadOnly("SELECT 1", BUDGET)).rejects.toThrow(
      "Read-only execution requires a provider opened under the agent read-only profile",
    );
    await provider.disconnect();
  });

  test("sends the statement alone, in one request, and answers its rows", async () => {
    readOnlyToken();
    const provider = await opened();
    calls = [];

    const result = await provider.queryReadOnly("SELECT name FROM sqlite_master WHERE type = 'table'", BUDGET);

    expect(result.rows).toEqual([{ name: "probe_customers" }, { name: "probe_orders" }]);
    expect(result.rowCount).toBe(2);
    expect(calls).toHaveLength(1);
    const requests = (JSON.parse(String(calls[0]?.body)) as { requests: { type: string }[] }).requests;
    expect(requests.map((request) => request.type)).toEqual(["execute", "close"]);
    await provider.disconnect();
  });

  test("a write the server refuses for the token is a statement refusal carrying the server's words", async () => {
    readOnlyToken();
    const provider = await opened();

    const error = await refusal(() => provider.queryReadOnly("DELETE FROM probe_orders", BUDGET));

    expect(error).toBeInstanceOf(QueryError);
    expect(error).not.toBeInstanceOf(AuthenticationError);
    expect(error.message).toContain("Authorization forbidden");
    await provider.disconnect();
  });

  test("refuses a budget that is not one, and the estimate-plan session mode, before sending anything", async () => {
    readOnlyToken();
    const provider = await opened();
    calls = [];

    await expect(provider.queryReadOnly("SELECT 1", { ...BUDGET, maxResultRows: 0 })).rejects.toThrow(
      "Read-only execution budget field maxResultRows must be a positive integer",
    );
    await expect(provider.queryReadOnly("SELECT 1", BUDGET, "estimate-plan")).rejects.toThrow(
      "libSQL composes its estimating plan as a statement, so the estimate-plan session mode is refused",
    );
    expect(calls).toEqual([]);
    await provider.disconnect();
  });

  test("refuses a result past the row budget and one past the byte budget", async () => {
    readOnlyToken();
    const provider = await opened();

    await expect(
      provider.queryReadOnly("SELECT name FROM sqlite_master WHERE type = 'table'", { ...BUDGET, maxResultRows: 1 }),
    ).rejects.toThrow("Read-only execution exceeded the row budget: 2 rows > 1 allowed");
    await expect(
      provider.queryReadOnly("SELECT name FROM sqlite_master WHERE type = 'table'", { ...BUDGET, maxResultBytes: 10 }),
    ).rejects.toThrow("Read-only execution exceeded the byte budget");
    await provider.disconnect();
  });

  test("stops reading an answer past the wire cap and refuses it as over the byte budget", async () => {
    readOnlyToken();
    const provider = await opened();
    // An answer far past the cap that is not JSON: only the cap can turn it into a byte-budget refusal, and the stream
    // ends on its own, so a read with no cap fails this test in moments rather than hanging on it.
    let cancelled = false;
    let sent = 0;
    const chunks = 64;
    const chunk = new TextEncoder().encode("x".repeat(64 * 1024));
    globalThis.fetch = (() =>
      Promise.resolve(
        new Response(
          new ReadableStream<Uint8Array>({
            pull(controller) {
              if (sent === chunks) {
                controller.close();
                return;
              }
              sent++;
              controller.enqueue(chunk);
            },
            cancel() {
              cancelled = true;
            },
          }),
          { status: 200 },
        ),
      )) as unknown as typeof fetch;

    const error = await refusal(() => provider.queryReadOnly("SELECT big", { ...BUDGET, maxResultBytes: 1 }));

    expect(error).toBeInstanceOf(QueryError);
    const counted = /^Read-only execution exceeded the byte budget: (\d+) bytes > 1 allowed$/.exec(error.message);
    // Six bytes for the budget and the 1 MiB envelope, stopped at the 64 KiB chunk that crossed them.
    expect(Number(counted?.[1])).toBe(17 * 64 * 1024);
    expect(cancelled).toBe(true);
    expect(sent).toBeLessThan(chunks);
    await provider.disconnect();
  });

  test("refuses a statement its deadline ended as over the time budget", async () => {
    readOnlyToken();
    const provider = await opened();
    globalThis.fetch = ((_input: RequestInfo | URL, init?: RequestInit) =>
      new Promise((_resolve, reject) => {
        init?.signal?.addEventListener("abort", () => reject(init.signal?.reason));
      })) as unknown as typeof fetch;

    await expect(provider.queryReadOnly("SELECT 1", { ...BUDGET, statementTimeoutMs: 5 })).rejects.toThrow(
      "Read-only execution exceeded the time budget: no answer within 5ms",
    );
    await provider.disconnect();
  });

  test("an unreachable server is still a connection failure", async () => {
    readOnlyToken();
    const provider = await opened();
    globalThis.fetch = (() => Promise.reject(new Error("connect ECONNREFUSED"))) as unknown as typeof fetch;

    expect(await refusal(() => provider.queryReadOnly("SELECT 1", BUDGET))).toBeInstanceOf(ConnectionError);
    await provider.disconnect();
  });
});

// ============================================================================
// MCP run_read_query on libSQL
// ============================================================================

/**
 * MCP's `run_read_query` serves libSQL through the factory's own `agent-read-only` acquisition and the provider's
 * `queryReadOnly`, the path agent mode uses: the open proves the token read-only (its claims, then three writes the
 * server must refuse), and every statement then goes alone in one request under that token. The MCP fence refuses a
 * write-shaped statement before anything is acquired; a write that reaches the server, its fence removed, is the
 * server's 403. The seed's own token is the credential here, as it is on a seed that sets no agent token.
 */
describe("MCP run_read_query on a libSQL connection", () => {
  const fakeToken = (claims: unknown): string => {
    const part = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url");
    return `${part({ alg: "EdDSA", typ: "JWT" })}.${part(claims)}.not-a-real-signature`;
  };
  const READ_ONLY_TOKEN = fakeToken({ a: "ro" });
  const NAME = '"libredb_agent_probe_[0-9a-f]{32}"';
  const PROBES = [
    new RegExp(`^DROP TABLE IF EXISTS ${NAME}$`),
    new RegExp(`^DELETE FROM ${NAME} WHERE 0$`),
    new RegExp(`^INSERT INTO ${NAME} SELECT 1 WHERE 0$`),
  ];
  const FORBIDDEN = "Authorization forbidden: Current session doesn't not have Write permission to namespace default";
  const ENGINES = RUN_READ_QUERY_ENGINES;

  function seed(overrides: Partial<ManagedConnection> = {}): ManagedConnection {
    return {
      id: "seed:edge",
      name: "edge",
      type: "libsql",
      seedId: "edge",
      managed: true,
      mcp: true,
      roles: ["*"],
      host: "127.0.0.1",
      port: 18081,
      password: READ_ONLY_TOKEN,
      createdAt: new Date("2026-11-01T00:00:00.000Z"),
      ...overrides,
    } as ManagedConnection;
  }

  let acquired: ExecutionProfile[] = [];

  /** A context over one seed, acquiring through the factory as the MCP route does. */
  function contextFor(target: ManagedConnection): McpConnectionContext {
    return {
      caller: { username: "alice", role: "admin" },
      visibleConnections: async () => [target],
      resolve: async () => target,
      acquire: (connection: ManagedConnection, profile: ExecutionProfile) => {
        acquired.push(profile);
        return acquireExecutionProfileProvider(connection, profile);
      },
    } as unknown as McpConnectionContext;
  }

  function call(args: Record<string, unknown>, target: ManagedConnection = seed()) {
    const parsed = RunReadQueryInputSchema.parse({ connection_id: target.id, ...args }) as McpStatementInput;
    return runReadQuery(parsed, { context: contextFor(target), signal: new AbortController().signal });
  }

  /** The same tool with its statement fence removed, so a write reaches the boundary under test. */
  const UNFENCED: McpStatementArm = { ...mcpStatementArm("run_read_query"), fence: () => null };

  function unfenced(args: Record<string, unknown>, target: ManagedConnection = seed()) {
    const parsed = UNFENCED.inputSchema.parse({ connection_id: target.id, ...args }) as McpStatementInput;
    return runStatement(UNFENCED, parsed, { context: contextFor(target), signal: new AbortController().signal });
  }

  /** A server that refuses a request carrying a write with 403 as a whole, as sqld does for a read-only token. */
  function readOnlyServer(answer: Server = answerFor): void {
    server = answer;
    const answering = globalThis.fetch;
    globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => {
      const body = init?.body === undefined ? "" : String(init.body);
      const requests = (JSON.parse(body || "{}") as { requests?: { stmt?: { sql?: string } }[] }).requests ?? [];
      if (requests.some((request) => /^\s*(DROP|INSERT|DELETE|UPDATE|CREATE)\b/i.test(String(request.stmt?.sql)))) {
        calls.push({ url: String(input), body });
        return Promise.resolve(new Response(JSON.stringify({ error: FORBIDDEN }), { status: 403 }));
      }
      return answering(input, init);
    }) as unknown as typeof fetch;
  }

  const ids = (count: number) =>
    result(
      [["id", "INTEGER"]],
      Array.from({ length: count }, (_, index) => [int(index + 1)]),
    );

  beforeEach(() => {
    acquired = [];
  });

  afterEach(async () => {
    await clearProviderCache();
  });

  test("the provider doc says MCP read is open here and names the statements it runs, from their records", () => {
    const doc = readFileSync(join(import.meta.dir, "../../../docs/providers/libsql.md"), "utf8");

    expect(MCP_STATEMENT_READ.libsql).toBe(true);
    expect(doc).toContain("`MCP_STATEMENT_READ.libsql` is true");
    expect(doc).not.toContain("MCP's statement read stays closed on libSQL");
    expect(doc).not.toContain("| No MCP statement read |");
    expect(doc.replaceAll("`", "")).toContain(`It runs ${readQueryFormsOf("libsql").statements}:`);
  });

  test("reads rows under the agent profile after the open's proof, paged one row past max_rows", async () => {
    readOnlyServer((sql) => (sql === "SELECT 1" ? answerFor(sql) : ids(3)));

    const result = await call({ sql: "SELECT id FROM probe_orders ORDER BY id", max_rows: 2 });

    expect(result.isError).toBeUndefined();
    expect(acquired).toEqual(["agent-read-only"]);
    const sent = sentStatements();
    expect(sent).toHaveLength(5);
    expect(sent[0]).toBe("SELECT 1");
    PROBES.forEach((probe, index) => expect(sent[index + 1]).toMatch(probe));
    expect(sent[4]).toBe("SELECT id FROM probe_orders ORDER BY id LIMIT 3");
    expect(result.structuredContent).toMatchObject({
      connection_id: "seed:edge",
      columns: [{ name: "id", type: "INTEGER" }],
      rows: [{ id: 1 }, { id: 2 }],
      row_count: 2,
      truncated: true,
      truncated_by: "max_rows",
      pagination: { offset: 0, limit: 2, hasMore: true, nextOffset: 2, wasLimited: true },
      hint: "Call again with offset 2 for the next page.",
    });
    expect((result.content as { text: string }[])[0]?.text).toBe(MCP_UNTRUSTED_NOTICE);
  });

  test("the proof runs at the open of the shared agent handle, not before each statement", async () => {
    readOnlyServer((sql) => (sql === "SELECT 1" ? answerFor(sql) : ids(1)));

    await call({ sql: "SELECT id FROM probe_orders" });
    calls = [];
    const second = await call({ sql: "SELECT id FROM probe_orders" });

    expect(second.isError).toBeUndefined();
    expect(sentStatements()).toEqual(["SELECT id FROM probe_orders LIMIT 101"]);
  });

  test("pages with libSQL's own LIMIT n OFFSET m", async () => {
    readOnlyServer((sql) => (sql === "SELECT 1" ? answerFor(sql) : ids(1)));

    const result = await call({ sql: "SELECT id FROM probe_orders ORDER BY id", max_rows: 2, offset: 2 });

    expect(sentStatements().at(-1)).toBe("SELECT id FROM probe_orders ORDER BY id LIMIT 3 OFFSET 2");
    expect(result.structuredContent).toMatchObject({
      rows: [{ id: 1 }],
      pagination: { offset: 2, limit: 2, hasMore: false, nextOffset: null, wasLimited: true },
    });
  });

  test("refuses an offset on a statement it cannot page, before anything is acquired", async () => {
    readOnlyServer();

    for (const [sql, text] of [
      ["VALUES (1), (2), (3)", "This statement's result cannot be paged. Call again without offset."],
      [
        "SELECT id FROM probe_orders LIMIT 5",
        "This query has its own LIMIT or TOP, so it cannot be paged with offset. Remove it and page with offset, or page it in SQL: LIMIT n OFFSET m.",
      ],
    ] as const) {
      const result = await call({ sql, offset: 1 });

      expect(result).toEqual({ content: [{ type: "text", text }], isError: true });
      expect(sentStatements().filter((sent) => sent.includes("probe_orders") || sent.startsWith("VALUES"))).toEqual([]);
    }
  });

  const fenced = (code: string) =>
    `The statement was refused before it reached the database (${code}). On this engine run_read_query runs one read-only statement: a SELECT (a WITH is fine), VALUES, or EXPLAIN.`;

  test.each([
    ["INSERT INTO probe_orders (id) VALUES (9)", "NON_READ_STATEMENT"],
    ["INSERT INTO probe_orders (id) VALUES (9) RETURNING id", "NON_READ_STATEMENT"],
    ["REPLACE INTO probe_orders (id) VALUES (1)", "NON_READ_STATEMENT"],
    ["UPDATE probe_orders SET id = 0", "NON_READ_STATEMENT"],
    ["DELETE FROM probe_orders", "NON_READ_STATEMENT"],
    ["DROP TABLE probe_orders", "NON_READ_STATEMENT"],
    ["CREATE TABLE probe_copy AS SELECT * FROM probe_orders", "NON_READ_STATEMENT"],
    ["ATTACH DATABASE ':memory:' AS other", "NON_READ_STATEMENT"],
    ["DETACH DATABASE other", "NON_READ_STATEMENT"],
    ["VACUUM INTO '/tmp/not-a-real-copy.db'", "NON_READ_STATEMENT"],
    ["REINDEX", "NON_READ_STATEMENT"],
    ["PRAGMA user_version = 5", "NON_READ_STATEMENT"],
    ["PRAGMA writable_schema = ON", "NON_READ_STATEMENT"],
    ["PRAGMA table_info(probe_orders)", "NON_READ_STATEMENT"],
    ["BEGIN IMMEDIATE", "NON_READ_STATEMENT"],
    ["COMMIT", "NON_READ_STATEMENT"],
    ["SELECT 1; DELETE FROM probe_orders", "MULTIPLE_STATEMENTS"],
    [
      "WITH doomed AS (SELECT 1 AS id) DELETE FROM probe_orders WHERE id IN (SELECT id FROM doomed)",
      "NON_READ_STATEMENT",
    ],
    ["SELECT load_extension('/tmp/not-a-real-extension')", "SIDE_EFFECT_KEYWORD"],
    ["EXPLAIN DELETE FROM probe_orders", "SIDE_EFFECT_KEYWORD"],
  ])("%p is refused at admission, before anything is acquired or sent", async (sql, code) => {
    readOnlyServer();

    const result = await call({ sql });

    expect(result).toEqual({ content: [{ type: "text", text: fenced(code) }], isError: true });
    expect(acquired).toEqual([]);
    expect(calls).toEqual([]);
  });

  test("a write that reaches the server, its fence removed, is answered as the server's own refusal", async () => {
    readOnlyServer();

    const result = await unfenced({ sql: "DELETE FROM probe_orders" });

    expect(acquired).toEqual(["agent-read-only"]);
    expect(result).toEqual({
      content: [
        { type: "text", text: MCP_UNTRUSTED_NOTICE },
        { type: "text", text: `${MCP_ENGINE_ERROR_PREFIX}libSQL request failed: ${FORBIDDEN}` },
      ],
      isError: true,
    });
  });

  const refusedAtOpen = (sentence: string) => ({
    content: [
      {
        type: "text" as const,
        text: `run_read_query cannot run on this connection: ${sentence}. It runs on ${ENGINES}.`,
      },
    ],
    isError: true as const,
  });

  const CAN_WRITE =
    'The libSQL token is not a read-only token (a JWT whose claims carry "a": "ro"), so it can write, and the agent read-only profile is refused: use a read-only token';

  test.each([
    ["a writable token", { password: fakeToken({ a: "rw" }) }, CAN_WRITE],
    [
      "a read-only token whose permission claim widens it",
      { password: fakeToken({ a: "ro", p: { rw: { ns: ["default"] } } }) },
      'The libSQL token carries claims beyond the read-only marker and the standard ones ("p"), which the server may read to grant writes, so it can write, and the agent read-only profile is refused: use a read-only token that carries no other claim',
    ],
    [
      "a user name and password (HTTP Basic)",
      { user: "admin", password: "not-a-real-password" },
      "A libSQL connection that signs in with a user name and password (HTTP Basic) has no read-only form, so the agent read-only profile is refused: use a read-only token instead",
    ],
    [
      "no token",
      { password: undefined },
      "A libSQL connection with no token reaches only a server that checks no credential, so the agent read-only profile is refused: use a read-only token",
    ],
    [
      "a connection string carrying a writable token",
      {
        host: undefined,
        port: undefined,
        password: undefined,
        connectionString: `libsql://db.example.invalid?authToken=${fakeToken({ a: "rw" })}`,
      },
      CAN_WRITE,
    ],
  ] as const)(
    "%s is refused at open, in the tool's sentence, before anything is sent",
    async (_label, overrides, sentence) => {
      readOnlyServer();

      const result = await call({ sql: "SELECT id FROM probe_orders" }, seed(overrides as Partial<ManagedConnection>));

      expect(acquired).toEqual(["agent-read-only"]);
      expect(result).toEqual(refusedAtOpen(sentence));
      expect(calls).toEqual([]);
    },
  );

  test("a server that runs the open's probes, one that checks no token, is refused before the statement", async () => {
    // No 403 anywhere: the fixture runs every statement, as a server with no credential check does.
    const result = await call({ sql: "SELECT id FROM probe_orders" });

    expect(acquired).toEqual(["agent-read-only"]);
    expect(sentStatements().some((sent) => sent.startsWith("SELECT id FROM probe_orders"))).toBe(false);
    const text = (result.content as { text: string }[])[0]?.text ?? "";
    expect(result.isError).toBe(true);
    expect(text).toMatch(
      new RegExp(
        `^run_read_query cannot run on this connection: libSQL ran the write probe DROP TABLE IF EXISTS ${NAME} sent with a token whose claims declare read-only access, so the server does not hold the token to them \\(it checks no token, or admits one it did not sign\\), and the agent read-only profile is refused\\. It runs on ${ENGINES.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}\\.$`,
      ),
    );
    expect(text).not.toContain(READ_ONLY_TOKEN);
  });

  test("an agent credential beside a connection string is refused before the provider is built", async () => {
    const target = seed({
      host: undefined,
      port: undefined,
      password: undefined,
      connectionString: `libsql://db.example.invalid?authToken=${READ_ONLY_TOKEN}`,
      agentPassword: READ_ONLY_TOKEN,
    } as Partial<ManagedConnection>);

    const result = await call({ sql: "SELECT 1" }, target);

    expect(result).toEqual(
      refusedAtOpen(
        'Connection "seed:edge" configures an agent credential alongside a connection string; the credential cannot be applied, so acquisition is refused',
      ),
    );
    expect(calls).toEqual([]);
  });

  test("a result past the 1000-row provider budget is refused, never cut", async () => {
    readOnlyServer((sql) => (sql === "SELECT 1" ? answerFor(sql) : ids(1_001)));

    const result = await call({ sql: "SELECT id FROM probe_orders LIMIT 2000" });

    expect(result).toEqual({
      content: [
        {
          type: "text",
          text: "The result has more than 1000 rows, the most this server reads for one call. Add a LIMIT of 1000 or less, or remove your own LIMIT and page with offset.",
        },
      ],
      isError: true,
    });
  });

  test("a result past the 1 MiB provider budget is refused, never cut", async () => {
    readOnlyServer((sql) =>
      sql === "SELECT 1" ? answerFor(sql) : result([["s", "TEXT"]], [[text("x".repeat(1_048_577))]]),
    );

    const answer = await call({ sql: "SELECT s FROM probe_wide" });

    expect(answer).toEqual({
      content: [
        {
          type: "text",
          text: "The result is larger than 1 MiB, the most this server reads for one call. Select fewer or narrower columns, or add a smaller LIMIT.",
        },
      ],
      isError: true,
    });
  });

  test("a statement past timeout_ms is answered as the timeout, and its request is closed at the deadline", async () => {
    readOnlyServer();
    const answering = globalThis.fetch;
    let closed = false;
    globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => {
      if (!String(init?.body).includes("recursive_count")) return answering(input, init);
      return new Promise<Response>((_resolve, reject) => {
        init?.signal?.addEventListener(
          "abort",
          () => {
            closed = true;
            reject(new Error("aborted"));
          },
          { once: true },
        );
      });
    }) as unknown as typeof fetch;

    const answer = await call({ sql: "SELECT count(*) AS recursive_count FROM probe_orders", timeout_ms: 500 });

    expect(answer).toEqual({
      content: [
        {
          type: "text",
          text: "The statement did not finish within timeout_ms (500 ms). Narrow it, or raise timeout_ms up to 30000.",
        },
      ],
      isError: true,
    });
    await Bun.sleep(50);
    expect(closed).toBe(true);
  });
});