engine.ts

// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.

import { AsyncLocalStorage } from 'node:async_hooks';
import { QueryAST, HookContext, ServiceObject } from '@objectstack/spec/data';
import {
  EngineQueryOptions,
  DataEngineInsertOptions,
  EngineUpdateOptions,
  EngineDeleteOptions,
  EngineAggregateOptions,
  EngineCountOptions,
  RPC_QUERY_ALIAS_SLOTS,
  foldQueryAliasSlots,
  QUERY_CURSOR_REMOVED,
  QUERY_DISTINCT_REMOVED,
  type QueryAliasSlot,
  type DroppedFieldsEvent
} from '@objectstack/spec/data';
import type { WriteObservabilityOptions } from '@objectstack/spec/contracts';
import { parseAutonumberFormat, renderAutonumber, missingFieldValues, isTenancyDisabled, FILE_REFERENCE_TYPES, REFERENCE_VALUE_TYPES, referenceTargetOf, isFileIdToken, RAW_FILE_VALUES_CONTEXT_KEY, isCurrentUserDefaultToken } from '@objectstack/spec/data';
import {
  DATA_MIGRATION_FLAG_OBJECT,
  FILE_REFERENCES_MIGRATION_ID,
  VALUE_SHAPES_MIGRATION_ID,
  isDataMigrationFlagVerified,
} from '@objectstack/spec/system';
import { ExecutionContext, ExecutionContextInput, ExecutionContextSchema } from '@objectstack/spec/kernel';
import type { FlowFunctionEffect } from '@objectstack/spec/automation';
import {
  IDataDriver,
  IDataEngine,
  type IObjectQLEngine,
  Logger,
  createLogger,
  withTransientRetry,
  type RetryOptions,
  filterTokenContextFrom,
  resolveFilterTokens,
} from '@objectstack/core';
import { SummaryRecomputeError, type SummaryRecomputeFailure } from './summary-errors.js';
import {
  DriverConnectError,
  DatasourceUnavailableError,
  emitDegradedBootBanner,
  type DriverConnectFailure,
  type DriverHealth,
  type DatasourceUnavailableInfo,
  type DatasourceUnavailableKind,
} from './driver-connect-errors.js';
import { resolveAllowDriverConnectFailure } from '@objectstack/types';

/**
 * Per-row outcome of {@link ObjectQL.insertMany} (framework#3172). One entry
 * per input row, in input order: written rows carry the after-hook record,
 * failed rows carry the per-row error (validation / autonumber / encryption).
 */
export type InsertManyRowOutcome =
  | { ok: true; record: any }
  | { ok: false; error: unknown };
import { CoreServiceName, StorageNameMapping } from '@objectstack/spec/system';
import { IRealtimeService, RealtimeEventPayload } from '@objectstack/spec/contracts';
import {
  BulkDataEventSchema,
  DataEventSchema,
  type BulkDataEvent,
  type DataEvent,
} from '@objectstack/spec/api';
import type { ICryptoProvider, CryptoHandle } from '@objectstack/spec/contracts';
import {
  collectSecretFields,
  collectMaskedReadFields,
  collectCredentialFields,
  makeSecretRef,
  parseSecretRef,
  isSecretRef,
  SECRET_MASK,
} from './secret-fields.js';
import { pluralToSingular, ExternalWriteForbiddenError } from '@objectstack/spec/shared';
import { SchemaRegistry, computeFQN } from './registry.js';
import { expandSearchToFilter } from './search-filter.js';
import { ExpressionEngine } from '@objectstack/formula';
import type { Expression } from '@objectstack/spec';
import { isAggregatedViewContainer, expandViewContainer } from '@objectstack/spec';
import { bindHooksToEngine } from './hook-binder.js';
import { validateRecord, normalizeMultiValueFields, coerceBooleanFields, ValidationError, buildFieldError, valueShapePostureSetByEnv, mediaPostureSetByEnv, isScannableValueShapeField } from './validation/record-validator.js';
import { evaluateValidationRules, needsPriorRecord, stripReadonlyWhenFields, stripReadonlyWhenFieldsMulti, hasReadonlyWhenInPayload, stripReadonlyFields } from './validation/rule-validator.js';
import { applyInMemoryAggregation } from './in-memory-aggregation.js';
import { applyHaving } from './having-filter.js';
import {
  auditDanglingReferences,
  type AuditableObject,
  type DanglingReferenceAuditOptions,
  type DanglingReferenceReport,
} from './integrity/dangling-reference-audit.js';

/**
 * The lifecycle events the engine actually dispatches via `triggerHooks`. This
 * is the single source of truth for what a hook can subscribe to — kept in
 * lockstep with the `triggerHooks(...)` call sites and with `HookEvent` in the
 * spec. `beforeFind`/`afterFind` cover both `find` and `findOne`; the write
 * events cover both single-id and bulk (`multi: true`) writes (#3195). A hook
 * subscribing to anything outside this set would silently never fire, so
 * `registerHook` warns rather than accepting it blindly.
 */
const DISPATCHABLE_HOOK_EVENTS: ReadonlySet<string> = new Set([
  'beforeFind', 'afterFind',
  'beforeInsert', 'afterInsert',
  'beforeUpdate', 'afterUpdate',
  'beforeDelete', 'afterDelete',
]);

/**
 * [#4346] The alias slots the ENGINE option bags still admit, cut from the
 * spec's own table (#3795) so the engine never re-declares a mapping.
 *
 * The deprecated `DataEngine{Query,Update,Delete,Count,Aggregate}Options`
 * contracts declare `filter` on every read AND write method, but only `find`
 * folded it — `findOne`/`count`/`update`/`delete`/`aggregate` passed the bag
 * through with `where === undefined`, which every driver reads as "no
 * predicate": a caller filtering with `{ filter }` silently matched EVERY row
 * (an over-grant on the reads, an unbounded write on `update`/`delete`).
 *
 * `where` is the slot every method folds; `limit` additionally applies to the
 * find-shaped bags, which declare `top` (OData) as its alias. The other four
 * pairs in the table are RPC/wire spellings folded at parse by
 * `RpcQueryOptionsSchema` / the protocol normalizer — their values need shape
 * lowering (`sort` records, `populate` lists) that belongs to those layers, so
 * the engine deliberately does not fold them.
 */
const ENGINE_WHERE_SLOTS: readonly QueryAliasSlot[] =
  RPC_QUERY_ALIAS_SLOTS.filter((slot) => slot.canonical === 'where');
const ENGINE_QUERY_SLOTS: readonly QueryAliasSlot[] =
  RPC_QUERY_ALIAS_SLOTS.filter((slot) => slot.canonical === 'where' || slot.canonical === 'limit');

/**
 * [#4371] The slots the engine does NOT fold — the wire-only pairs
 * (`select`→`fields`, `sort`→`orderBy`, `skip`→`offset`, `populate`→`expand`),
 * derived as the complement of {@link ENGINE_QUERY_SLOTS} so a seventh pair
 * added to the spec table lands on exactly one side of the split.
 *
 * Their values need shape lowering (`sort`'s `{field: 'asc'}` record form,
 * `populate`'s name list) that belongs to the RPC/protocol layers, so folding
 * here would re-implement the lowering per reader — the #3795 condition. But a
 * DIRECT engine call never crosses those layers: the alias key used to ride
 * the AST verbatim, drivers read only the canonical name, and the parameter
 * was silently dropped — three shipped "latest N in arbitrary order" bugs
 * (#4370) plus the engine's own `seedAutonumber`. Declared ≠ enforced
 * (AGENTS.md PD #10): the find-shaped entry points now REJECT these spellings,
 * naming the canonical key and shape, so the mistake throws at the call site
 * instead of degrading the result.
 *
 * Deliberately NOT applied to the where-only methods
 * (`update`/`delete`/`count`/`aggregate`): their contracts honour no
 * sort/projection/pagination at all, so "pass `orderBy` instead" would
 * redirect the caller to a key those methods silently ignore too. Unknown-key
 * enforcement for those bags is #4371's option (2), scoped separately.
 */
const ENGINE_WIRE_ONLY_SLOTS: readonly QueryAliasSlot[] =
  RPC_QUERY_ALIAS_SLOTS.filter((slot) => !ENGINE_QUERY_SLOTS.includes(slot));

/**
 * Canonical shape each wire-only slot's value must be rewritten into — quoted
 * by the rejection so the error carries the full migration, not just the key
 * rename (the value shapes differ; that is WHY the engine cannot fold them).
 */
const WIRE_ONLY_CANONICAL_SHAPES: Record<string, string> = {
  fields: "a string[] of field names",
  orderBy: "SortNode[]: [{ field, order: 'asc' | 'desc' }]",
  offset: 'a number of rows to skip',
  expand: 'a record of { relationName: QueryAST }',
};

/**
 * [#4371 option 2] The driver-option keys the engine forwards verbatim: on
 * `find`/`findOne`/`update`/`delete` the option bag IS the base of the driver
 * options (`buildDriverOptions(object, ctx, bag)`), which is how a caller's
 * explicit `tenantId` / `bypassTenantAudit` reaches the driver (pinned in
 * engine.test.ts). `count`/`aggregate` never forward the bag, so these keys
 * are deliberately NOT legal there — accepting them would be the exact
 * silently-ignored contract this gate exists to close.
 */
const ENGINE_DRIVER_PASSTHROUGH_KEYS = [
  'transaction', 'tenantId', 'tenantIds', 'timezone', 'bypassTenantAudit', 'preserveAudit',
] as const;

/**
 * [#4371 option 2] Per-method legal option keys. An option bag key outside
 * the method's set is REJECTED at the entry point: the engine executes none
 * of them, so the call would otherwise succeed with the option silently
 * ignored — the `declared ≠ enforced` shape (PD #10) one layer below the
 * wire-alias rejection above.
 *
 * Sources, in order: the method's `Engine*OptionsSchema` declared keys (minus
 * the `retiredKey` tombstones `cursor`/`distinct`, which get their tombstone
 * quoted instead of a generic rejection — the schema keeps them ONLY to carry
 * that message, and this runtime path never parses); `searchFields` (read by
 * `find` at the `$search` expansion, sent by the protocol layer);
 * `onFieldsDropped` (`WriteObservabilityOptions` — contract-declared,
 * unrepresentable in the serializable Zod schema); and the driver
 * pass-through keys above. The alias spellings (`filter`/`top`) are folded
 * and deleted BEFORE this check runs, so they never reach it.
 *
 * A drift pin in engine-unknown-option.test.ts asserts each set equals its
 * schema's shape (minus tombstones, plus the documented extras) so a key
 * added to the spec cannot be silently rejected here.
 */
const ENGINE_FIND_OPTION_KEYS: ReadonlySet<string> = new Set([
  'context', 'where', 'fields', 'orderBy', 'limit', 'offset',
  'search', 'searchFields', 'expand',
  ...ENGINE_DRIVER_PASSTHROUGH_KEYS,
]);
const ENGINE_UPDATE_OPTION_KEYS: ReadonlySet<string> = new Set([
  'context', 'where', 'upsert', 'multi', 'returning', 'onFieldsDropped',
  ...ENGINE_DRIVER_PASSTHROUGH_KEYS,
]);
const ENGINE_DELETE_OPTION_KEYS: ReadonlySet<string> = new Set([
  'context', 'where', 'multi',
  ...ENGINE_DRIVER_PASSTHROUGH_KEYS,
]);
const ENGINE_COUNT_OPTION_KEYS: ReadonlySet<string> = new Set(['context', 'where']);
const ENGINE_AGGREGATE_OPTION_KEYS: ReadonlySet<string> = new Set([
  'context', 'where', 'groupBy', 'aggregations', 'having', 'timezone',
]);

/** Tombstoned option keys: rejected with the spec's own removal notice. */
const ENGINE_RETIRED_OPTION_MESSAGES: Record<string, string> = {
  cursor: QUERY_CURSOR_REMOVED,
  distinct: QUERY_DISTINCT_REMOVED,
};

/**
 * The per-method legal key sets, exported for the drift pin ONLY
 * (engine-unknown-option.test.ts asserts each set against its schema's shape,
 * so a key added to the spec cannot be silently rejected here). Not a public
 * API surface — consumers pass options, they do not read this table.
 */
export const ENGINE_OPTION_KEY_SETS: Readonly<Record<string, ReadonlySet<string>>> = {
  find: ENGINE_FIND_OPTION_KEYS,
  findOne: ENGINE_FIND_OPTION_KEYS,
  update: ENGINE_UPDATE_OPTION_KEYS,
  delete: ENGINE_DELETE_OPTION_KEYS,
  count: ENGINE_COUNT_OPTION_KEYS,
  aggregate: ENGINE_AGGREGATE_OPTION_KEYS,
};

/**
 * Reject option-bag keys the engine does not execute (#4371 option 2).
 *
 * Runs AFTER `foldEngineOptionAliases`, so alias spellings are already folded
 * away (or thrown on). `null`-valued keys pass — a `null` is a withdrawal
 * carrying no intent a drop could lose, same rule as the fold. Retired keys
 * (`cursor`/`distinct`) quote their tombstone. Everything else gets the legal
 * key set, so the error carries the fix.
 */
function rejectUnknownEngineOptions(
  object: string,
  operation: string,
  bag: object | undefined,
  legal: ReadonlySet<string>,
): void {
  if (!bag) return;
  let unknown: string[] | undefined;
  for (const [key, value] of Object.entries(bag)) {
    if (value == null || legal.has(key)) continue;
    (unknown ??= []).push(key);
  }
  if (!unknown) return;
  const details = unknown.map((k) =>
    ENGINE_RETIRED_OPTION_MESSAGES[k] ? `'${k}': ${ENGINE_RETIRED_OPTION_MESSAGES[k]}` : `'${k}'`,
  );
  throw new Error(
    `${operation}('${object}') does not recognise option${unknown.length > 1 ? 's' : ''} ` +
    `${details.join('; ')}. The engine executes none of ${unknown.length > 1 ? 'them' : 'it'}, ` +
    `so the call would succeed with the option silently ignored (#4371). ` +
    `Legal keys for ${operation}: ${[...legal].sort().join(', ')}.`,
  );
}

/**
 * Fold the deprecated alias spellings of an engine option bag into their
 * canonical QueryAST keys, under the #3795/#4181 rule: an alias alone moves to
 * the canonical key, redundant identical spellings collapse, DIFFERENT values
 * for one slot are irreconcilable and throw (picking a winner IS the silent
 * drop), and an explicit `null` alias is a withdrawal.
 *
 * `rejectSlots` ({@link ENGINE_WIRE_ONLY_SLOTS}) names the slots whose alias
 * spellings the engine can neither fold nor honour: a non-null value under one
 * throws, quoting the canonical key and shape (#4371). `null` stays a
 * withdrawal here too — it carries no intent a drop could lose — and rides
 * through for drivers to ignore, exactly as before.
 *
 * Returns the SAME reference when no alias spelling is present (the common
 * path allocates nothing — `withResolvedWhere` discipline); otherwise folds a
 * shallow copy, because the bag belongs to the caller and may be reused (view
 * metadata, flow node config).
 */
function foldEngineOptionAliases<T extends object | undefined>(
  object: string,
  operation: string,
  bag: T,
  slots: readonly QueryAliasSlot[],
  rejectSlots?: readonly QueryAliasSlot[],
): T {
  if (!bag) return bag;
  if (rejectSlots) {
    const refused = rejectSlots.flatMap((slot) =>
      slot.aliases
        .filter((alias) => (bag as Record<string, unknown>)[alias] != null)
        .map((alias) => ({ alias, canonical: slot.canonical })),
    );
    if (refused.length > 0) {
      throw new Error(
        `${operation}('${object}') does not accept ` +
        `${refused.map((r) => `'${r.alias}'`).join(', ')}: ` +
        refused
          .map(
            (r) =>
              `'${r.alias}' is a wire spelling of '${r.canonical}', folded by the RPC/protocol ` +
              `layer — a direct engine call bypasses that fold, so the value would be silently ` +
              `dropped, not applied. Pass '${r.canonical}' ` +
              `(${WIRE_ONLY_CANONICAL_SHAPES[r.canonical] ?? 'the canonical QueryAST shape'}) instead.`,
          )
          .join(' '),
      );
    }
  }
  if (!slots.some((slot) => slot.aliases.some((alias) => alias in bag))) return bag;
  const folded: Record<string, unknown> = { ...bag };
  foldQueryAliasSlots(folded, slots, (conflict) => {
    throw new Error(
      `Conflicting options on ${operation}('${object}'): ` +
      `${conflict.spellings.map((s) => `'${s}'`).join(', ')} are spellings of the same ` +
      `parameter (canonical '${conflict.canonical}') and were given different values. ` +
      'Send exactly one.',
    );
  });
  return folded as T;
}

interface FormulaPlanEntry { name: string; expression: Expression; }

function planFormulaProjection(
  schema: any,
  requestedFields: string[] | undefined
): { plan: FormulaPlanEntry[]; projected?: string[] } {
  if (!schema?.fields) return { plan: [] };
  const allFieldNames = Object.keys(schema.fields);
  // When no explicit projection, evaluate every formula field on the schema —
  // matches REST default of "return everything". Explicit projection still
  // honours the caller's selection.
  const targets = (Array.isArray(requestedFields) && requestedFields.length > 0)
    ? requestedFields
    : allFieldNames;
  const plan: FormulaPlanEntry[] = [];
  const projected = new Set<string>();
  for (const f of targets) {
    const def = (schema.fields as any)[f];
    if (def?.type === 'formula' && def.expression) {
      // Normalize string-shorthand → Expression envelope (M9 transition).
      const expr: Expression = typeof def.expression === 'string'
        ? { dialect: 'cel', source: def.expression }
        : def.expression;
      plan.push({ name: f, expression: expr });
      // Pre-compile to surface syntax errors at planning stage rather than
      // per-row eval. Dependency discovery (which fields the formula reads)
      // is no longer used — CEL uses dynamic projection via `record.<field>`.
      ExpressionEngine.compile(expr);
    } else if (Array.isArray(requestedFields) && requestedFields.length > 0) {
      projected.add(f);
    }
  }
  if (plan.length === 0) return { plan: [] };
  // For formulas: project all schema fields so CEL `record.<field>` lookups
  // see complete data. Static dependency analysis on AST is M9.7 work.
  if (Array.isArray(requestedFields) && requestedFields.length > 0) {
    if (!projected.has('id')) projected.add('id');
    for (const fname of allFieldNames) {
      // Skip formula fields themselves — they are virtual and not
      // projectable by the underlying driver. Without this guard the
      // SQL driver emits `SELECT response_rate ...` which fails as
      // "no such column" and the driver returns [] (silently).
      const fdef = (schema.fields as any)[fname];
      if (fdef?.type === 'formula') continue;
      projected.add(fname);
    }
    return { plan, projected: Array.from(projected) };
  }
  // Implicit/full projection — leave projected undefined so the driver
  // returns its default columns (typically *).
  return { plan };
}

/**
 * Evaluate read-time formula virtual fields against the raw rows.
 *
 * The eval context mirrors `applyFieldDefaults` so formula and default
 * expressions see the same shape: a `now` pinned ONCE per operation (every row
 * and every formula field in one `find()` observes the same instant —
 * determinism, and no per-eval `new Date()` drift), plus `os.user` / `os.org`
 * resolved from the execution context (so a computed field can reference the
 * caller, e.g. `os.user.id`). Previously this passed only `{ record }`, so
 * `now()`/`today()` ran against live wall-clock and user/org were unreachable.
 *
 * (ADR-0053 Phase 2 will additionally thread `timezone` here once
 * `ExecutionContext.timezone` exists — see #1980; this change is independent
 * of timezone.)
 */
function applyFormulaPlan(
  plan: FormulaPlanEntry[],
  records: any[],
  execCtx?: ExecutionContextInput,
  nowSnapshot?: Date,
): void {
  if (!plan.length) return;
  const now = nowSnapshot ?? new Date();
  const timezone = execCtx?.timezone;
  const user = execCtx?.userId ? { id: String(execCtx.userId), positions: execCtx?.positions ?? [] } : undefined;
  const org = execCtx?.tenantId ? { id: String(execCtx.tenantId) } : undefined;
  for (const rec of records) {
    if (rec == null) continue;
    for (const fp of plan) {
      const r = ExpressionEngine.evaluate(fp.expression, { now, timezone, user, org, record: rec });
      rec[fp.name] = r.ok ? r.value : null;
    }
  }
}

export type HookHandler = (context: HookContext) => Promise<void> | void;

/**
 * Per-object hook entry with priority support
 */
export interface HookEntry {
  handler: HookHandler;
  object?: string | string[];  // undefined = global hook
  priority: number;
  packageId?: string;
  /**
   * Original metadata-form `Hook` definition this entry was bound from
   * (when registered via `bindHooksToEngine`). Pure code-paths that call
   * `engine.registerHook` directly leave this undefined.
   */
  meta?: any;
  /** Hook `name` from metadata; used for diagnostics & deduplication. */
  hookName?: string;
}

/** Function registry entry — see `registerFunction`. */
export interface FunctionEntry {
  handler: HookHandler;
  packageId?: string;
  /**
   * What this function does to data, as DECLARED at registration (#4396).
   * Only a `script`-node caller reads it: a flow function is contractually
   * pure, and the run summary counts on that, so a function that legitimately
   * writes declares `'writes'` and its step is reported as an effect the
   * platform cannot count rather than as none. Absent ⇒ `'pure'`.
   *
   * Carried on the registry entry rather than beside it because this IS part of
   * the registration — one registry, and what a caller needs to know about a
   * function travels with the function.
   */
  effect?: FlowFunctionEffect;
}

/**
 * Declarations that may accompany a `registerFunction` call. The bare
 * `packageId` string is still accepted in that position (every existing caller
 * passes one), so this widens the signature without a migration.
 */
export interface FunctionRegistrationOptions {
  /** Owning package — the unit `unregisterFunctionsByPackage` removes. */
  packageId?: string;
  /** Declared data effect (#4396); omit for the pure default. */
  effect?: FlowFunctionEffect;
}

/**
 * Operation Context for Middleware Chain
 */
export interface OperationContext {
  object: string;
  operation: 'find' | 'findOne' | 'insert' | 'update' | 'delete' | 'count' | 'aggregate';
  ast?: QueryAST;
  data?: any;
  options?: any;
  context?: ExecutionContextInput;
  result?: any;
}

/**
 * Trailing options for the READ methods (find / findOne / count / aggregate).
 *
 * Historically the read methods took their execution context INSIDE the query
 * (`query.context`), while the WRITE methods (insert / update) took it in a
 * trailing `options.context`. That split was a footgun: the same `{ context }`
 * object is correct as the 3rd arg to `insert` but was SILENTLY DROPPED as the
 * 3rd arg to `find` — a class of bugs where an intended `isSystem` bypass just
 * vanished (e.g. control-plane reads coming back empty once org-scoping hooks
 * were added). We now ALSO accept `context` via this trailing options arg on the
 * read methods, so "execution context goes in the trailing options argument" is
 * one rule across reads and writes. `query.context` remains supported; when both
 * are given, `options.context` wins (it is the explicit channel).
 */
export interface EngineReadOptions {
  context?: ExecutionContextInput;
}

/** Merge read-path execution context from the query and the trailing options. */
function mergeReadContext(
  fromQuery?: ExecutionContextInput,
  fromOptions?: ExecutionContextInput,
): ExecutionContextInput | undefined {
  if (fromOptions == null) return fromQuery;
  if (fromQuery == null) return fromOptions;
  return { ...fromQuery, ...fromOptions };
}

/**
 * True when this write is exempt from the `state_machine` validation rule —
 * both the insert `initialStates` entry check and the update `transitions`
 * check are skipped for it. Either the seed-specific `seedReplay` flag (#3433)
 * or the general `skipStateMachine` flag (#3479, set by the REST import runner
 * for a "historical" import) turns it off. Both are server-set, never
 * client-supplied.
 */
function shouldSkipStateMachine(ctx?: ExecutionContextInput): boolean {
  return ctx?.seedReplay === true || ctx?.skipStateMachine === true;
}

/**
 * Engine Middleware (Onion model)
 */
export type EngineMiddleware = (
  ctx: OperationContext,
  next: () => Promise<void>
) => Promise<void>;

/**
 * Derive the registry key for a metadata item.
 *
 * Most metadata items expose a top-level `name` (or `id`). The `View`
 * container defined by `@objectstack/spec/ui` is special: it aggregates
 * `list / form / listViews / formViews` for a single object and is
 * keyed implicitly by its target object name (see `data.object`).
 *
 * Per spec, `ViewSchema` does NOT have a top-level `name` field
 * (view.zod.ts), so we resolve it from the inner data source. This
 * matches the server-side metadata API contract (`/api/v1/meta/views/:object`).
 */
function resolveMetadataItemName(key: string, item: any): string | undefined {
  if (!item) return undefined;
  if (item.name) return item.name;
  if (item.id) return item.id;
  if (key === 'views') {
    // Independent ViewItems ("Object has-many View") carry a top-level `name`
    // (handled above) and bind to their object via `object`. The aggregated
    // container has no top-level name/object, so fall back to its inner data
    // source — matching the loader's expansion key.
    return (
      item?.object ||
      item?.list?.data?.object ||
      item?.form?.data?.object ||
      undefined
    );
  }
  return undefined;
}

/**
 * ObjectQL Engine
 * 
 * Implements the IDataEngine interface for data persistence.
 * Acts as the reference implementation for:
 * - CoreServiceName.data (CRUD)
 * - CoreServiceName.metadata (Schema Registry)
 */
/** A roll-up `summary` field on a parent object that aggregates a child. */
interface SummaryDescriptor {
  parentObject: string;
  summaryField: string;
  /** FK field on the child pointing back to the parent. */
  fkField: string;
  fn: 'count' | 'sum' | 'min' | 'max' | 'avg';
  /** Child field aggregated (unused for count). */
  sourceField: string;
  /**
   * Optional predicate (a query `where` FilterCondition) restricting which child
   * rows are aggregated. ANDed with the parent-FK match when the aggregate runs.
   * Undefined ⇒ aggregate every child of the parent.
   */
  filter?: Record<string, unknown>;
}

// `implements IObjectQLEngine` is the verification step of #4251 B3: every
// member the `objectql` slot's contract declares is checked against this class
// on every build, so the seven consumer-local surface declarations the contract
// replaced can never silently drift from the engine again. IObjectQLEngine
// extends IDataEngine, so the old claim rides along.
/**
 * [#4441] "The caller did not name a record here."
 *
 * `null` / `undefined` / `''` mean NO LINK — exactly what
 * `deleteBehavior: 'set_null'` writes — and an empty array is the multi-value
 * spelling of the same thing. None of them is an id to resolve.
 */
function isEmptyReferenceValue(v: unknown): boolean {
  if (v === null || v === undefined || v === '') return true;
  if (Array.isArray(v)) return v.length === 0 || v.every((e) => e === null || e === undefined || e === '');
  return false;
}

/**
 * RFC-4122 v4 uuid for the realtime `DataEvent.id` (#4626).
 *
 * The twin of the generator `MetadataManager` uses for `MetadataEvent.id`
 * (#4602/#4628) — same shape, same fallback, kept local rather than shared so
 * the engine's `core` import closure gains nothing (ADR-0076 D2 ratchet).
 * Prefers `crypto.randomUUID`; the fallback keeps browser-compatible (Pure)
 * environments without WebCrypto working while still satisfying
 * `DataEventSchema`'s `z.string().uuid()`.
 */
function generateEventUuid(): string {
  const c = globalThis.crypto;
  if (c && typeof c.randomUUID === 'function') {
    return c.randomUUID();
  }
  return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, (ch) => {
    const r = (Math.random() * 16) | 0;
    const v = ch === 'x' ? r : (r & 0x3) | 0x8;
    return v.toString(16);
  });
}

/**
 * Coerce a driver-returned primary key into `DataEvent.recordId` (a required
 * `string`). Returns `undefined` when the write has no single record identity
 * — a bulk `updateMany`/`deleteMany` returns only a count — so the caller can
 * decline to publish rather than fabricate one (#4626).
 */
function eventRecordId(value: unknown): string | undefined {
  if (typeof value === 'string') return value === '' ? undefined : value;
  if (typeof value === 'number' || typeof value === 'bigint') return String(value);
  return undefined;
}

/** `DataEvent.changes`/`before`/`after` are `z.record(...)` — only a plain object qualifies. */
function eventRecordBody(value: unknown): Record<string, unknown> | undefined {
  return value !== null && typeof value === 'object' && !Array.isArray(value)
    ? (value as Record<string, unknown>)
    : undefined;
}

/** `DataEvent.userId` — the acting user, when the execution context names one. */
function eventUserId(execCtx?: ExecutionContextInput): string | undefined {
  const userId = execCtx?.userId;
  if (userId == null) return undefined;
  const asString = String(userId);
  return asString === '' ? undefined : asString;
}

/**
 * Coerce a multi-row driver result into `BulkDataEvent.matched` (#4639).
 *
 * `IDataDriver.updateMany`/`deleteMany` are contracted to resolve the affected
 * row count (`Promise<number>`). A driver that resolves something else has not
 * met that contract, and the count is the ONLY substantive thing a bulk event
 * says — so this returns `undefined` and the caller declines to publish rather
 * than inventing a `matched: 0` that reads as "nothing was affected" when rows
 * very likely were.
 */
function eventMatchedCount(value: unknown): number | undefined {
  if (typeof value !== 'number' || !Number.isInteger(value) || value < 0) return undefined;
  return value;
}

export class ObjectQL implements IObjectQLEngine {
  /**
   * Ambient transaction store (ADR-0034). While a `transaction()` callback
   * runs, the active transaction handle lives here so that EVERY data
   * operation — including internal reads done during a write (reference
   * checks, hooks, expand) — automatically binds to the same connection
   * instead of asking the pool for another one and deadlocking on the
   * single-connection SQLite pool.
   */
  private readonly txStore = new AsyncLocalStorage<{ transaction: unknown }>();

  private drivers = new Map<string, IDataDriver>();
  private defaultDriver: string | null = null;
  private logger: Logger;

  // Datasource mapping rules (imported from defineStack)
  private datasourceMapping: Array<{
    namespace?: string;
    package?: string;
    objectPattern?: string;
    default?: boolean;
    datasource: string;
    priority?: number;
  }> = [];

  // Package manifests registry (for defaultDatasource lookup)
  private manifests = new Map<string, any>();

  // Datasource definitions by name (ADR-0015): carries schemaMode +
  // external.allowWrites so the write gate (Gate 3) can enforce federation
  // ownership. Populated from manifests in registerApp and via
  // registerDatasourceDef. Absent entry ⇒ treated as managed (default DB).
  private datasourceDefs = new Map<string, { schemaMode?: string; external?: { allowWrites?: boolean } }>();

  // Declared-but-unusable datasources, keyed by name (framework#3828). Written
  // by the datasource connection layer via markDatasourceUnavailable; read only
  // by getDriver, to explain a missing driver instead of blaming a typo. Empty
  // is the normal state — an entry here means a connect was refused or failed.
  private unavailableDatasources = new Map<string, DatasourceUnavailableInfo>();

  // Per-object hooks with priority support
  private hooks: Map<string, HookEntry[]> = new Map([
    ['beforeFind', []], ['afterFind', []],
    ['beforeInsert', []], ['afterInsert', []],
    ['beforeUpdate', []], ['afterUpdate', []],
    ['beforeDelete', []], ['afterDelete', []],
  ]);

  // Middleware chain (onion model)
  private middlewares: Array<{
    fn: EngineMiddleware;
    object?: string;
  }> = [];

  // Action registry: key = "objectName:actionName"
  private actions = new Map<string, { handler: (ctx: any) => Promise<any> | any; package?: string }>();

  // Function registry: name → handler. Used by `bindHooksToEngine` to
  // resolve string-named hook handlers (the JSON-safe form). Populated by
  // `defineStack({ functions })` via `AppPlugin`, or directly via
  // `engine.registerFunction(...)`.
  private functions = new Map<string, FunctionEntry>();

  // Realtime service for event publishing
  private realtimeService?: IRealtimeService;

  // i18n service backing validation-message + field-label localization (#3957).
  // Optional: without it, messages render from the built-in catalog against the
  // declared labels.
  private i18nService?: { t?: (key: string, locale: string, params?: Record<string, unknown>) => string };

  // Crypto provider backing `secret`-typed fields. Optional: when absent,
  // writing an object that declares a secret field fails closed (never
  // persists cleartext). Injected by the host via setCryptoProvider().
  private cryptoProvider?: ICryptoProvider;

  // [ADR-0105 D2 / #3623] Posture accessor for driver-scope widening under the
  // `group` posture. Injected by SecurityPlugin via setTenancyPostureProvider();
  // absent = equality scoping (fail toward isolation).
  private tenancyPostureProvider?: () => string | undefined;

  // Per-engine SchemaRegistry instance.
  //
  // Historically SchemaRegistry was a process-wide singleton of static state,
  // which broke multi-environment servers: a project kernel would inherit every
  // object registered by the control plane (e.g. sys_metadata), and
  // getDriver()'s owner lookup would route CRUD to the wrong database. Each
  // engine now owns its registry so kernels are fully isolated.
  private _registry: SchemaRegistry = new SchemaRegistry();

  constructor(hostContext: Record<string, any> = {}) {
    // Use provided logger or create a new one
    this.logger = hostContext.logger || createLogger({ level: 'info', format: 'pretty' });
    // Pick up production hardening switches from env so deployers can
    // enforce strict-body without code changes:
    //   OBJECTQL_STRICT_HOOKS=1 → unresolved hooks throw at bind time
    //   OBJECTQL_WARN_LEGACY_HANDLER=1 → log a deprecation per legacy bind
    if (process?.env?.OBJECTQL_STRICT_HOOKS === '1') {
      (this as any)._strictHookBinding = true;
    }
    if (process?.env?.OBJECTQL_WARN_LEGACY_HANDLER === '1') {
      (this as any)._warnLegacyHandler = true;
    }
    this.logger.info('ObjectQL Engine Instance Created');
  }

  /**
   * Service Status Report
   * Used by Kernel to verify health and capabilities.
   */
  getStatus() {
      return {
          name: CoreServiceName.enum.data,
          status: 'running',
          version: '0.9.0',
          features: ['crud', 'query', 'aggregate', 'transactions', 'metadata']
      };
  }

  /**
   * Expose the SchemaRegistry for plugins to register metadata.
   *
   * Returns the per-engine instance, NOT the class. Each ObjectQL engine
   * owns its registry so multi-environment kernels remain isolated.
   */
  get registry(): SchemaRegistry {
    return this._registry;
  }

  /**
   * Register a hook
   * @param event The event name (e.g. 'beforeFind', 'afterInsert')
   * @param handler The handler function
   * @param options Optional: target object(s) and priority
   */
  registerHook(event: string, handler: HookHandler, options?: {
    object?: string | string[];
    priority?: number;
    packageId?: string;
    /** Original metadata Hook definition (set by `bindHooksToEngine`). */
    meta?: any;
    /** Stable name from metadata (set by `bindHooksToEngine`). */
    hookName?: string;
  }) {
    // [#3195] Guard against enum-vs-dispatch drift: a hook on an event the
    // engine never triggers would register "successfully" and then silently
    // never fire. Warn loudly rather than swallow it. Not a hard reject — a
    // custom driver/plugin may dispatch its own events via `triggerHooks`.
    if (!DISPATCHABLE_HOOK_EVENTS.has(event)) {
      this.logger.warn(
        `Hook registered for '${event}', which the engine never dispatches — it will never fire. ` +
          `Dispatchable events: ${[...DISPATCHABLE_HOOK_EVENTS].join(', ')}. ` +
          `(Read filtering → RLS/permissions; field masking → field metadata; delete guards → beforeDelete.)`,
        { event, object: options?.object, hookName: options?.hookName },
      );
    }
    if (!this.hooks.has(event)) {
        this.hooks.set(event, []);
    }
    const entries = this.hooks.get(event)!;
    entries.push({
      handler,
      object: options?.object,
      priority: options?.priority ?? 100,
      packageId: options?.packageId,
      meta: options?.meta,
      hookName: options?.hookName,
    });
    // Sort by priority (lower runs first)
    entries.sort((a, b) => a.priority - b.priority);
    this.logger.debug('Registered hook', { event, object: options?.object, priority: options?.priority ?? 100, totalHandlers: entries.length });
  }

  /**
   * Remove all hooks registered under a given `packageId`. Used by
   * `bindHooksToEngine` to make re-binding (hot reload, app reinstall)
   * idempotent, and by app uninstall flows.
   */
  unregisterHooksByPackage(packageId: string): number {
    if (!packageId) return 0;
    let removed = 0;
    for (const [event, entries] of this.hooks.entries()) {
      const before = entries.length;
      const kept = entries.filter((e) => e.packageId !== packageId);
      if (kept.length !== before) {
        this.hooks.set(event, kept);
        removed += before - kept.length;
      }
    }
    if (removed > 0) {
      this.logger.debug('Unregistered hooks by package', { packageId, removed });
    }
    return removed;
  }

  /**
   * Register a named function handler that can later be referenced by
   * string from a `Hook.handler` field, an `Action.target`, or a flow
   * `script` node's `config.function`. This is the JSON-safe form of
   * handler binding — declarative metadata persisted to disk or shipped
   * over the wire only carries the name.
   *
   * The third parameter accepts either the owning `packageId` (its original
   * shape, unchanged for every existing caller) or a
   * {@link FunctionRegistrationOptions} record that also carries what the
   * function DECLARES about itself — today its data `effect` (#4396).
   */
  registerFunction(
    name: string,
    handler: HookHandler,
    packageIdOrOptions?: string | FunctionRegistrationOptions,
  ): void {
    if (!name || typeof handler !== 'function') return;
    const opts: FunctionRegistrationOptions =
      typeof packageIdOrOptions === 'string' ? { packageId: packageIdOrOptions } : (packageIdOrOptions ?? {});
    const { packageId, effect } = opts;
    this.functions.set(name, { handler, packageId, ...(effect ? { effect } : {}) });
    this.logger.debug('Registered function', { name, packageId, effect });
  }

  /** Look up a registered function by name. */
  resolveFunction(name: string): HookHandler | undefined {
    return this.functions.get(name)?.handler;
  }

  /**
   * Look up a registered function's FULL entry — the handler plus whatever it
   * declared about itself (#4396). `resolveFunction` above answers "can I call
   * it"; a caller that must also report what the call did (the automation
   * engine's `script` node, feeding the #4354 run summary) needs the
   * declaration, and reading it off the same registry keeps the two from
   * drifting.
   */
  resolveFunctionEntry(name: string): Readonly<FunctionEntry> | undefined {
    return this.functions.get(name);
  }

  /** Remove all functions registered under a given `packageId`. */
  unregisterFunctionsByPackage(packageId: string): number {
    if (!packageId) return 0;
    let removed = 0;
    for (const [name, entry] of this.functions.entries()) {
      if (entry.packageId === packageId) {
        this.functions.delete(name);
        removed += 1;
      }
    }
    if (removed > 0) {
      this.logger.debug('Unregistered functions by package', { packageId, removed });
    }
    return removed;
  }

  /**
   * Bind a list of declarative `Hook` metadata definitions to this engine.
   *
   * Convenience proxy to the canonical `bindHooksToEngine` so callers do
   * not need a separate import. Use `import { bindHooksToEngine } from
   * '@objectstack/objectql'` directly when you want the result object.
   */
  bindHooks(hooks: any[] | undefined, opts?: {
    packageId?: string;
    functions?: Record<string, HookHandler>;
    bodyRunner?: any;
    strict?: boolean;
    warnLegacyHandler?: boolean;
    metrics?: any;
  }): void {
    const merged = { ...(opts ?? {}), logger: this.logger } as any;
    if (!merged.bodyRunner && this._defaultBodyRunner) {
      merged.bodyRunner = this._defaultBodyRunner;
    }
    if (merged.strict === undefined && (this as any)._strictHookBinding) {
      merged.strict = true;
    }
    if (merged.warnLegacyHandler === undefined && (this as any)._warnLegacyHandler) {
      merged.warnLegacyHandler = true;
    }
    if (!merged.metrics && (this as any)._hookMetricsRecorder) {
      merged.metrics = (this as any)._hookMetricsRecorder;
    }
    bindHooksToEngine(this, hooks, merged);
  }

  /** Default hook body-runner — see {@link setDefaultBodyRunner}. */
  private _defaultBodyRunner?: any;
  /** Default action body-runner factory — see {@link setDefaultActionRunner}. */
  private _defaultActionRunner?: (actionDef: any) => ((ctx: any) => Promise<unknown>) | undefined;

  /**
   * Install a default body-runner used when `bindHooks` is called without
   * an explicit one. The runtime layer sets this once on each per-project
   * engine so every binding path (template seed, metadata sync, AppPlugin)
   * can execute hook `body.source` consistently.
   *
   * FIRST-WINS (#4251): "set once per engine" is this method's own contract,
   * so the method enforces it — a second call is ignored and returns `false`.
   * Callers used to implement the guard themselves by probing the private
   * `_defaultBodyRunner` field through `any` (multiple AppPlugin instances on
   * one kernel must not clobber each other's runner), which meant the
   * invariant lived in every caller and belonged to none. Nobody replaces a
   * runner on a live engine: every setter call site either owns a fresh
   * engine or wants exactly this keep-the-first behaviour.
   *
   * @returns `true` when this call installed the runner, `false` when one was
   * already present (kept unchanged).
   */
  setDefaultBodyRunner(runner: any): boolean {
    if (this._defaultBodyRunner) {
      this.logger.debug('Default body runner already installed — keeping the first');
      return false;
    }
    this._defaultBodyRunner = runner;
    return true;
  }

  /** The installed default body-runner, if any — the public read the first-wins guard implies. */
  getDefaultBodyRunner(): any {
    return this._defaultBodyRunner;
  }

  /**
   * Install a default ACTION body-runner factory: `(actionDef) => handler |
   * undefined`. The runtime layer sets this once per engine (same boot point
   * as {@link setDefaultBodyRunner}) so runtime-authored `action` metadata —
   * which registers through paths that have no sandbox access of their own,
   * notably ObjectQLPlugin's metadata-service re-sync — can turn a declarative
   * `body` into an executable `registerAction` handler. The factory returns
   * `undefined` for actions it cannot run (no `body`, invalid shape), which
   * callers must treat as "skip", not an error.
   *
   * FIRST-WINS (#4251) — same contract and rationale as
   * {@link setDefaultBodyRunner}.
   *
   * @returns `true` when this call installed the runner, `false` when one was
   * already present (kept unchanged).
   */
  setDefaultActionRunner(runner: (actionDef: any) => ((ctx: any) => Promise<unknown>) | undefined): boolean {
    if (this._defaultActionRunner) {
      this.logger.debug('Default action runner already installed — keeping the first');
      return false;
    }
    this._defaultActionRunner = runner;
    return true;
  }

  /** The installed default action-runner factory, if any. */
  getDefaultActionRunner(): ((actionDef: any) => ((ctx: any) => Promise<unknown>) | undefined) | undefined {
    return this._defaultActionRunner;
  }

  /**
   * Toggle strict hook-binding mode for this engine. When enabled, every
   * subsequent `bindHooks` call rejects on the first unresolved hook
   * instead of silently warning. Production runtimes should enable this.
   */
  setStrictHookBinding(strict: boolean): void {
    (this as any)._strictHookBinding = strict;
  }

  /** Toggle deprecation warnings for hooks still using legacy `handler` ref. */
  setWarnLegacyHandler(warn: boolean): void {
    (this as any)._warnLegacyHandler = warn;
  }

  /**
   * Install a metrics recorder used by every subsequent `bindHooks` call.
   * The recorder's methods are invoked per-execution to count outcomes
   * (success / error / timeout / capability_rejected), skips, and retries.
   * Defaults to no-op so the engine pays zero cost when nobody is observing.
   */
  setHookMetricsRecorder(recorder: any): void {
    (this as any)._hookMetricsRecorder = recorder;
  }

  /** Read the engine's installed metrics recorder, if any. */
  getHookMetricsRecorder(): any {
    return (this as any)._hookMetricsRecorder;
  }

  public async triggerHooks(event: string, context: HookContext) {
    const entries = this.hooks.get(event) || [];
    
    if (entries.length === 0) {
      this.logger.debug('No hooks registered for event', { event });
      return;
    }

    this.logger.debug('Triggering hooks', { event, count: entries.length });

    // `session.skipAutomations` (set from ExecutionContext.skipAutomations —
    // import with "run automations & triggers" unchecked, import undo)
    // suppresses hooks bound FROM METADATA (`bindHooksToEngine` stamps
    // `entry.meta`). Hooks registered in code by plugins — audit, capability
    // gates, sharing projection — have no `meta` and always run: the opt-out
    // must never bypass security or audit (#2922).
    const skipAutomations =
      (context.session as { skipAutomations?: boolean } | undefined)?.skipAutomations === true;

    for (const entry of entries) {
      // Per-object matching
      if (entry.object) {
        const targets = Array.isArray(entry.object) ? entry.object : [entry.object];
        if (!targets.includes('*') && !targets.includes(context.object)) {
          continue; // Skip non-matching hooks
        }
      }
      if (skipAutomations && entry.meta) {
        this.logger.debug('Skipping metadata-bound hook (skipAutomations)', { event, hook: entry.hookName });
        continue;
      }
      await entry.handler(context);
    }
  }

  // ========================================
  // Action System
  // ========================================

  /**
   * Register a named action on an object.
   * Actions are custom business logic callable via `repo.execute(actionName, params)`.
   *
   * @param objectName Target object
   * @param actionName Unique action name within the object
   * @param handler Handler function. Authoring sites should annotate it with
   *   `ActionHandler` from `@objectstack/spec/ui` (ADR-0104 D2) rather than an
   *   inline `(ctx: any)`; the params on `ctx` are validated against the
   *   action's declared param contract at dispatch before the handler runs.
   *   The seam itself stays untyped so existing untyped handlers keep working.
   * @param packageName Optional package owner (for cleanup)
   */
  registerAction(objectName: string, actionName: string, handler: (ctx: any) => Promise<any> | any, packageName?: string): void {
    const key = `${objectName}:${actionName}`;
    this.actions.set(key, { handler, package: packageName });
    this.logger.debug('Registered action', { objectName, actionName, package: packageName });
  }

  /**
   * Execute a named action on an object.
   */
  async executeAction(objectName: string, actionName: string, ctx: any): Promise<any> {
    const entry = this.actions.get(`${objectName}:${actionName}`);
    if (!entry) {
      throw new Error(`Action '${actionName}' on object '${objectName}' not found`);
    }
    return entry.handler(ctx);
  }

  /**
   * Every action handler currently registered, as `{ objectName, actionName }`
   * pairs (plus the owning package when one was given).
   *
   * [ADR-0110 D5] The handler registry is one half of the
   * declaration↔executable bijection; with no way to enumerate it, the other
   * half could only be checked when someone happened to invoke a route. The
   * boot reconciliation reads this to list handlers no declaration covers —
   * which since D3 are refused at dispatch, so having the inventory is what
   * makes that refusal a checklist instead of a support ticket.
   */
  listRegisteredActions(): Array<{ objectName: string; actionName: string; package?: string }> {
    const out: Array<{ objectName: string; actionName: string; package?: string }> = [];
    for (const [key, entry] of this.actions.entries()) {
      const sep = key.indexOf(':');
      if (sep < 0) continue;
      out.push({
        objectName: key.slice(0, sep),
        actionName: key.slice(sep + 1),
        ...(entry.package ? { package: entry.package } : {}),
      });
    }
    return out;
  }

  /**
   * Remove all actions registered by a specific package.
   */
  removeActionsByPackage(packageName: string): void {
    for (const [key, entry] of this.actions.entries()) {
      if (entry.package === packageName) {
        this.actions.delete(key);
      }
    }
  }

  /**
   * Register a middleware function
   * Middlewares execute in onion model around every data operation.
   * @param fn The middleware function
   * @param options Optional: target object filter
   */
  registerMiddleware(fn: EngineMiddleware, options?: { object?: string }): void {
    this.middlewares.push({ fn, object: options?.object });
    this.logger.debug('Registered middleware', { object: options?.object, total: this.middlewares.length });
  }

  /**
   * Execute an operation through the middleware chain
   */
  private async executeWithMiddleware(ctx: OperationContext, executor: () => Promise<any>): Promise<any> {
    const applicable = this.middlewares.filter(m =>
      !m.object || m.object === '*' || m.object === ctx.object
    );

    let index = 0;
    const next = async (): Promise<void> => {
      if (index < applicable.length) {
        const mw = applicable[index++];
        await mw.fn(ctx, next);
      } else {
        ctx.result = await executor();
      }
    };

    await next();
    return ctx.result;
  }

  /**
   * Build a HookContext.session from ExecutionContext — WHO is calling.
   *
   * Returns `undefined` when the context yields nothing session-worthy, which
   * keeps `!ctx.session` meaning exactly one thing for hooks: "no identity
   * envelope was supplied" — the bare-kernel / programmatic call that
   * caller-gating hooks skip for. A context carrying only write PROVENANCE
   * (`{ flowRunId }`, all an identity-less flow run has — #3712) is such a
   * case: it says what produced the write, not who is calling, and surfaces
   * through {@link buildProvenance} instead. Every real transport resolves
   * `positions` into the context, so an anonymous HTTP request still yields a
   * session and stays gated.
   */
  private buildSession(execCtx?: ExecutionContextInput): HookContext['session'] {
    if (!execCtx) return undefined;
    const session = {
      userId: execCtx.userId,
      // `organizationId` is the blessed developer-facing name for the caller's
      // active org (matches the `organization_id` column, `current_user`
      // RLS shape, and seed rows). It comes from `execCtx.tenantId`, which the
      // kernel resolves from `session.activeOrganizationId`. The deprecated
      // `session.tenantId` alias (#3280) was removed here in v11 (#3290) — the
      // driver-layer `execCtx.tenantId` knob is a separate axis and stays.
      organizationId: execCtx.tenantId,
      positions: execCtx.positions,
      accessToken: execCtx.accessToken,
      // Propagate system-elevated flag so hooks can distinguish engine
      // self-writes (e.g. approval status mirror) from genuine user writes.
      ...((execCtx as any).isSystem ? { isSystem: true } : {}),
      // Propagate the service-principal label (`ExecutionContext.actor`,
      // e.g. `svc:flow:<name>`) so a non-user write stays attributable in the
      // audit log — the writer's `userId ?? session.actor` fallback is dead
      // without this hop (ADR-0014 D2, #4366).
      ...(typeof (execCtx as any).actor === 'string' && (execCtx as any).actor
        ? { actor: (execCtx as any).actor }
        : {}),
      // Propagate the automation-suppression flag so the record-change trigger
      // can skip flow dispatch for seed/bulk writes (ADR: seed loads end-state
      // data, not user events). `skipAutomations` implies `skipTriggers` —
      // suppressing metadata hooks while still dispatching flows would leave
      // the "run automations & triggers" opt-out half-working (#2922).
      ...((execCtx as any).skipTriggers || (execCtx as any).skipAutomations ? { skipTriggers: true } : {}),
      // Propagate the full automation opt-out so `triggerHooks` can skip
      // metadata-bound hooks (import with "run automations" unchecked, undo).
      ...((execCtx as any).skipAutomations ? { skipAutomations: true } : {}),
      // Propagate the historical-import audit-preservation flag so the built-in
      // audit hook keeps a client-supplied updated_at/updated_by instead of
      // stamping now (#3493). Opt-in, server-set only.
      ...((execCtx as any).preserveAudit ? { preserveAudit: true } : {}),
    } as HookContext['session'];
    // Nothing to say about the caller → say nothing. An object whose every
    // property is `undefined` conveys no more than no session at all, and
    // returning it would turn "no caller" into "an anonymous caller".
    return Object.values(session as Record<string, unknown>).some((v) => v !== undefined)
      ? session
      : undefined;
  }

  /**
   * Build the HookContext.provenance envelope — WHERE this write came from.
   *
   * Deliberately separate from {@link buildSession}: provenance is server-
   * stamped, evaluated by no security middleware, and can exist with no
   * identity beside it. A schedule-triggered flow run resolves no principal
   * yet still owns its writes, and that is the case the approvals record lock
   * needs to recognize (#3456 / #3712).
   *
   * `attributedUserId` rides the SAME envelope for the same reason (#4586):
   * a better-auth-originated write authorizes as the system, so the human who
   * triggered it must reach the audit writer WITHOUT appearing in `session` —
   * where every caller-gating hook would read them as the caller. Attribution
   * here, authorization in `session`/`isSystem`, never the two mixed.
   */
  private buildProvenance(execCtx?: ExecutionContextInput): HookContext['provenance'] {
    const flowRunId = (execCtx as any)?.flowRunId;
    const attributedUserId = (execCtx as any)?.attributedUserId;
    if (!flowRunId && !attributedUserId) return undefined;
    return {
      ...(flowRunId ? { flowRunId: String(flowRunId) } : {}),
      ...(attributedUserId ? { attributedUserId: String(attributedUserId) } : {}),
    };
  }

  /**
   * Build the acting-user object (ADR-0068 EvalUser shape) surfaced to
   * validation-time predicates as `current_user` — notably per-option
   * `visibleWhen` authorization gating (objectui#2284). Returns undefined for
   * system / unauthenticated writes, where membership predicates then fail-open.
   */
  private buildEvalUser(
    execCtx?: ExecutionContextInput,
  ): { id: string; positions: string[]; organizationId: string | null } | undefined {
    if (!execCtx || execCtx.userId == null) return undefined;
    return {
      id: String(execCtx.userId),
      positions: execCtx.positions ?? [],
      organizationId: execCtx.tenantId != null ? String(execCtx.tenantId) : null,
    };
  }

  /**
   * Build the `HookContext.user` shortcut — the ergonomic "current user"
   * object surfaced to JS hooks. Carries `organizationId` (the blessed name
   * for the caller's active org, identical to `session.organizationId` and
   * `current_user.organizationId`) so a hook author who needs "the current org
   * to filter by" writes `ctx.user.organizationId` with zero relearning (#3280).
   *
   * Returns undefined for system / unauthenticated writes (no acting user) —
   * hooks that need an org regardless of a resolved user read
   * `ctx.session.organizationId`, which is populated whenever a session is.
   */
  private buildUser(execCtx?: ExecutionContextInput): HookContext['user'] {
    if (!execCtx || execCtx.userId == null) return undefined;
    return {
      id: String(execCtx.userId),
      ...(execCtx.email ? { email: execCtx.email } : {}),
      // Always equals `session.organizationId` (both from `execCtx.tenantId`).
      // Undefined on unscoped (platform/community) calls.
      ...(execCtx.tenantId != null ? { organizationId: String(execCtx.tenantId) } : {}),
    };
  }

  /**
   * Build the DriverOptions blob passed to every IDataDriver call.
   *
   * Carries `tenantId` from the active ExecutionContext so the driver can
   * enforce per-tenant isolation (SQL driver auto-scopes reads and
   * auto-injects the tenant column on writes) — EXCEPT for objects that
   * declare `tenancy.enabled: false` (ADR-0066 platform-global posture,
   * e.g. `sys_license`): stamping the caller's active-org tenantId there
   * would org-scope a global catalog at the driver, and its NULL-org rows
   * would vanish for authenticated org-context reads while anonymous
   * reads still see them (#3249). The SQL driver has its own opt-out
   * (sticky tenant-field cache), but withholding tenantId here protects
   * every driver at the source. Existing user-supplied shapes
   * (transactions, AST extras) are preserved by spreading them first — an
   * explicitly-passed `base.tenantId` is deliberate caller intent and
   * still wins.
   *
   * System / isSystem callers may still cross tenants by clearing
   * `tenantId` themselves on the resulting object; this helper does not
   * mask the system path.
   */
  private buildDriverOptions(object: string, execCtx?: ExecutionContextInput, base?: any): any {
    // The open transaction may arrive explicitly via the context, or ambiently
    // via txStore when an internal query runs during a transactional write
    // (ADR-0034). Explicit wins; ambient is the safety net.
    const tx = execCtx?.transaction !== undefined
      ? execCtx.transaction
      : this.txStore.getStore()?.transaction;
    const hasTx = tx !== undefined;
    const hasTenant =
      execCtx?.tenantId !== undefined &&
      !isTenancyDisabled(this._registry.getObject(object));
    const hasTz = execCtx?.timezone !== undefined;
    const isSystem = execCtx?.isSystem === true;
    const preserveAudit = (execCtx as any)?.preserveAudit === true;
    if (!hasTx && !hasTenant && !isSystem && !hasTz && !preserveAudit) return base;
    const opts: any = base && typeof base === 'object' ? { ...base } : {};
    if (hasTx && opts.transaction === undefined) {
      opts.transaction = tx;
    }
    if (hasTenant && opts.tenantId === undefined) {
      opts.tenantId = execCtx!.tenantId;
    }
    // [ADR-0105 D2 / #3623] Under the `group` posture the caller's read reach
    // is their whole membership set, not the active org — thread it so the
    // driver's native scope widens to the SAME union Layer 0 enforces, instead
    // of ANDing an active-org equality under it (which collapsed group reads
    // to isolated semantics). Inserts still stamp from `tenantId` (the active
    // org is the write target, D5). No provider / other postures / absent set
    // → no `tenantIds`, and drivers fall back to equality: fail toward
    // isolation, never toward exposure.
    if (hasTenant && opts.tenantIds === undefined && this.tenancyPostureProvider?.() === 'group') {
      const set = (execCtx as any)?.accessible_org_ids;
      if (Array.isArray(set) && set.length > 0) {
        opts.tenantIds = set.map(String);
      }
    }
    if (hasTz && opts.timezone === undefined) {
      // Thread the business timezone so date-dependent driver generation
      // (autonumber `{YYYYMMDD}` tokens) resolves the calendar day correctly.
      opts.timezone = execCtx!.timezone;
    }
    if (isSystem && opts.bypassTenantAudit === undefined) {
      // System-elevated writes (boot-time seeds, internal mirrors, scheduled
      // hooks) are unscoped by design — silence the audit warn for them but
      // still flag genuine user-path bugs.
      opts.bypassTenantAudit = true;
    }
    if (preserveAudit && opts.preserveAudit === undefined) {
      // Historical import (#3493): let the driver keep a supplied `updated_at`
      // instead of force-stamping now on the update path.
      opts.preserveAudit = true;
    }
    return opts;
  }

  /**
   * Build a HookContext.api: a ScopedContext that hooks can use to
   * read/write other objects within the same execution context.
   * Falls back to a system-elevated empty context when no execCtx
   * is supplied (e.g. system-triggered hooks).
   */
  private buildHookApi(execCtx?: ExecutionContextInput): ScopedContext {
    const safeCtx: ExecutionContextInput = execCtx ?? ({ isSystem: true } as any);
    return new ScopedContext(safeCtx, this as unknown as IDataEngine);
  }

  /**
   * Apply field defaults to an incoming insert payload. Defaults that are
   * Expression envelopes (e.g. `{ dialect: 'cel', source: 'today()' }`,
   * `{ dialect: 'cel', source: 'os.user.id' }`) are evaluated via
   * `ExpressionEngine` against the calling user/org/now snapshot. Static
   * defaults are applied verbatim. Records that already supplied a value for a
   * field are left untouched.
   *
   * "Supplied a value" means the field is present with a non-null value. Both an
   * OMITTED field (`undefined`) and an EXPLICIT `null` are treated as "not
   * supplied" and get the default. This runs on the INSERT path only, where a
   * `null` from a form (an unpicked control serializes to `null`, not omission)
   * unambiguously means "no value"; the UPDATE path never calls this, so a
   * deliberate "set to null" on update is preserved (#2706). Empty string `''`
   * is a real, user-entered value and is left as-is.
   *
   * Implements ROADMAP §M9.9b — `defaultValue` accepts Expression so authors
   * can replace "write a hook to default to today/current-user" with a
   * declarative `defaultValue: cel\`today()\``.
   */
  private applyFieldDefaults(
    object: string,
    record: Record<string, unknown>,
    execCtx?: ExecutionContextInput,
    nowSnapshot?: Date,
  ): Record<string, unknown> {
    const schema = this.getSchema(object);
    const fieldsRaw = (schema as any)?.fields;
    if (!fieldsRaw || typeof fieldsRaw !== 'object') return record;
    // `fields` may be a Record<string, Field> (canonical) or an array (legacy).
    const fieldEntries: Array<{ name: string; defaultValue?: unknown }> = Array.isArray(fieldsRaw)
      ? fieldsRaw
      : Object.entries(fieldsRaw).map(([name, def]) => ({ name, ...(def as object) }));
    const out = { ...record };
    const now = nowSnapshot ?? new Date();
    for (const f of fieldEntries) {
      // Apply the default when the field is either OMITTED (`undefined`) or an
      // EXPLICIT `null` — both mean "no value supplied" on insert (#2706). A
      // real value (including `''`) is respected. Insert-only path, so an
      // intentional "set to null" on update is never touched here.
      if (out[f.name] != null) continue;
      if (f.defaultValue == null) continue;
      const dv = f.defaultValue;
      if (typeof dv === 'object' && dv !== null && (dv as any).dialect && typeof (dv as any).source === 'string') {
        const result = ExpressionEngine.evaluate(dv as any, {
          now,
          timezone: execCtx?.timezone,
          user: execCtx?.userId ? { id: String(execCtx.userId), positions: execCtx?.positions ?? [] } : undefined,
          org: execCtx?.tenantId ? { id: String(execCtx.tenantId) } : undefined,
          record: out,
          extra: { object },
        });
        if (result.ok) {
          out[f.name] = result.value as unknown;
        } else {
          this.logger.warn('Failed to evaluate default expression', {
            object, field: f.name, error: result.error,
          });
        }
      } else if (isCurrentUserDefaultToken(dv)) {
        // `current_user` token → the acting user's id at insert time. Declarative
        // counterpart to writing a beforeInsert hook; mirrors the 'NOW()' string
        // convention and is resolved app-side per request (driver-agnostic), so
        // `Field.user({ defaultValue: 'current_user' })` auto-fills the actor.
        // When there is no authenticated user (system/anonymous), leave it unset
        // and let required-validation decide — never stamp a bogus owner.
        //
        // The token spelling comes from `@objectstack/spec/data`
        // (`DEFAULT_VALUE_TOKENS`), the one place the family is declared, so a
        // driver's DDL reads the SAME set when deciding which `defaultValue`s
        // may become a physical column DEFAULT. When the two sides disagreed,
        // SQL emitted `DEFAULT 'current_user'` and the DATABASE overrode the
        // "leave it unset" decision below with a literal non-id (#4560).
        if (execCtx?.userId != null) out[f.name] = String(execCtx.userId);
      } else {
        out[f.name] = dv;
      }
    }
    return out;
  }

  /**
   * Generate values for empty `autonumber` fields on insert — ONLY for drivers
   * that do not generate them natively (memory, mongodb). For SQL-backed objects
   * the driver owns a persistent, atomic `_objectstack_sequences` table and
   * advertises `supports.autonumber === true`; the engine then defers entirely
   * and never pre-fills (so the persistent sequence is the single source of
   * truth — see #1603). Required-validation exempts `autonumber` either way, so
   * a `required` record number is never rejected for "missing" — the runtime
   * owns the value, not the client.
   *
   * In the fallback path the next value is `max(existing) + 1`, seeded once per
   * `object.field.<scope>` from the store then incremented in memory (monotonic
   * within the process, resilient to deletions). The shared `autonumberFormat`
   * renderer is honored end-to-end, so date tokens (`AD{YYYYMMDD}{0000}`), field
   * interpolation (`{island_zone}{000}`) and per-scope reset behave identically
   * to the SQL driver's persistent sequence (#1603). NOTE: this in-memory seeding
   * is single-instance.
   */
  private async applyAutonumbers(
    object: string,
    record: Record<string, unknown>,
    execCtx?: ExecutionContextInput,
    driverOwnsAutonumber?: boolean,
  ): Promise<void> {
    if (driverOwnsAutonumber) return; // driver generates persistently in create()
    const fields = (this.getSchema(object) as any)?.fields;
    if (!fields || typeof fields !== 'object' || Array.isArray(fields)) return;
    const now = new Date();
    const timezone = execCtx?.timezone;
    for (const [name, def] of Object.entries(fields)) {
      if ((def as any)?.type !== 'autonumber') continue;
      const current = record[name];
      if (current != null && current !== '') continue; // respect explicit value
      // Honor either the spec-canonical `autonumberFormat` or the shorthand
      // `format` (both appear in metadata; the driver reads both too) — #1603.
      const fmt = (def as any).autonumberFormat ?? (def as any).format;
      const tokens = parseAutonumberFormat(typeof fmt === 'string' ? fmt : '');
      // Refuse to generate when an interpolated `{field}` is empty — it would
      // render to an empty prefix and merge this record into the wrong counter
      // scope. Mirror the SQL driver so both paths fail identically (#1603).
      const missing = missingFieldValues(tokens, record);
      if (missing.length > 0) {
        throw new Error(
          `Cannot generate autonumber "${object}.${name}" (format "${fmt}"): ` +
            `referenced field(s) [${missing.join(', ')}] are empty on the record. ` +
            `Fields interpolated into an autonumber format must be set before the record is created.`,
        );
      }
      // The counter scope is the rendered prefix (date/field tokens before the
      // sequence slot); it is independent of the counter value, so a throwaway
      // render with seq 0 yields the scope and the literal prefix to seed from.
      const probe = renderAutonumber({ tokens, seq: 0, record, now, timezone });
      const counterKey = `${object}.${name}.${probe.scope}`;
      let next = this.autonumberCounters.get(counterKey);
      if (next == null) next = await this.seedAutonumber(object, name, probe.prefix, execCtx);
      next += 1;
      this.autonumberCounters.set(counterKey, next);
      record[name] = renderAutonumber({ tokens, seq: next, record, now, timezone }).value;
    }
  }

  /**
   * Seed the autonumber counter from the current max in store, scoped to
   * `prefix`. With a non-empty prefix (date/field formats) only rows in the
   * same scope count, and the counter is the digit-run immediately after the
   * prefix; with an empty prefix (legacy fixed-prefix formats) the last digit
   * run of the whole value is used, preserving the original behaviour.
   */
  private async seedAutonumber(
    object: string,
    field: string,
    prefix: string,
    execCtx?: ExecutionContextInput,
  ): Promise<number> {
    try {
      // Canonical `fields`, not the wire spelling `select` — this call sat on
      // the exact #4371 silent drop (the projection never applied; the scan
      // worked only because an unprojected row still carries `field`), and the
      // catch below would have swallowed the guard's rejection into "seed
      // from 0", i.e. duplicate autonumbers.
      const rows = await this.find(object, {
        fields: ['id', field],
        limit: 5000,
        context: execCtx,
      } as any);
      let max = 0;
      for (const r of rows || []) {
        const v = r?.[field];
        if (v == null) continue;
        const s = String(v);
        if (prefix && !s.startsWith(prefix)) continue;
        const tail = prefix ? s.slice(prefix.length) : s;
        // With a prefix the counter is the digit run right after it; without one
        // (legacy fixed-prefix formats) it is the LAST digit run. Both use the
        // linear /\d+/g — a backtracking lookahead here is a polynomial-ReDoS
        // sink on stored values full of zeros (CodeQL js/polynomial-redos).
        let digits: string | undefined;
        if (prefix) {
          const head = tail.match(/^\d+/);
          digits = head ? head[0] : undefined;
        } else {
          const runs = tail.match(/\d+/g);
          digits = runs ? runs[runs.length - 1] : undefined;
        }
        if (digits) max = Math.max(max, parseInt(digits, 10) || 0);
      }
      return max;
    } catch {
      return 0;
    }
  }

  /**
   * Register contribution (Manifest)
   * 
   * Installs the manifest as a Package (the unit of installation),
   * then decomposes it into individual metadata items (objects, apps, actions, etc.)
   * and registers each into the SchemaRegistry.
   * 
   * Key: Package ≠ App. The manifest is the package. The apps[] array inside
   * the manifest contains UI navigation definitions (AppSchema).
   */
  registerApp(manifest: any) {
      const id = manifest.id || manifest.name;
      const namespace = manifest.namespace as string | undefined;
      this.invalidateSummaryIndex(); // new objects may add/change summary fields
      this.logger.debug('Registering package manifest', { id, namespace });

      // Store manifest for defaultDatasource lookup
      if (id) {
        this.manifests.set(id, manifest);
      }

      // Index datasource definitions (ADR-0015) so the write gate can read
      // schemaMode + external.allowWrites. Manifests may carry `datasources`
      // as an array or a name-keyed map.
      if (manifest.datasources) {
        const dsList = Array.isArray(manifest.datasources)
          ? manifest.datasources
          : Object.entries(manifest.datasources).map(([name, def]) => ({ name, ...(def as any) }));
        for (const ds of dsList) {
          if (ds?.name) this.registerDatasourceDef(ds);
        }
      }

      // 1. Register the Package (manifest + lifecycle state)
      this._registry.installPackage(manifest);
      this.logger.debug('Installed Package', { id: manifest.id, name: manifest.name, namespace });

      // 2. Register owned objects
      if (manifest.objects) {
          if (Array.isArray(manifest.objects)) {
             this.logger.debug('Registering objects from manifest (Array)', { id, objectCount: manifest.objects.length });
             for (const objDef of manifest.objects) {
                const fqn = this._registry.registerObject(objDef, id, namespace, 'own');
                this.logger.debug('Registered Object', { fqn, from: id });
             }
          } else {
             this.logger.debug('Registering objects from manifest (Map)', { id, objectCount: Object.keys(manifest.objects).length });
             for (const [name, objDef] of Object.entries(manifest.objects)) {
                // Ensure name in definition matches key
                (objDef as any).name = name;
                const fqn = this._registry.registerObject(objDef as any, id, namespace, 'own');
                this.logger.debug('Registered Object', { fqn, from: id });
             }
          }
      }

      // 2b. Register object extensions (fields added to objects owned by other packages)
      if (Array.isArray(manifest.objectExtensions) && manifest.objectExtensions.length > 0) {
          this.logger.debug('Registering object extensions', { id, count: manifest.objectExtensions.length });
          for (const ext of manifest.objectExtensions) {
              const targetFqn = ext.extend;
              const priority = ext.priority ?? 200;
              // Create a partial object definition for the extension
              const extDef = {
                  name: targetFqn, // Use the target FQN as name
                  fields: ext.fields,
                  label: ext.label,
                  pluralLabel: ext.pluralLabel,
                  description: ext.description,
                  validations: ext.validations,
                  indexes: ext.indexes,
              };
              // Register as extension (namespace is undefined since we're targeting by FQN)
              this._registry.registerObject(extDef as any, id, undefined, 'extend', priority);
              this.logger.debug('Registered Object Extension', { target: targetFqn, priority, from: id });
          }
      }

      // 3. Register apps (UI navigation definitions) as their own metadata type
      //    Resolve short objectName references in navigation to FQN so the
      //    Console UI can match them against the object registry.
      if (Array.isArray(manifest.apps) && manifest.apps.length > 0) {
          this.logger.debug('Registering apps from manifest', { id, count: manifest.apps.length });
          for (const app of manifest.apps) {
              const appName = app.name || app.id;
              if (appName) {
                  const resolved = namespace ? this.resolveNavObjectNames(app, namespace) : app;
                  this._registry.registerApp(resolved, id);
                  this.logger.debug('Registered App', { app: appName, from: id });
              }
          }
      }

      // 4. If manifest itself looks like an App (has navigation), also register as app
      //    This handles the case where the manifest IS the app definition (legacy/simple packages)
      if (manifest.name && manifest.navigation && !manifest.apps?.length) {
          const resolved = namespace ? this.resolveNavObjectNames(manifest, namespace) : manifest;
          this._registry.registerApp(resolved, id);
          this.logger.debug('Registered manifest-as-app', { app: manifest.name, from: id });
      }

      // 4b. Register navigation contributions (ADR-0029 D7) — nav items this
      //     package injects into apps owned by other packages (e.g. a
      //     capability plugin adding its menu into the `setup` app). Merged
      //     into the target app's navigation on read by group id + priority.
      if (Array.isArray((manifest as any).navigationContributions) && (manifest as any).navigationContributions.length > 0) {
          for (const contribution of (manifest as any).navigationContributions) {
              this._registry.registerAppNavContribution(contribution, id);
          }
          this.logger.debug('Registered navigation contributions', {
              from: id,
              count: (manifest as any).navigationContributions.length,
          });
      }

      // 5. Register all other metadata types generically
      const metadataArrayKeys = [
        // UI Protocol
        'actions', 'views', 'pages', 'dashboards', 'reports', 'datasets', 'themes',
        // Automation Protocol
        'flows', 'workflows', 'approvals', 'webhooks',
        'jobs',
        // Security Protocol
        'roles', 'permissions', 'profiles', 'sharingRules', 'policies',
        // AI Protocol
        'agents', 'tools', 'skills', 'ragPipelines',
        // API Protocol
        'apis',
        // Data Extensions
        'hooks', 'mappings', 'analyticsCubes',
        // Integration Protocol
        'connectors',
        // System Protocol — outbound mail templates. Registered here so the
        // email plugin's materializer can read them back into
        // `sys_email_template` (#4509); without this key an authored
        // `emailTemplates:` entry never reached the registry at all, which is
        // the far end of the disconnect the bridge closes.
        'emailTemplates',
        // System Protocol — package documentation (ADR-0046); inert data
        'docs',
        // Documentation navigation spine (ADR-0046 §6)
        'books',
      ];
      for (const key of metadataArrayKeys) {
          const items = (manifest as any)[key];
          if (Array.isArray(items) && items.length > 0) {
              this.logger.debug(`Registering ${key} from manifest`, { id, count: items.length });
              for (const item of items) {
                  const itemName = resolveMetadataItemName(key, item);
                  if (itemName) {
                      const toRegister = item.name === itemName ? item : { ...item, name: itemName };
                      this._registry.registerItem(pluralToSingular(key), toRegister, 'name' as any, id);
                      // "Object has-many View" (ADR-0017): a `defineView` document
                      // aggregates an object's views. Register the container under
                      // the bare <object> key (above, back-compat) AND expand it
                      // into independent ViewItems registered under <object>.<key>,
                      // so `getViewsByObject()` / `GET /meta/view?object=` surface
                      // the per-view `package` layer the switcher + Studio consume.
                      if (key === 'views' && isAggregatedViewContainer(toRegister)) {
                          for (const vi of expandViewContainer(itemName, toRegister)) {
                              for (const w of vi._diagnostics?.warnings ?? []) {
                                  this.logger.warn(`View expansion warning for '${vi.name}': ${w.message}`, { from: id });
                              }
                              this._registry.registerItem('view', vi, 'name' as any, id);
                          }
                      }
                  } else {
                      this.logger.warn(`Skipping ${pluralToSingular(key)} without a derivable name`, { id });
                  }
              }
          }
      }

      // 6. Register seed data as metadata (keyed by target object name)
      const seedData = (manifest as any).data;
      if (Array.isArray(seedData) && seedData.length > 0) {
          this.logger.debug('Registering seed data datasets', { id, count: seedData.length });
          for (const dataset of seedData) {
              if (dataset.object) {
                  this._registry.registerItem('data', dataset, 'object' as any, id);
              }
          }
      }

      // 6. Register contributions
       if (manifest.contributes?.kinds) {
          this.logger.debug('Registering kinds from manifest', { id, kindCount: manifest.contributes.kinds.length });
          for (const kind of manifest.contributes.kinds) {
            this._registry.registerKind(kind);
            this.logger.debug('Registered Kind', { kind: kind.name || kind.type, from: id });
          }
       }

      // 7. Recursively register nested plugins
      if (Array.isArray(manifest.plugins) && manifest.plugins.length > 0) {
          this.logger.debug('Processing nested plugins', { id, count: manifest.plugins.length });
          for (const plugin of manifest.plugins) {
              if (plugin && typeof plugin === 'object') {
                  const pluginName = plugin.name || plugin.id || 'unnamed-plugin';
                  this.logger.debug('Registering nested plugin', { pluginName, parentId: id });
                  this.registerPlugin(plugin, id, namespace);
              }
          }
      }
  }

  /**
   * Deep-clone an app definition, resolving objectName references in navigation
   * items via the registry. Object names are canonical identifiers — no FQN
   * expansion is applied.
   */
  private resolveNavObjectNames(app: any, namespace: string): any {
      if (!app.navigation) return app;

      const resolveItems = (items: any[]): any[] =>
          items.map((item: any) => {
              const resolved = { ...item };
              if (resolved.objectName && !resolved.objectName.includes('__')) {
                  resolved.objectName = computeFQN(namespace, resolved.objectName);
              }
              if (Array.isArray(resolved.children)) {
                  resolved.children = resolveItems(resolved.children);
              }
              return resolved;
          });

      return { ...app, navigation: resolveItems(app.navigation) };
  }

  /**
   * Register a nested plugin's metadata (objects, actions, views, etc.)
   *
   * Unlike registerApp(), this does NOT call SchemaRegistry.installPackage()
   * because plugins are not formal manifests — they are lightweight config
   * bundles with objects, actions, triggers, and navigation.
   *
   * @param plugin - The plugin config object
   * @param parentId - The parent package ID (for ownership tracking)
   * @param parentNamespace - The parent package's namespace (for FQN resolution)
   */
  private registerPlugin(plugin: any, parentId: string, parentNamespace?: string) {
      const pluginName = plugin.name || plugin.id || 'unnamed';
      const pluginNamespace = plugin.namespace || parentNamespace;

      // Use parentId as the owning package for namespace consistency.
      // The parent package already claimed the namespace — nested plugins
      // contribute objects UNDER the parent's ownership.
      const ownerId = parentId;

      // Register objects (supports both Array and Map formats)
      if (plugin.objects) {
          try {
              if (Array.isArray(plugin.objects)) {
                  this.logger.debug('Registering plugin objects (Array)', { pluginName, count: plugin.objects.length });
                  for (const objDef of plugin.objects) {
                      const fqn = this._registry.registerObject(objDef, ownerId, pluginNamespace, 'own');
                      this.logger.debug('Registered Object', { fqn, from: pluginName });
                  }
              } else {
                  const entries = Object.entries(plugin.objects);
                  this.logger.debug('Registering plugin objects (Map)', { pluginName, count: entries.length });
                  for (const [name, objDef] of entries) {
                      (objDef as any).name = name;
                      const fqn = this._registry.registerObject(objDef as any, ownerId, pluginNamespace, 'own');
                      this.logger.debug('Registered Object', { fqn, from: pluginName });
                  }
              }
          } catch (err: any) {
              this.logger.warn('Failed to register plugin objects', { pluginName, error: err.message });
          }
      }

      // Register plugin as app if it has navigation (for sidebar display)
      if (plugin.name && plugin.navigation) {
          try {
              const resolved = pluginNamespace ? this.resolveNavObjectNames(plugin, pluginNamespace) : plugin;
              this._registry.registerApp(resolved, ownerId);
              this.logger.debug('Registered plugin-as-app', { app: plugin.name, from: pluginName });
          } catch (err: any) {
              this.logger.warn('Failed to register plugin as app', { pluginName, error: err.message });
          }
      }

      // Register metadata arrays (actions, views, triggers, etc.)
      const metadataArrayKeys = [
          'actions', 'views', 'pages', 'dashboards', 'reports', 'datasets', 'themes',
          'flows', 'workflows', 'approvals', 'webhooks',
          'roles', 'permissions', 'profiles', 'sharingRules', 'policies',
          'agents', 'ragPipelines', 'apis',
          'hooks', 'mappings', 'analyticsCubes', 'connectors',
          'docs', 'books',
      ];
      for (const key of metadataArrayKeys) {
          const items = (plugin as any)[key];
          if (Array.isArray(items) && items.length > 0) {
              for (const item of items) {
                  const itemName = resolveMetadataItemName(key, item);
                  if (itemName) {
                      const toRegister = item.name === itemName ? item : { ...item, name: itemName };
                      this._registry.registerItem(pluralToSingular(key), toRegister, 'name' as any, ownerId);
                  }
              }
          }
      }
  }

  /**
   * Register a new storage driver
   */
  registerDriver(driver: IDataDriver, isDefault: boolean = false) {
    if (this.drivers.has(driver.name)) {
      this.logger.warn('Driver already registered, skipping', { driverName: driver.name });
      return;
    }

    this.drivers.set(driver.name, driver);
    this.logger.info('Registered driver', {
      driverName: driver.name,
      version: driver.version
    });

    if (isDefault || this.drivers.size === 1) {
      this.defaultDriver = driver.name;
      this.logger.info('Set default driver', { driverName: driver.name });
    }
  }

  /**
   * Register a Datasource *definition* (ADR-0015).
   *
   * Distinct from {@link registerDriver}, which registers a live connection.
   * This captures the declarative `schemaMode` + `external.allowWrites` so the
   * write gate ({@link assertWriteAllowed}) can enforce external-datasource
   * ownership. Safe to call repeatedly; last write wins.
   */
  registerDatasourceDef(def: { name: string; schemaMode?: string; external?: { allowWrites?: boolean } }): void {
    if (!def?.name) return;
    this.datasourceDefs.set(def.name, { schemaMode: def.schemaMode, external: def.external });
  }

  /**
   * Record that a **declared** datasource has no live driver, and why
   * (framework#3828). Called by `DatasourceConnectionService` when a connect is
   * refused by the host policy or fails while the operator has opted into a
   * degraded boot.
   *
   * The engine only needs this to answer a query well: without it
   * {@link getDriver} cannot tell a refused datasource from a misspelled one and
   * says `is not registered` to both. Deliberately carries no operator-facing
   * cause — see {@link DatasourceUnavailableError}.
   */
  markDatasourceUnavailable(info: { name: string; kind: DatasourceUnavailableKind; publicDetail?: string }): void {
    if (!info?.name) return;
    this.unavailableDatasources.set(info.name, {
      kind: info.kind,
      ...(info.publicDetail ? { publicDetail: info.publicDetail } : {}),
    });
  }

  /** Drop a {@link markDatasourceUnavailable} record (successful (re)connect / pool removal). */
  clearDatasourceUnavailable(name: string): void {
    this.unavailableDatasources.delete(name);
  }

  /**
   * Name of the DEFAULT driver, when one is registered (#3826). The default
   * driver keeps its natural name (`registerDriver(driver, true)` — nothing
   * routes by `drivers.get('default')`), so the datasource connection layer's
   * `asDefault` idempotency guard needs this rather than a name lookup.
   */
  getDefaultDriverName(): string | undefined {
    return this.defaultDriver ?? undefined;
  }

  /**
   * Datasources that were declared but are NOT usable, with the reason class.
   *
   * Distinct from {@link checkDriversHealth}, which answers "can a REGISTERED
   * driver serve a query right now" and therefore cannot see these at all — a
   * datasource that never connected was never registered (framework#3827).
   */
  listUnavailableDatasources(): Array<{ name: string } & DatasourceUnavailableInfo> {
    return Array.from(this.unavailableDatasources, ([name, info]) => ({ name, ...info }));
  }

  /**
   * Write gate — Gate 3 of ADR-0015 §5.3.
   *
   * Blocks insert/update/delete against a federated datasource
   * (`schemaMode !== 'managed'`) unless BOTH the datasource opts in
   * (`external.allowWrites`) AND the object opts in (`external.writable`).
   * Managed datasources (the common case, including the absence of any
   * definition) are unaffected.
   */
  private assertWriteAllowed(objectName: string, operation: 'insert' | 'update' | 'delete'): void {
    const object = this._registry.getObject(objectName) as any;
    const dsName = object?.datasource;
    if (!dsName || dsName === 'default') return;

    const ds = this.datasourceDefs.get(dsName);
    // No recorded definition, or an explicitly managed one ⇒ allow.
    if (!ds || !ds.schemaMode || ds.schemaMode === 'managed') return;

    const dsAllows = ds.external?.allowWrites ?? false;
    const objAllows = object?.external?.writable ?? false;
    if (!(dsAllows && objAllows)) {
      throw new ExternalWriteForbiddenError(
        `Write '${operation}' blocked on object '${objectName}': datasource '${dsName}' is external ` +
          `(schemaMode=${ds.schemaMode}). Requires datasource.external.allowWrites=true (got ${dsAllows}) ` +
          `AND object.external.writable=true (got ${objAllows}).`,
      );
    }
  }

  /**
   * Set the realtime service for publishing data change events.
   * Should be called after kernel resolves the realtime service.
   *
   * @param service - An IRealtimeService instance for event publishing
   */
  setRealtimeService(service: IRealtimeService): void {
    this.realtimeService = service;
    this.logger.info('RealtimeService configured for data events');
  }

  /**
   * Publish a realtime {@link DataEvent} for a record write (#4626 —
   * contract-first, the data-side twin of #4602/#4628).
   *
   * What reaches a `subscribeData` callback must BE the spec's `DataEvent`
   * (`@objectstack/spec/api`): top-level `id` (uuid), `type`, `object`,
   * `recordId` (REQUIRED), plus `changes`/`after`/`userId` when they apply.
   * The transport keeps its `RealtimeEventPayload` envelope — `payload`
   * carries the complete `DataEvent`, and the client SDK unwraps + validates
   * it at the boundary instead of double-casting the envelope.
   *
   * Only per-record writes reach here. A predicate (`multi: true`) write has
   * its own contract — see {@link publishBulkDataEvent} (#4639) — and the
   * multi branches of `update()`/`delete()` route to it directly, so they
   * never arrive at the identity gate below.
   *
   * Two loud-by-design gates:
   *  - **No record identity → no event.** `DataEvent.recordId` is required, so
   *    a write that names no record has no truthful per-record event to
   *    publish. It publishes NONE (warn log) instead of the pre-#4626
   *    fabrication (`recordId: ''`, `after: <affected count>`) that every
   *    schema-compliant consumer must reject. Reaching this gate now means a
   *    single-id write whose driver returned no usable primary key — a driver
   *    bug — because the bulk callers no longer come through here.
   *  - The event body is `DataEventSchema.parse`d before publish, so a
   *    malformed producer fails here (warn log, event not published) rather
   *    than delivering a lie downstream.
   *
   * Never throws and never fails the write: a realtime transport problem must
   * not roll back a committed record.
   */
  private async publishDataEvent(
    action: 'created' | 'updated' | 'deleted',
    object: string,
    input: {
      recordId: unknown;
      changes?: unknown;
      after?: unknown;
      context?: ExecutionContextInput;
    },
  ): Promise<void> {
    if (!this.realtimeService) return;

    const recordId = eventRecordId(input.recordId);
    if (!recordId) {
      this.logger.warn(
        `No data.record.${action} event published for '${object}': the write names no single record, ` +
          `and DataEvent.recordId is required — refusing to publish an off-contract event. ` +
          `A predicate write publishes data.records.${action} instead (#4639), so reaching this ` +
          `means a single-id write whose driver returned no usable primary key (#4626)`,
        { object },
      );
      return;
    }

    try {
      const timestamp = new Date().toISOString();
      const changes = eventRecordBody(input.changes);
      const after = eventRecordBody(input.after);
      const userId = eventUserId(input.context);
      const event: DataEvent = DataEventSchema.parse({
        id: generateEventUuid(),
        type: `data.record.${action}`,
        object,
        recordId,
        ...(changes !== undefined ? { changes } : {}),
        ...(after !== undefined ? { after } : {}),
        ...(userId !== undefined ? { userId } : {}),
        timestamp,
      });

      const envelope: RealtimeEventPayload = {
        type: event.type,
        object,
        payload: { ...event },
        timestamp,
      };

      await this.realtimeService.publish(envelope);
      this.logger.debug(`Published data.record.${action} event`, { object, recordId });
    } catch (error) {
      this.logger.warn('Failed to publish data event', { object, recordId, error });
    }
  }

  /**
   * Publish a realtime {@link BulkDataEvent} for a predicate write (#4639).
   *
   * The `multi: true` branches of `update()`/`delete()` reach
   * `IDataDriver.updateMany`/`deleteMany`, which resolve an affected COUNT and
   * nothing else. That is too little for the per-record `DataEvent` contract
   * (`recordId` is required), so those writes were silent from #4626 until now
   * — honest, but it meant webhooks and every other event consumer saw
   * nothing at all when a predicate write emptied half a table.
   *
   * They now get their own event instead of impersonating a per-record one:
   * `data.records.updated` / `data.records.deleted`, carrying the object and
   * the count. A consumer reading the type knows immediately that no
   * `recordId` is coming — the failure mode of the pre-#4626 fabrication,
   * where `recordId: ''` looked like a record until you tried to use it.
   *
   * Deliberately NOT carried: the query predicate. The only one in hand here
   * is the middleware-composed AST, whose `where` embeds the security layer's
   * injected row scoping (RLS, sharing) — publishing it would ship tenant
   * internals to whatever external URL a webhook points at. See
   * `BulkDataEventSchema`'s TSDoc for the full reasoning.
   *
   * Same two disciplines as the per-record twin: validate before publish, and
   * never throw — a realtime transport problem must not roll back a committed
   * write.
   */
  private async publishBulkDataEvent(
    action: 'updated' | 'deleted',
    object: string,
    input: { matched: unknown; context?: ExecutionContextInput },
  ): Promise<void> {
    if (!this.realtimeService) return;

    const matched = eventMatchedCount(input.matched);
    if (matched === undefined) {
      this.logger.warn(
        `No data.records.${action} event published for '${object}': the driver's multi-row result is ` +
          `not an affected-row count (IDataDriver.updateMany/deleteMany are contracted to resolve ` +
          `a number). The count is the only thing a bulk event states, so publishing one here would ` +
          `assert something unverified (#4639)`,
        { object },
      );
      return;
    }

    // A predicate that matched nothing changed no data, so it is not a data
    // event — the per-record path is silent for the same reason (no rows
    // written, no events). This is what keeps an idle hourly LifecycleService
    // sweep from becoming a webhook delivery per object per hour saying
    // "0 records". Debug, not warn: matching nothing is normal, not a fault.
    if (matched === 0) {
      this.logger.debug(`No data.records.${action} event for '${object}': predicate matched no rows`, { object });
      return;
    }

    try {
      const timestamp = new Date().toISOString();
      const userId = eventUserId(input.context);
      const event: BulkDataEvent = BulkDataEventSchema.parse({
        id: generateEventUuid(),
        type: `data.records.${action}`,
        object,
        matched,
        ...(userId !== undefined ? { userId } : {}),
        timestamp,
      });

      const envelope: RealtimeEventPayload = {
        type: event.type,
        object,
        payload: { ...event },
        timestamp,
      };

      await this.realtimeService.publish(envelope);
      this.logger.debug(`Published data.records.${action} event`, { object, matched });
    } catch (error) {
      this.logger.warn('Failed to publish bulk data event', { object, matched, error });
    }
  }

  /**
   * Set the i18n service used to localize write-path validation messages and
   * the field labels inside them (#3957). Bridged by `ObjectQLPlugin` on start,
   * the same way the realtime service is.
   *
   * Optional by design: with no service the built-in message catalog
   * (`@objectstack/spec/system`) still renders each message in the caller's
   * locale against the field's DECLARED label. The service adds two things —
   * a deployment's `validation.field.*` message overrides, and the field's
   * TRANSLATED label for apps whose declared labels are in another language.
   */
  setI18nService(service: { t?: (key: string, locale: string, params?: Record<string, unknown>) => string }): void {
    this.i18nService = service;
    this.logger.info('I18nService configured for validation messages');
  }

  /**
   * Locale + translation hooks handed to the validators so a rejected write is
   * reported in the caller's language (#3957).
   *
   * `ExecutionContext.locale` is resolved once per request from the
   * `localization` settings (ADR-0053 Phase 2) and its contract already reads
   * "Drives message catalogs" — this is the consumer that makes that true.
   * Undefined locale (anonymous / programmatic / system write) leaves the
   * built-in `en` rendering in place.
   */
  private validationMessageContext(
    objectName: string,
    // Only `locale` is read — narrower than `ExecutionContext` so both the
    // resolved and the input-shaped envelope satisfy it.
    context?: { locale?: string },
  ): { locale?: string; translate?: (key: string, locale: string, params?: Record<string, unknown>) => string; objectName: string } {
    const t = this.i18nService?.t;
    return {
      objectName,
      locale: context?.locale,
      translate: t ? (key, locale, params) => t.call(this.i18nService, key, locale, params) : undefined,
    };
  }

  /**
   * [#4441] Referential integrity on the WRITE path: a `lookup` (or any
   * reference-typed field) may not be given an id that exists in no row of the
   * object it declares.
   *
   * The field metadata is unambiguous — `{"type":"lookup","required":true,
   * "reference":"sys_permission_set"}` — and the DELETE side already reasons
   * about the edge (`deleteBehavior: 'set_null'`). Only the insert side never
   * checked, so `POST /data/sys_position_permission_set
   * {"permission_set_id":"ps_does_not_exist_at_all"}` created the row.
   *
   * On the RBAC link tables that is a security-surface record that resolves to
   * nothing: an administrator auditing permissions sees a binding whose target
   * cannot be inspected, and the audience-anchor gate has to resolve that very
   * set to evaluate the grant — so a dangling row is an unevaluable gate input,
   * not merely an untidy one.
   *
   * ## Scope, deliberately narrow
   *
   * - **Caller-supplied keys only.** Server stamps (`owner_id`,
   *   `organization_id`, `created_by`/`updated_by`) are lookups too; they are
   *   written by hooks and middleware, not by the request, and re-validating
   *   them here would turn a platform stamp into a caller-facing rejection.
   * - **Non-system writes only**, like every other write-path guard in this
   *   engine (`stripReadonlyFields`, `stripReadonlyForInsert`). Seed replay,
   *   package install and boot-time provisioning legitimately write rows in an
   *   order that resolves only once the batch completes; failing them closed
   *   would turn an ordering detail into a boot failure. This leaves a real
   *   residual — an `isSystem` caller can still write a dangling reference —
   *   which is recorded on the issue rather than silently accepted.
   * - **Empty values are not references.** `null` / `undefined` / `''` mean
   *   "no link", which is what `deleteBehavior: 'set_null'` produces.
   * - **Already-expanded objects are skipped.** A read round-trip can hand back
   *   `{id, name, …}` in the slot; that is not an id write.
   *
   * ## Why the probe is unscoped
   *
   * Existence is a fact about the database, not about the caller's visibility —
   * the same distinction the #4435 existence probe turns on. A scoped probe
   * would refuse a link to a permission set the caller cannot READ, which is
   * ordinary in an RLS-scoped deployment and would make the platform's own
   * admin flows fail. Whether the caller may create the binding at all is the
   * RBAC/RLS layer's decision, made where it already is.
   *
   * Fails OPEN when the target cannot be checked (unregistered object, no
   * driver, a probe that throws): an integrity check that cannot run must not
   * invent a rejection, and the alternative — refusing every write to an object
   * whose target lives on an unreachable datasource — converts a connectivity
   * problem into data loss.
   */
  private async assertReferencesResolve(
    schema: any,
    data: Record<string, unknown> | null | undefined,
    supplied: Record<string, unknown> | null | undefined,
    context: any,
    msgCtx?: { locale?: string; translate?: any; objectName?: string },
  ): Promise<void> {
    if (context?.isSystem) return;
    const fields = schema?.fields;
    if (!fields || !data) return;

    const failures: any[] = [];
    for (const name of Object.keys(fields)) {
      // A `readonly` field is never the caller's to answer for — BY
      // CONSTRUCTION, not by exemption.
      //
      // `stripReadonlyFields` removes a non-system caller's value from a
      // readonly field before the write, and the create ingress does the same
      // (`stripReadonlyForInsert`, #3043). So any value still sitting in one at
      // this point was written by the PLATFORM, which puts it outside this
      // check's own stated scope ("the reference the caller named").
      //
      // Found by the dogfood gate rather than by reasoning: `sys_metadata_history.
      // recorded_by` is `Field.lookup('sys_user', { readonly: true })` that the
      // metadata repository fills with `actor ?? 'system'` — a SENTINEL STRING,
      // not a user id, on a write that does not carry `isSystem`. Checking it
      // rejected ordinary metadata authoring (package create / publish / clone).
      // The sentinel-in-a-lookup is a real modelling wart and is filed
      // separately; it is not this change's to fix, and rejecting the
      // platform's own write is not the way to report it.
      //
      // This does NOT weaken #4441: the fields the issue names —
      // `sys_position_permission_set.permission_set_id` and
      // `showcase_task.project` — are ordinary author-facing lookups with no
      // `readonly`, and both stay enforced (pinned in the unit suite).
      if (fields[name]?.readonly === true) continue;
      // Only a value the CALLER actually supplied is theirs to answer for.
      //
      // Key presence is not enough: a form serializes an unpicked control as
      // an explicit `null`, and `applyFieldDefaults` then fills it from
      // `defaultValue` — including the `current_user` token (#2706). The key is
      // in the payload, but the ID that lands is the PLATFORM's, so validating
      // it would report a server-derived value as the caller's bad reference
      // (and reject a perfectly ordinary insert against a driver that has no
      // `sys_user` row for the acting principal).
      //
      // So the value read for the check comes from the post-normalization
      // `data` (multi-value strings are already split by then), while WHETHER
      // to check is decided by the caller's own raw value being non-empty.
      if (!supplied || isEmptyReferenceValue((supplied as Record<string, unknown>)[name])) continue;
      if (!(name in data)) continue;
      const def = fields[name];
      const target = referenceTargetOf(def);
      if (!target) continue;
      const raw = (data as Record<string, unknown>)[name];
      const values = Array.isArray(raw) ? raw : [raw];
      for (const v of values) {
        if (v === null || v === undefined || v === '') continue;
        if (typeof v === 'object') continue;
        const resolved = await this.referenceExists(target, v);
        if (resolved === false) {
          failures.push(buildFieldError(
            {
              field: name,
              code: 'reference_not_found',
              def,
              value: String(v),
              constraint: { target },
            },
            msgCtx as any,
          ));
        }
      }
    }
    if (failures.length > 0) throw new ValidationError(failures);
  }

  /**
   * Does `id` name a row in `target`? `false` only when the probe RAN and found
   * nothing; `null` when it could not run at all (see the fail-open note on
   * {@link assertReferencesResolve}).
   */
  private async referenceExists(target: string, id: unknown): Promise<boolean | null> {
    try {
      const resolved = this.resolveObjectName(target);
      if (!this._registry.getObject(resolved)) return null;
      const row = await this.findOne(resolved, {
        where: { id },
        fields: ['id'],
        context: { isSystem: true },
      } as any);
      return !!row;
    } catch {
      return null;
    }
  }

  /**
   * [#4551] Report stored references that resolve to nothing. **Read-only** —
   * this issues no writes at all.
   *
   * The follow-up to {@link assertReferencesResolve}'s deliberate `isSystem`
   * exemption. That exemption stays exactly as #4441 wrote it (seed replay and
   * boot provisioning must keep their ordering freedom); what it left behind is
   * a residual — the platform itself can still write a reference into the void
   * and nothing says so. This is the "something says so".
   *
   * The existence oracle passed to the audit is **this engine's own**
   * {@link referenceExists}, not a second copy: the audit and the write-path
   * guard therefore answer "does this id exist" — and "could I even tell?" —
   * with one predicate, so the report can never be more or less strict than the
   * rule it reports on.
   *
   * See {@link auditDanglingReferences} for the judgments (readonly skip, empty
   * values, unknown ≠ absent) and the bounded-scan honesty of the report.
   */
  async inspectDanglingReferences(
    options?: DanglingReferenceAuditOptions,
  ): Promise<DanglingReferenceReport> {
    return auditDanglingReferences(
      {
        objects: () => this._registry.getAllObjects() as unknown as AuditableObject[],
        find: (object, opts) => this.find(object, opts as any) as Promise<Array<Record<string, unknown>>>,
        probe: (target, id) => this.referenceExists(target, id),
        warn: (msg, meta) => this.logger?.warn?.(msg, meta as any),
      },
      options,
    );
  }

  /**
   * Register the crypto provider that backs `secret`-typed fields.
   *
   * When set, the engine encrypts secret fields on write (storing ciphertext in
   * `sys_secret` and only an opaque ref on the business row) and masks them on
   * read. When NOT set, writing to an object that declares a secret field is
   * **fail-closed** — the write throws rather than persist cleartext.
   *
   * Mirrors the Settings subsystem's ICryptoProvider wiring; the host (e.g.
   * `serve`) injects `LocalCryptoProvider` in dev and a KMS/Vault-backed
   * provider in production.
   */
  setCryptoProvider(provider: ICryptoProvider): void {
    this.cryptoProvider = provider;
    this.logger.info('CryptoProvider configured for secret fields');
  }

  /**
   * [ADR-0105 D2 / #3623] Inject the tenancy-posture accessor that decides
   * whether driver-level native tenant scoping widens to the caller's whole
   * membership set (`DriverOptions.tenantIds`, the `group` posture's union)
   * instead of the active-org equality (`DriverOptions.tenantId`).
   *
   * Wired by the enforcement layer (SecurityPlugin) — deliberately NOT
   * self-derived from env here: the posture in force is an entitlement
   * question (`tenancy` service), and widening the driver wall is only safe
   * when the Layer 0 union wall is actually enforcing above it. No provider
   * (an embedding without plugin-security) keeps today's equality scoping —
   * fail toward isolation, never toward exposure.
   */
  setTenancyPostureProvider(provider: () => string | undefined): void {
    this.tenancyPostureProvider = provider;
  }

  /**
   * Normalize credential fields on `row` in place before it reaches the driver.
   *
   * Two channels share the read mask ({@link SECRET_MASK}) but differ on write
   * (see ADR-0100):
   *
   *  - **`secret`** — encrypted: each plaintext is wrapped by the ICryptoProvider,
   *    persisted as a `sys_secret` row, and replaced on `row` by an opaque ref.
   *    Cleartext never reaches the business table.
   *  - **`password`** (generic, non-`better-auth`) — plaintext at rest: stored
   *    verbatim, no encryption and no `sys_secret` row. Only the echoed-mask drop
   *    below applies to it.
   *
   * Rules:
   *  - Any masked field (secret or password) whose value equals the read mask ⇒
   *    the key is dropped, so a form round-trip that echoes the mask does not
   *    overwrite the stored value.
   *  - No secret fields on the object ⇒ no further work (fast path, no crypto).
   *  - `null`/`undefined` secret value ⇒ left as-is (clears the secret).
   *  - Secret value already a ref (re-save of an unchanged ref) ⇒ left as-is.
   *  - **Fail-closed:** any other secret value with no CryptoProvider registered,
   *    or no reachable `sys_secret` store, THROWS — never persists cleartext.
   *    (A `password` field needs no CryptoProvider — it is stored as-is.)
   */
  private async encryptSecretFields(
    object: string,
    row: Record<string, unknown>,
    context: ExecutionContextInput | undefined,
    driverOptions: unknown,
  ): Promise<void> {
    if (!row || typeof row !== 'object') return;
    const schema = this._registry.getObject(object);

    // Echoed-mask drop for every field the read path masks (secret + generic
    // password). The read path returns SECRET_MASK; a client that PATCHes it
    // back means "unchanged", so drop the key rather than persist the literal
    // mask. Doing this up front keeps the better-auth exemption in one place
    // (collectMaskedReadFields) and covers objects that have a password field
    // but no secret field. (#2036, ADR-0100)
    for (const field of collectMaskedReadFields(schema)) {
      if (field in row && row[field] === SECRET_MASK) delete row[field];
    }

    const secretFields = collectSecretFields(schema);
    if (secretFields.length === 0) return;

    for (const field of secretFields) {
      if (!(field in row)) continue;
      const value = row[field];

      if (value === null || typeof value === 'undefined') continue; // clear
      if (isSecretRef(value)) continue; // already encrypted ref

      if (!this.cryptoProvider) {
        throw new Error(
          `Cannot persist secret field "${object}.${field}": no CryptoProvider is registered. `
            + 'Wire one via engine.setCryptoProvider(...) (e.g. LocalCryptoProvider in dev, '
            + 'a KMS/Vault provider in production). Refusing to store cleartext (fail-closed).',
        );
      }

      const plain = typeof value === 'string' ? value : JSON.stringify(value);
      const handle: CryptoHandle = await this.cryptoProvider.encrypt(plain, {
        namespace: object,
        key: field,
        tenantId: context?.tenantId,
      });

      let secretDriver;
      try {
        secretDriver = this.getDriver('sys_secret');
      } catch {
        throw new Error(
          `Cannot persist secret field "${object}.${field}": the sys_secret store is not available. `
            + 'Ensure the platform-objects (sys_secret) are registered before writing secret fields (fail-closed).',
        );
      }

      await secretDriver.create(
        'sys_secret',
        {
          id: handle.id,
          namespace: object,
          key: field,
          kms_key_id: handle.kmsKeyId,
          alg: handle.alg,
          version: handle.version,
          ciphertext: handle.ciphertext,
          created_at: new Date().toISOString(),
        },
        driverOptions as any,
      );

      row[field] = makeSecretRef(handle.id);
    }
  }

  /**
   * Mask credential fields on read so plaintext never leaves the engine through
   * the normal query path. Covers `secret` fields (always) and `password` fields
   * on generic, non-`better-auth` objects (see {@link collectMaskedReadFields}
   * and ADR-0100). A set value becomes {@link SECRET_MASK}; an unset one stays
   * `null`. Privileged callers that genuinely need a secret's plaintext use
   * {@link resolveSecret} against the stored ref; a `password` field is stored
   * as plaintext at rest, so its cleartext is only ever reachable off this path.
   */
  private maskSecretFields(object: string, rows: any): void {
    if (!rows) return;
    const schema = this._registry.getObject(object);
    const maskedFields = collectMaskedReadFields(schema);
    if (maskedFields.length === 0) return;
    const list = Array.isArray(rows) ? rows : [rows];
    for (const row of list) {
      if (!row || typeof row !== 'object') continue;
      for (const field of maskedFields) {
        if (!(field in row)) continue;
        row[field] = row[field] == null ? null : SECRET_MASK;
      }
    }
  }

  /**
   * Dereference a stored secret ref back to its plaintext. Intended for
   * privileged, server-side consumers (e.g. a datasource connection-pool
   * binder) — NOT exposed through the generic read path, which only ever
   * returns the mask.
   *
   * Fail-closed: throws when no CryptoProvider is registered or the
   * `sys_secret` row is missing. Returns `null` when `ref` is not a secret ref.
   */
  async resolveSecret(ref: unknown, opts?: { tenantId?: string }): Promise<string | null> {
    const id = parseSecretRef(ref);
    if (!id) return null;
    if (!this.cryptoProvider) {
      throw new Error('Cannot resolve secret: no CryptoProvider is registered (fail-closed).');
    }
    const secretDriver = this.getDriver('sys_secret');
    const found = await secretDriver.find('sys_secret', { object: 'sys_secret', where: { id } } as QueryAST);
    const secret: any = Array.isArray(found) ? found[0] : found;
    if (!secret) {
      throw new Error(`Cannot resolve secret: sys_secret row "${id}" not found (fail-closed).`);
    }
    const handle: CryptoHandle = {
      id: secret.id,
      kmsKeyId: secret.kms_key_id,
      alg: secret.alg,
      version: secret.version,
      ciphertext: secret.ciphertext,
    };
    return this.cryptoProvider.decrypt(handle, {
      namespace: secret.namespace,
      key: secret.key,
      tenantId: opts?.tenantId,
    });
  }

  /**
   * Helper to get object definition
   */
  getSchema(objectName: string): ServiceObject | undefined {
    return this._registry.getObject(objectName);
  }

  /**
   * Resolve any object identifier to the physical storage name used by drivers.
   *
   * Accepts the canonical short name (e.g., 'account') or, for explicit
   * cross-package disambiguation, the canonical object name (e.g., 'account'). The result is
   * the physical table name derived via `StorageNameMapping.resolveTableName`.
   */
  private resolveObjectName(name: string): string {
    const schema = this._registry.getObject(name);
    if (schema) {
      return StorageNameMapping.resolveTableName(schema);
    }
    // Return name as-is (canonical name = table name; no FQN prefix to strip)
    return StorageNameMapping.resolveTableName({ name });
  }

  /**
   * Name of the dedicated datasource lifecycle-classed system data prefers
   * when one is registered (ADR-0057 §3.6 / P3). Purely opt-in by the
   * datasource's existence — no `telemetry` driver registered ⇒ resolution
   * is exactly what it was before.
   */
  static readonly LIFECYCLE_DATASOURCE = 'telemetry';

  /**
   * Helper to get the target driver
   *
   * Resolution priority (first match wins):
   * 1. Object's explicit `datasource` field (if not 'default')
   * 2. DatasourceMapping rules (namespace/package/pattern matching)
   * 3. Lifecycle-class separation (ADR-0057 §3.6): telemetry/event/audit
   *    objects route to the dedicated 'telemetry' datasource when registered
   * 4. Package's `defaultDatasource` from manifest
   * 5. Global default driver
   */
  private getDriver(objectName: string): IDataDriver {
    const object = this._registry.getObject(objectName);

    // 1. Object's explicit datasource field (highest priority)
    if (object?.datasource && object.datasource !== 'default') {
      if (this.drivers.has(object.datasource)) {
        return this.drivers.get(object.datasource)!;
      }
      // The datasource layer may have recorded WHY this one has no driver —
      // refused by the host policy, or failed to connect under
      // OS_ALLOW_DRIVER_CONNECT_FAILURE (framework#3828). Saying so beats
      // sending the reader hunting for a typo that isn't there.
      const unavailable = this.unavailableDatasources.get(object.datasource);
      if (unavailable) {
        throw new DatasourceUnavailableError(
          object.datasource,
          objectName,
          unavailable.kind,
          unavailable.publicDetail,
        );
      }
      // No record: nothing ever tried to connect this name, so it is genuinely
      // undeclared (or misspelled). Unchanged message — there is nothing to add.
      throw new Error(`[ObjectQL] Datasource '${object.datasource}' configured for object '${objectName}' is not registered.`);
    }

    // 2. Check datasourceMapping rules
    //
    // A rule that MATCHES is a routing decision, not a hint (#4462). It used to
    // fall through to steps 3-5 whenever the named datasource had no live
    // driver, which put an object's rows in the DEFAULT store while every
    // signal said otherwise: boot succeeded, `/ready` answered 200, the
    // datasource name appeared nowhere in the log, and the write returned 201.
    // An operator who routes an object to Postgres and gets the URL wrong finds
    // out by going to look in Postgres and finding it empty.
    //
    // `default` is the one name that legitimately resolves onward: the default
    // driver keeps its NATURAL name (#3826), so `drivers.has('default')` is
    // false by construction and step 5 is how routing to it works.
    const mappedDatasource = this.resolveDatasourceFromMapping(objectName, object);
    if (mappedDatasource && mappedDatasource !== 'default') {
      if (this.drivers.has(mappedDatasource)) {
        this.logger.debug('Resolved datasource from mapping', {
          object: objectName,
          datasource: mappedDatasource
        });
        return this.drivers.get(mappedDatasource)!;
      }
      // Same three-way diagnosis as an explicit `object.datasource` binding —
      // the two are the same promise made in two places, so they owe the reader
      // the same answer.
      const unavailable = this.unavailableDatasources.get(mappedDatasource);
      if (unavailable) {
        throw new DatasourceUnavailableError(
          mappedDatasource,
          objectName,
          unavailable.kind,
          unavailable.publicDetail,
        );
      }
      throw new Error(
        `[ObjectQL] Datasource '${mappedDatasource}' mapped for object '${objectName}' is not registered. ` +
        `A datasourceMapping rule routes this object to it, so falling back to the default store would ` +
        `write the object's data to a different database than the one it declares. Fix the datasource ` +
        `configuration, or remove the mapping rule.`,
      );
    }

    // 3. Lifecycle-class separation (ADR-0057 §3.6): high-frequency
    // platform-generated data (telemetry / event) and the audit ledger move
    // to a dedicated 'telemetry' datasource when one is registered, so their
    // growth can never again pollute the business DB. `transient` stays on
    // the primary deliberately: those objects are user-session data, and
    // some (e.g. better-auth's sys_device_code) are also accessed outside
    // the engine — splitting their storage would split their brain.
    const lifecycleClass = (object as { lifecycle?: { class?: string } } | undefined)?.lifecycle?.class;
    if (
      (lifecycleClass === 'telemetry' || lifecycleClass === 'event' || lifecycleClass === 'audit') &&
      this.drivers.has(ObjectQL.LIFECYCLE_DATASOURCE)
    ) {
      return this.drivers.get(ObjectQL.LIFECYCLE_DATASOURCE)!;
    }

    // 4. Check package's defaultDatasource
    // Use the object's FQN name (from getObject) for ownership lookup
    const fqn = object?.name || objectName;
    const owner = this._registry.getObjectOwner(fqn);
    if (owner?.packageId) {
      const manifest = this.manifests.get(owner.packageId);
      if (manifest?.defaultDatasource && manifest.defaultDatasource !== 'default') {
        if (this.drivers.has(manifest.defaultDatasource)) {
          this.logger.debug('Resolved datasource from package manifest', {
            object: objectName,
            package: owner.packageId,
            datasource: manifest.defaultDatasource
          });
          return this.drivers.get(manifest.defaultDatasource)!;
        }
      }
    }

    // 5. Fallback to global default driver
    if (this.defaultDriver && this.drivers.has(this.defaultDriver)) {
      return this.drivers.get(this.defaultDriver)!;
    }

    throw new Error(`[ObjectQL] No driver available for object '${objectName}'`);
  }

  /**
   * Which datasource do the mapping rules route `objectName` to, if any?
   *
   * The PUBLIC face of {@link resolveDatasourceFromMapping}, added for the boot
   * path (#4462): the datasource-connection service must connect the
   * datasources a mapping actually routes objects to, and it must learn which
   * those are from the same resolver the query path uses. A second
   * implementation of "does this rule match?" living in the connection service
   * would drift by one clause and produce the worst of both postures — a
   * datasource connected that routing does not use, or routed to and never
   * connected, which is the defect itself.
   *
   * Returns `null` when no rule matches, and the datasource name (including
   * `'default'`) when one does. Rule matching only — an explicit
   * `object.datasource` binding outranks this and is not consulted here.
   */
  resolveMappedDatasource(objectName: string): string | null {
    return this.resolveDatasourceFromMapping(objectName, this._registry.getObject(objectName));
  }

  /**
   * Resolve datasource from mapping rules
   *
   * Rules are evaluated in order (or by priority if specified).
   * First matching rule wins.
   */
  private resolveDatasourceFromMapping(
    objectName: string,
    object?: any
  ): string | null {
    if (!this.datasourceMapping || this.datasourceMapping.length === 0) {
      return null;
    }

    // Sort rules by priority if any have priority set
    const sortedRules = [...this.datasourceMapping].sort((a, b) => {
      const aPriority = a.priority ?? 1000;
      const bPriority = b.priority ?? 1000;
      return aPriority - bPriority;
    });

    for (const rule of sortedRules) {
      // 1. Match by namespace
      if (rule.namespace && object?.namespace === rule.namespace) {
        return rule.datasource;
      }

      // 2. Match by package ID
      if (rule.package && object?.packageId === rule.package) {
        return rule.datasource;
      }

      // 3. Match by object name pattern (glob-style)
      if (rule.objectPattern && this.matchPattern(objectName, rule.objectPattern)) {
        return rule.datasource;
      }

      // 4. Default fallback rule
      if (rule.default) {
        return rule.datasource;
      }
    }

    return null;
  }

  /**
   * Simple glob pattern matching
   * Supports * (any chars) and ? (single char)
   */
  private matchPattern(objectName: string, pattern: string): boolean {
    const regexPattern = pattern
      .replace(/[.+^${}()|[\]\\]/g, '\\$&')  // Escape regex special chars
      .replace(/\*/g, '.*')                   // * → .*
      .replace(/\?/g, '.');                   // ? → .

    const regex = new RegExp(`^${regexPattern}$`);
    return regex.test(objectName);
  }

  /**
   * Set datasource mapping rules
   * Called by ObjectQLPlugin during bootstrap
   */
  setDatasourceMapping(rules: Array<{
    namespace?: string;
    package?: string;
    objectPattern?: string;
    default?: boolean;
    datasource: string;
    priority?: number;
  }>) {
    this.datasourceMapping = rules;
    this.logger.info('Datasource mapping rules configured', {
      ruleCount: rules.length
    });
  }

  /**
   * Initialize the engine and all registered drivers.
   *
   * **Fail-fast by default** (framework#3741): if any boot-registered driver's
   * `connect()` rejects, this throws {@link DriverConnectError} and kernel
   * bootstrap aborts. Two reasons, both load-bearing:
   *
   *  1. Booting without a reachable datasource produces a server that reports
   *     itself started and then 500s every request with an error that reads
   *     nothing like "the database is unreachable". The caller immediately
   *     makes it worse: `ObjectQLPlugin.start()` runs `syncRegisteredSchemas()`
   *     right after this, issuing DDL against a driver that isn't there — so
   *     the boot leaves the schema half-applied even if the database appears
   *     a second later. Recovery is not the point: the underlying clients DO
   *     re-establish connections on their own (framework#3759 verified this),
   *     but nothing re-runs the boot sequence that was skipped.
   *  2. Swallowing the rejection removed a driver's ability to REFUSE STARTUP.
   *     Any fatal startup check — licence, server version, incompatible
   *     configuration, missing capability, not just an unreachable socket — is
   *     expressed by throwing from `connect()`, and every one of them used to
   *     be silently downgraded to a runtime error. (That is why
   *     driver-mongodb's multi-tenancy guard had to be hoisted into its
   *     constructor in #3734; `connect()` is a supported place for it now.)
   *
   * Operators who need the old lenient behaviour opt in explicitly with
   * `OS_ALLOW_DRIVER_CONNECT_FAILURE=1`, which boots in a state that is warned
   * about loudly rather than assumed.
   */
  async init() {
    this.logger.info('Initializing ObjectQL engine', {
      driverCount: this.drivers.size,
      drivers: Array.from(this.drivers.keys())
    });

    const failures: DriverConnectFailure[] = [];
    for (const [name, driver] of this.drivers) {
      try {
        await driver.connect();
        this.logger.info('Driver connected successfully', { driverName: name });
      } catch (e) {
        failures.push({ driverName: name, error: e });
        this.logger.error('Failed to connect driver', e as Error, { driverName: name });
      }
    }

    if (failures.length > 0) {
      if (!resolveAllowDriverConnectFailure()) {
        throw new DriverConnectError(failures, this.drivers.size);
      }
      const failedDrivers = failures.map(f => f.driverName);
      const banner =
        `⚠️ DEGRADED BOOT: ${failures.length} of ${this.drivers.size} driver(s) failed to connect ` +
        `(${failedDrivers.join(', ')}), but OS_ALLOW_DRIVER_CONNECT_FAILURE is set — starting anyway. ` +
        `Every query routed to them fails until the datasource becomes reachable, and the boot-time ` +
        `schema sync is SKIPPED FOR GOOD — the client will reconnect on its own, but nothing re-runs ` +
        `the DDL, so those objects may have no tables even after the database comes back. Unset ` +
        `OS_ALLOW_DRIVER_CONNECT_FAILURE to restore fail-fast boot.`;
      this.logger.warn(banner, { failedDrivers });
      // …and again on a channel the operator's log level cannot filter away: the
      // line above is dropped outright at `error`/`fatal`/`silent`, which is a
      // normal production setting and precisely where this flag gets used. See
      // the helper's note — the duplication is deliberate, and it is NOT about
      // `os serve`'s boot-quiet window (that swallowed it too until #4012).
      emitDegradedBootBanner(banner);
    }

    this.logger.info('ObjectQL engine initialization complete');
  }

  /**
   * Ping every registered driver and report which ones are usable RIGHT NOW
   * (framework#3756).
   *
   * `init()` answers "could the drivers connect at boot"; this answers "can
   * they serve a query at this instant", which is a different question the
   * moment the database restarts, fails over, or drops the pool. Readiness
   * probes are the intended caller — a replica whose driver is down fails 100%
   * of its requests and must leave the load-balancer rotation.
   *
   * Every check is bounded by `timeoutMs` (default 2s) and settled
   * independently. The bound is not optional: `IDataDriver.checkHealth()`
   * swallows its own errors and returns `false`, but on a dead pool it does
   * not return at all — knex's `SELECT 1` waits out `acquireConnectionTimeout`
   * (60s by default). A probe that hangs is as useless as one that lies, so a
   * timed-out driver is reported unhealthy rather than awaited.
   *
   * A driver that implements no `checkHealth()` is reported healthy: absence of
   * a probe is not evidence of failure, and reporting "unhealthy" would take a
   * working deployment out of rotation over a driver that simply never had the
   * optional method.
   */
  async checkDriversHealth(opts?: { timeoutMs?: number }): Promise<DriverHealth[]> {
    const timeoutMs = opts?.timeoutMs ?? 2_000;

    return Promise.all(
      Array.from(this.drivers, async ([driverName, driver]): Promise<DriverHealth> => {
        if (typeof driver.checkHealth !== 'function') {
          return { driverName, healthy: true, skipped: true };
        }
        let timer: ReturnType<typeof setTimeout> | undefined;
        try {
          const healthy = await Promise.race([
            Promise.resolve(driver.checkHealth()),
            new Promise<never>((_, reject) => {
              timer = setTimeout(
                () => reject(new Error(`checkHealth did not settle within ${timeoutMs}ms`)),
                timeoutMs,
              );
              // Never hold the event loop open for a probe.
              (timer as { unref?: () => void }).unref?.();
            }),
          ]);
          return healthy
            ? { driverName, healthy: true }
            : { driverName, healthy: false, error: 'checkHealth() returned false' };
        } catch (e) {
          return {
            driverName,
            healthy: false,
            error: e instanceof Error ? e.message : String(e),
          };
        } finally {
          if (timer) clearTimeout(timer);
        }
      }),
    );
  }

  /**
   * Does this object declare a media field at all? Cached per schema object —
   * the registry hands back the same instance per object, so this scans a
   * field map once rather than on every write.
   */
  private objectHasMediaField(objectSchema: any): boolean {
    if (!objectSchema?.fields) return false;
    const cached = ObjectQL.mediaFieldPresence.get(objectSchema);
    if (cached !== undefined) return cached;
    const present = Object.values(objectSchema.fields).some(
      (def: any) => def && FILE_REFERENCE_TYPES.has(def.type),
    );
    ObjectQL.mediaFieldPresence.set(objectSchema, present);
    return present;
  }
  private static readonly mediaFieldPresence = new WeakMap<object, boolean>();

  /**
   * The media value-shape verdict for a write against `objectSchema`.
   *
   * Dormant by design, mirroring the storage module's own rule: an object with
   * no file-class field can hold no malformed media value, so it must not pay
   * even one query to learn that. Only objects that actually declare media
   * consult the deployment flag — and that read is itself memoized, so the
   * whole mechanism costs one query per process, and zero for an app that
   * stores no files.
   */
  private async mediaValueShapeStrictFor(objectSchema: any): Promise<boolean> {
    if (!this.objectHasMediaField(objectSchema)) return false;
    return this.isFileReferencesMigrationVerified();
  }

  /**
   * Does this object declare a reference or structured-JSON field that a write
   * would actually check? Same per-schema cache and same dormancy rule as
   * {@link objectHasMediaField}: an object holding none can hold no violation
   * of them, so it must not pay a query to learn that.
   *
   * The membership test is `isScannableValueShapeField` — the validator's own
   * — and not raw type membership, because the registry INJECTS covered-type
   * fields into every object it registers: `organization_id` and `owner_id`
   * (both `system`), plus `created_by` / `updated_by` (both in `SKIP_FIELDS`),
   * are all `lookup`s. `validateRecord` skips every one of them before it ever
   * reaches the value-shape check, so counting them made this answer `true` for
   * literally every object — the dormancy rule above never fired, and this
   * cache memoized a constant. Same predicate as the scanner for the same
   * reason the scanner imports it: three readings of "a covered field" drifting
   * by one clause is how a gate ends up governing fields nothing enforces.
   */
  private objectHasCoveredValueField(objectSchema: any): boolean {
    if (!objectSchema?.fields) return false;
    const cached = ObjectQL.coveredValueFieldPresence.get(objectSchema);
    if (cached !== undefined) return cached;
    const present = Object.entries(objectSchema.fields).some(
      ([name, def]: [string, any]) => isScannableValueShapeField(name, def),
    );
    ObjectQL.coveredValueFieldPresence.set(objectSchema, present);
    return present;
  }
  private static readonly coveredValueFieldPresence = new WeakMap<object, boolean>();

  /** The reference / structured-JSON value-shape verdict for a write. */
  private async valueShapeStrictFor(objectSchema: any): Promise<boolean> {
    if (!this.objectHasCoveredValueField(objectSchema)) return false;
    return this.isValueShapesMigrationVerified();
  }

  /**
   * Has this deployment completed AND verified the ADR-0104 file-as-reference
   * migration (#3617)? Read once per process from the `sys_migration` flag.
   *
   * Every way of not knowing answers `false` — no storage service (so no
   * `sys_migration` object registered), no row, an unreadable table, a
   * malformed row. Enforcement derives from evidence, and absent evidence is
   * not permission. Here "no" means media value shapes keep warning instead
   * of rejecting, so a deployment that cannot be asked keeps writing.
   *
   * Public because the flag's other in-process consumer reads it through this
   * same memoized seam: the storage service's release path (#3459 PR-5b) asks
   * it whether a released field file may be tombstoned, duck-typed as an
   * optional method so a fake or an older engine reads as "not verified".
   * One read, one invalidation (`invalidateDataMigrationFlags`), no way for
   * the two consumers to see different answers.
   *
   * Costs nothing on a kernel without the storage objects: the registry lookup
   * short-circuits before any query.
   */
  async isFileReferencesMigrationVerified(): Promise<boolean> {
    if (!this.fileReferencesMigrationVerified) {
      this.fileReferencesMigrationVerified = this.readMigrationFlagVerified(
        FILE_REFERENCES_MIGRATION_ID,
        '[value-shape] this deployment has verified the file-as-reference migration — ' +
          'media value shapes are enforced and released field files may be collected ' +
          '(ADR-0104 / #3617)',
      );
    }
    return this.fileReferencesMigrationVerified;
  }

  /**
   * Has this deployment completed AND verified the ADR-0104 non-media
   * value-shape scan (`os migrate value-shapes`, #3438)? Same memoized seam and
   * same fail-lenient posture as the file flag above — and a SEPARATE flag,
   * because it attests a different fact. The file migration says file values
   * were converted and reconciled; it says nothing about whether a `lookup` id
   * or a `location` payload is well formed, so it may not vouch for these
   * classes.
   */
  async isValueShapesMigrationVerified(): Promise<boolean> {
    if (!this.valueShapesMigrationVerified) {
      this.valueShapesMigrationVerified = this.readMigrationFlagVerified(
        VALUE_SHAPES_MIGRATION_ID,
        '[value-shape] this deployment has verified the value-shape scan — reference and ' +
          'structured-JSON value shapes are enforced (ADR-0104 / #3438)',
      );
    }
    return this.valueShapesMigrationVerified;
  }

  /**
   * Read one deployment migration flag and answer whether it authorises its
   * consumers. Shared by both flags so the "every way of not knowing answers
   * false" rule is written once: no `sys_migration` object registered, no row,
   * an unreadable table, a malformed row — all `false`. Enforcement derives
   * from evidence, and absent evidence is not permission.
   *
   * Costs nothing on a kernel without the platform objects: the registry
   * lookup short-circuits before any query.
   */
  private async readMigrationFlagVerified(migrationId: string, verifiedLog: string): Promise<boolean> {
    if (!this._registry.getObject(DATA_MIGRATION_FLAG_OBJECT)) return false;
    try {
      const rows = await this.find(DATA_MIGRATION_FLAG_OBJECT, {
        where: { id: migrationId },
        limit: 1,
        context: { isSystem: true } as ExecutionContextInput,
      });
      const row: any = rows?.[0];
      if (!row || row.id !== migrationId) return false;
      const verified = isDataMigrationFlagVerified({
        id: migrationId,
        last_run_at: String(row.last_run_at ?? ''),
        verified_at: row.verified_at == null ? null : String(row.verified_at),
        // A non-numeric count must read as "not zero", not as 0 — a bad
        // coercion lands on NaN, which fails the === 0 test.
        blocking: typeof row.blocking === 'number' ? row.blocking : Number(row.blocking ?? Number.NaN),
      });
      if (verified) this.logger.info(verifiedLog);
      return verified;
    } catch {
      return false; // unreadable evidence → stay lenient
    }
  }

  /**
   * Drop the memoized deployment migration flags so the next write re-reads
   * them. For a host that runs a data migration in-process and wants its
   * effect without a restart.
   */
  invalidateDataMigrationFlags(): void {
    this.fileReferencesMigrationVerified = null;
    this.valueShapesMigrationVerified = null;
    this.migrationGatesAnnounced = false;
  }

  /**
   * Say, once at boot, which value-shape gates are still open here and what
   * closes them (ADR-0104's 2026-07-30 addendum, #3438).
   *
   * Both gates fail toward leniency: a deployment that never runs its
   * migration keeps warning instead of rejecting, and keeps every released
   * file forever. That default is deliberate — but silent, and a gate nobody
   * is told about is served by nobody. The LAX posture is the one worth
   * announcing, so this logs only when a gate is open *and* this deployment's
   * own metadata says the gate is about something it stores; the verified case
   * already logs from the flag read.
   *
   * A gate whose posture an environment switch has already settled says
   * nothing either, because the flag this line reports on is not what decides
   * that deployment's posture: under `OS_DATA_VALUE_SHAPE_STRICT_ENABLED`
   * enforcement is already on, so "checked but NOT enforced here" is simply
   * false, and under either opt-out the operator chose leniency deliberately,
   * so naming a migration that would not change what they get is noise. Each
   * gate consults its own pair, since the opt-outs are per-class. Cheapest
   * test first: a kernel with nothing to say never reaches the flag query.
   *
   * Costs one flag read per applicable gate (both memoized, and both already
   * paid by the first write to such an object), and nothing at all for an app
   * that declares neither class of field.
   */
  async announceOpenMigrationGates(): Promise<void> {
    if (this.migrationGatesAnnounced) return;
    this.migrationGatesAnnounced = true;
    try {
      const mediaByEnv = mediaPostureSetByEnv();
      const coveredByEnv = valueShapePostureSetByEnv();
      if (mediaByEnv && coveredByEnv) return;

      let media = false;
      let covered = false;
      for (const obj of this._registry.getAllObjects() as any[]) {
        if (!media && !mediaByEnv && this.objectHasMediaField(obj)) media = true;
        if (!covered && !coveredByEnv && this.objectHasCoveredValueField(obj)) covered = true;
        if ((media || mediaByEnv) && (covered || coveredByEnv)) break;
      }

      if (media && !(await this.isFileReferencesMigrationVerified())) {
        this.logger.info(
          '[value-shape] media values are checked but NOT enforced here, and released files are ' +
            'never collected — this deployment has not verified its file migration. Run ' +
            '`os migrate files-to-references` (dry run) to see what it would do, then `--apply` ' +
            'to close the gate (ADR-0104 / #3617).',
        );
      }
      if (covered && !(await this.isValueShapesMigrationVerified())) {
        this.logger.info(
          '[value-shape] reference and structured-JSON values are checked but NOT enforced here — ' +
            'this deployment has not verified its value-shape scan. Run `os migrate value-shapes` ' +
            '(dry run) to see what it would report, then `--apply` to close the gate ' +
            '(ADR-0104 / #3438).',
        );
      }
    } catch {
      // An advisory must never be the reason a boot fails.
    }
  }
  private migrationGatesAnnounced = false;

  /**
   * Did this process CREATE the datastore it is talking to, from empty?
   *
   * True only when every driver that can account for its schema sync created
   * tables and found none already there. That conjunction is the whole point:
   * one pre-existing table anywhere means something ran here before us, so
   * this store's history is not ours to vouch for. A driver that cannot
   * report (`getSchemaSyncStats` absent, deferred DDL, sync skipped) makes the
   * answer no rather than maybe — the consumers of this fact grant
   * permissions, so "cannot say" must read as "no".
   *
   * The one caller is the fresh-datastore attestation (#3438, ADR-0104's
   * 2026-07-30 addendum), which records that a store born empty needs no data
   * migration. Ask it only during boot: once the process starts serving, the
   * counts describe a moment that has passed.
   */
  wasDatastoreCreatedFromEmpty(): boolean {
    let created = 0;
    let existing = 0;
    let reporting = 0;
    for (const driver of this.drivers.values()) {
      const stats = (driver as any).getSchemaSyncStats?.();
      if (!stats) continue;
      reporting += 1;
      created += Number(stats.created) || 0;
      existing += Number(stats.existing) || 0;
    }
    return reporting > 0 && existing === 0 && created > 0;
  }

  async destroy() {
    this.logger.info('Destroying ObjectQL engine', { driverCount: this.drivers.size });
    
    for (const [name, driver] of this.drivers.entries()) {
      try {
        await driver.disconnect();
      } catch (e) {
        this.logger.error('Error disconnecting driver', e as Error, { driverName: name });
      }
    }
    
    this.logger.info('ObjectQL engine destroyed');
  }

  // ============================================
  // Helper: Expand Related Records
  // ============================================

  /** Maximum depth for recursive expand to prevent infinite loops */
  private static readonly MAX_EXPAND_DEPTH = 3;
  private static readonly MAX_CASCADE_DEPTH = 10;
  /** In-memory next-value cache per `object.field` for autonumber generation,
   *  lazily seeded from the current max in the store. */
  private readonly autonumberCounters = new Map<string, number>();

  /**
   * Memoized answer to "has THIS deployment completed and verified the
   * ADR-0104 file-as-reference migration?" (#3617) — the fact that decides
   * whether a malformed media value rejects or merely warns.
   *
   * Memoized as a PROMISE so concurrent first writes share one read rather
   * than racing several. Deliberately not refreshed on a timer: the flag is
   * written by `os migrate files-to-references --apply`, a deliberate
   * operator action, and a process that has not seen it yet simply stays
   * lenient — the safe direction. A host that migrates in-process can call
   * {@link invalidateDataMigrationFlags} instead of waiting for a restart.
   */
  private fileReferencesMigrationVerified: Promise<boolean> | null = null;
  private valueShapesMigrationVerified: Promise<boolean> | null = null;

  /** Lazily-built index: child object name → roll-up summary descriptors on
   *  parent objects that aggregate it. Invalidated when packages register. */
  private summaryIndex: Map<string, SummaryDescriptor[]> | null = null;

  /**
   * Retry options for roll-up summary recompute (framework#3147). Public so a
   * test can inject a no-op sleep for deterministic backoff; production uses
   * the transient-retry defaults.
   */
  summaryRetryOptions: RetryOptions = {};

  /** Invalidate the cached roll-up summary index (call when metadata changes). */
  private invalidateSummaryIndex(): void {
    this.summaryIndex = null;
  }

  /** Scan all registered objects for `summary` fields and index them by the
   *  child object they aggregate, resolving the child→parent FK field. */
  private buildSummaryIndex(): Map<string, SummaryDescriptor[]> {
    const index = new Map<string, SummaryDescriptor[]>();
    let objects: any[] = [];
    try { objects = (this._registry as any).getAllObjects?.() ?? []; } catch { objects = []; }
    for (const parent of objects) {
      const fields = parent?.fields;
      if (!fields || typeof fields !== 'object' || Array.isArray(fields)) continue;
      for (const [summaryField, def] of Object.entries(fields)) {
        const d: any = def;
        if (d?.type !== 'summary' || !d.summaryOperations) continue;
        const so = d.summaryOperations;
        const childObject = so.object;
        const fn = so.function;
        if (!childObject || !fn) continue;
        // Resolve the FK on the child pointing back to this parent.
        let fkField: string | undefined = so.relationshipField;
        if (!fkField) {
          const child = this._registry.getObject(childObject) as any;
          const cfields = child?.fields || {};
          for (const [cfName, cdef] of Object.entries(cfields)) {
            const cd: any = cdef;
            if ((cd?.type === 'master_detail' || cd?.type === 'lookup') && cd?.reference === parent.name) {
              fkField = cfName;
              break;
            }
          }
        }
        if (!fkField) continue; // can't resolve the relationship — skip
        // Optional per-summary predicate: only child rows matching it are
        // aggregated (e.g. sum receipts where { status: 'received' }). ANDed with
        // the parent-FK match at recompute time. Ignore a non-object filter.
        const filter = so.filter && typeof so.filter === 'object' && !Array.isArray(so.filter)
          ? so.filter as Record<string, unknown>
          : undefined;
        const list = index.get(childObject) ?? [];
        list.push({ parentObject: parent.name, summaryField, fkField, fn, sourceField: so.field, filter });
        index.set(childObject, list);
      }
    }
    return index;
  }

  /** `registry.objectRevision` the cached {@link summaryIndex} was built at. */
  private summaryIndexRevision = -1;

  private getSummaryDescriptors(childObject: string): SummaryDescriptor[] {
    // Rebuild whenever the REGISTRY's object set has moved since the index was
    // built — not only when someone remembered to call
    // `invalidateSummaryIndex`. That single site (`registerApp`) is bypassed by
    // the runtime publish path, which registers straight into the registry
    // (`protocol.saveMetaItem` → `registry.registerObject`). A kernel that had
    // already done one write — publishing itself writes `sys_metadata` — held an
    // index built before the new object existed, so every child write of a
    // freshly published roll-up skipped the recompute and the parent read null
    // until the process restarted. That is exactly how an AI-built app's
    // "已完成任务数" shipped empty over correct metadata (cloud#970).
    const revision = (this._registry as unknown as { objectRevision?: number })?.objectRevision;
    const stale = typeof revision === 'number' && revision !== this.summaryIndexRevision;
    if (!this.summaryIndex || stale) {
      this.summaryIndex = this.buildSummaryIndex();
      if (typeof revision === 'number') this.summaryIndexRevision = revision;
    }
    return this.summaryIndex.get(childObject) ?? [];
  }

  /**
   * Recompute roll-up `summary` fields on parent records after a child write.
   * For each affected parent (the FK value on the changed/old child record), it
   * aggregates the child collection and writes the result onto the parent's
   * summary field. Runs in the caller's execution context so it joins the same
   * transaction (e.g. the cross-object batch) when one is open.
   */
  private async recomputeSummaries(
    childObject: string,
    records: any,
    previous: any,
    execCtx?: ExecutionContextInput,
  ): Promise<SummaryRecomputeFailure[]> {
    const descriptors = this.getSummaryDescriptors(childObject);
    if (descriptors.length === 0) return [];
    const recs = Array.isArray(records) ? records : records ? [records] : [];
    const prevs = Array.isArray(previous) ? previous : previous ? [previous] : [];
    const failures: SummaryRecomputeFailure[] = [];
    for (const desc of descriptors) {
      const ids = new Set<string>();
      for (const r of recs) { const v = r?.[desc.fkField]; if (v != null && v !== '') ids.add(String(v)); }
      for (const p of prevs) { const v = p?.[desc.fkField]; if (v != null && v !== '') ids.add(String(v)); }
      for (const parentId of ids) {
        try {
          // Retry a transient failure (a turso `fetch failed` on the parent
          // aggregate/update) with backoff — a network blip here used to leave
          // the parent summary silently stale (framework#3147).
          await withTransientRetry(async () => {
            // AND the parent-FK match with the optional per-summary filter so
            // only matching child rows are aggregated (e.g. received receipts).
            const fkMatch = { [desc.fkField]: parentId };
            const where = desc.filter ? { $and: [fkMatch, desc.filter] } : fkMatch;
            const rows = await this.aggregate(childObject, {
              where,
              aggregations: [{
                function: desc.fn,
                ...(desc.fn === 'count' ? {} : { field: desc.sourceField }),
                alias: 'value',
              }],
              context: execCtx,
            } as any);
            let value = rows?.[0]?.value;
            if (value == null) value = (desc.fn === 'count' || desc.fn === 'sum') ? 0 : null;
            await this.update(desc.parentObject, { id: parentId, [desc.summaryField]: value }, { context: execCtx } as any);
          }, this.summaryRetryOptions);
        } catch (err) {
          // Retries exhausted (or a non-transient failure). Record it so the
          // caller can surface it — one parent's failure must not abort the
          // remaining parents' recompute.
          this.logger.warn('Roll-up summary recompute failed', {
            childObject, parentObject: desc.parentObject, parentId, field: desc.summaryField,
            error: (err as any)?.message,
          });
          failures.push({ childObject, parentObject: desc.parentObject, parentId, field: desc.summaryField, error: err });
        }
      }
    }
    return failures;
  }

  /**
   * Post-process expand: resolve lookup/master_detail fields by batch-loading related records.
   * 
   * This is a driver-agnostic implementation that uses secondary queries ($in batches)
   * to load related records, then injects them into the result set.
   * 
   * @param objectName - The source object name
   * @param records - The records returned by the driver
   * @param expand - The expand map from QueryAST (field name → nested QueryAST)
   * @param depth - Current recursion depth (0-based)
   * @returns Records with expanded lookup fields (IDs replaced by full objects)
   */
  private async expandRelatedRecords(
    objectName: string,
    records: any[],
    expand: Record<string, QueryAST>,
    depth: number = 0,
    execCtx?: ExecutionContextInput,
  ): Promise<any[]> {
    if (!records || records.length === 0) return records;
    if (depth >= ObjectQL.MAX_EXPAND_DEPTH) return records;

    const objectSchema = this._registry.getObject(objectName);
    // If no schema registered, skip expand — return raw data
    if (!objectSchema || !objectSchema.fields) return records;

    for (const [fieldName, nestedAST] of Object.entries(expand)) {
      // [#4371] The nested AST is caller-authored too: a wire spelling inside
      // `expand: { rel: { sort } }` used to be silently dropped exactly like
      // the top-level bag (this loop reads only canonical keys). Same
      // rejection, scoped to the four wire-only pairs — the nested shape is a
      // QueryAST, not an option bag, so the option-key gate does not apply.
      if (nestedAST && typeof nestedAST === 'object') {
        for (const slot of ENGINE_WIRE_ONLY_SLOTS) {
          for (const alias of slot.aliases) {
            if ((nestedAST as Record<string, unknown>)[alias] != null) {
              throw new Error(
                `expand['${fieldName}'] on '${objectName}' does not accept '${alias}': it is a wire ` +
                `spelling of '${slot.canonical}', folded by the RPC/protocol layer — a direct engine ` +
                `call bypasses that fold, so the value would be silently dropped, not applied. Pass ` +
                `'${slot.canonical}' (${WIRE_ONLY_CANONICAL_SHAPES[slot.canonical] ?? 'the canonical QueryAST shape'}) instead.`,
              );
            }
          }
        }
      }
      const fieldDef = objectSchema.fields[fieldName];

      // Skip if field not found or not a relationship type.
      //
      // `user` is a lookup specialized to sys_user and `tree` is a hierarchical
      // self-reference — both carry the same `reference` + id storage, so both
      // expand through this exact path (single or multiple).
      //
      // [#4226] The membership test is `REFERENCE_VALUE_TYPES` — the spec's own
      // list of types whose value "points at another record … the related record
      // object in expanded form" — rather than a hand-copied `!==` chain. The
      // chain had drifted: it omitted `tree`, so a `$expand` of a hierarchy
      // field returned the raw parent id and rendered as a bare placeholder,
      // while `field-value.zod.ts` and objectui's `EXPANDABLE_FIELD_TYPES` both
      // declared it expandable. Reading the shared set is what stops the
      // protocol's expand gate (which validates against the same set) from ever
      // admitting a field this loop then silently skips.
      //
      // [cloud#983] The TARGET comes from `referenceTargetOf` for that same
      // anti-drift reason. A raw `fieldDef.reference` read made `{ type:
      // 'user' }` (no `reference`) targetless here AND at the gate — but
      // `user`'s target is fixed BY THE TYPE (`sys_user`; `Field.user()` takes
      // no target argument), so the field was fully specified and the request
      // was refused `400 … declares no target object`. Both sides now ask the
      // one function what a reference field points at.
      if (!fieldDef) continue;
      if (!REFERENCE_VALUE_TYPES.has(fieldDef.type)) continue;
      const referenceObject = referenceTargetOf(fieldDef);
      if (!referenceObject) continue;

      // Collect all foreign key IDs from records (handle both single and multiple values)
      const allIds: any[] = [];
      for (const record of records) {
        const val = record[fieldName];
        if (val == null) continue;
        if (Array.isArray(val)) {
          allIds.push(...val.filter((id: any) => id != null));
        } else if (typeof val === 'object') {
          // Already expanded — skip
          continue;
        } else {
          allIds.push(val);
        }
      }

      // De-duplicate IDs
      const uniqueIds = [...new Set(allIds)];
      if (uniqueIds.length === 0) continue;

      // Batch-load related records using $in query
      try {
        // Constrain the batch to the collected FK ids. When the nested expand
        // AST carries its own `where`, AND-merge it via an explicit `$and`
        // group rather than spreading keys onto the id filter: a shallow spread
        // would clobber the `id` constraint (if the nested filter also keys
        // `id`) and is ambiguous when the nested filter is a top-level `$or` /
        // `$and`. The `$and` wrapper is correct for every FilterCondition shape.
        const idFilter = { id: { $in: uniqueIds } };
        const where = nestedAST.where
          ? { $and: [idFilter, nestedAST.where] }
          : idFilter;

        // [#2850] Resolve the referenced object through the engine's own read
        // path (`this.find`) rather than the raw driver, so the security
        // middleware applies the REFERENCED object's RLS + FLS to the expanded
        // batch — not merely tenant isolation. A single `find` over the
        // collected `id $in [...]` batch re-enters the middleware exactly once
        // (no N+1), preserving the batched load. `nestedAST.limit/offset` are
        // still intentionally NOT forwarded: this path batch-loads every
        // parent's related records in one query, so a *per-parent* limit/offset
        // can't be expressed — a global cap would silently drop records other
        // parents need. Paginate by querying the related object directly.
        //
        // The `__expandRead` marker (set here, never from client input —
        // `executionContext` is server-built) tells the security layer this is
        // an expansion sub-read: it waives ONLY the object-level CRUD /
        // requiredPermissions gate for PUBLIC referenced objects (already
        // broadly readable, so a common status/owner lookup isn't over-blocked),
        // while PRIVATE referenced objects keep the full RLS + CRUD treatment
        // (you may expand only rows you could read directly). FLS masking
        // applies to both. `expand` is intentionally omitted from this query so
        // `find` does not re-expand — nested relations recurse below under the
        // depth guard.
        const relatedRecords = await this.find(
          referenceObject,
          {
            where,
            ...(nestedAST.fields ? { fields: nestedAST.fields as any } : {}),
            ...(nestedAST.orderBy ? { orderBy: nestedAST.orderBy as any } : {}),
            context: { ...(execCtx ?? {}), __expandRead: true } as ExecutionContextInput,
          },
        ) ?? [];

        // Build a lookup map: id → record
        const recordMap = new Map<string, any>();
        for (const rec of relatedRecords) {
          const id = rec.id;
          if (id != null) recordMap.set(String(id), rec);
        }

        // Recursively expand nested relations if present
        if (nestedAST.expand && Object.keys(nestedAST.expand).length > 0) {
          const expandedRelated = await this.expandRelatedRecords(
            referenceObject,
            relatedRecords,
            nestedAST.expand,
            depth + 1,
            execCtx,
          );
          // Rebuild map with expanded records
          recordMap.clear();
          for (const rec of expandedRelated) {
            const id = rec.id;
            if (id != null) recordMap.set(String(id), rec);
          }
        }

        // Inject expanded records back into the original result set
        for (const record of records) {
          const val = record[fieldName];
          if (val == null) continue;

          if (Array.isArray(val)) {
            record[fieldName] = val.map((id: any) => recordMap.get(String(id)) ?? id);
          } else if (typeof val !== 'object') {
            record[fieldName] = recordMap.get(String(val)) ?? val;
          }
          // If val is already an object, leave it as-is
        }
      } catch (e) {
        // Graceful degradation: if expand fails, keep original IDs
        this.logger.warn('Failed to expand relationship field; retaining foreign key IDs', {
          object: objectName,
          field: fieldName,
          reference: referenceObject,
          error: (e as Error).message,
        });
      }
    }

    return records;
  }

  /**
   * Resolve file-field id references to their expanded `FileValueSchema` form
   * (ADR-0104 D3 wave 2). A `file`/`image`/`avatar`/`video`/`audio` value
   * stored as an opaque `sys_file` id string is enriched, in place, to
   * `{ id, name, size, mimeType, url }` — `url` derived from the stable
   * `/files/:fileId` resolver, never stored.
   *
   * DUAL-MODE SAFE: an inline-blob value (an object) and a string that does
   * NOT match a committed `sys_file` row (e.g. an external url) pass through
   * unchanged, so a field may hold either form during the pre-v17 window.
   *
   * Batched: at most one `sys_file` `id $in […]` read per call (no N+1); and
   * zero reads when no field holds a string value (the blob-only case), so the
   * step is free for objects that have not adopted references.
   */
  private async resolveFileReferences(
    objectName: string,
    records: any[],
    execCtx?: ExecutionContextInput,
  ): Promise<any[]> {
    if (!records || records.length === 0) return records;
    // A caller whose subject is the STORED form — the ADR-0104 backfill /
    // reconciliation (#3617) — opts out of resolution entirely; expanding
    // ids before that scan would hide the very state it exists to audit.
    if ((execCtx as any)?.[RAW_FILE_VALUES_CONTEXT_KEY] === true) return records;
    const objectSchema = this._registry.getObject(objectName);
    if (!objectSchema || !objectSchema.fields) return records;
    // Nothing to resolve against if the file object is not even registered
    // (e.g. the storage plugin is absent) — skip without a failing query.
    if (!this._registry.getObject('sys_file')) return records;

    const fileFields = Object.entries(objectSchema.fields)
      .filter(([, def]: [string, any]) => def && FILE_REFERENCE_TYPES.has(def.type))
      .map(([name]) => name);
    if (fileFields.length === 0) return records;

    // Collect candidate ids. A file field legitimately holds either an inline
    // blob (an object — already the rich form, skipped) OR, in the dual-mode /
    // legacy world, a URL string (`https://…`, `/api/…`, `data:…`, `blob:…`).
    // Only an OPAQUE id token — the minted uuid/nanoid form, never url-shaped —
    // is a `sys_file` reference to resolve. Filtering out url-shaped strings is
    // what keeps a seeded `data:`/CDN image value from firing a bogus lookup.
    const candidateIds: string[] = [];
    const addCandidate = (v: unknown) => {
      if (isFileIdToken(v)) candidateIds.push(v);
    };
    for (const record of records) {
      for (const fieldName of fileFields) {
        const val = record[fieldName];
        if (val == null) continue;
        if (Array.isArray(val)) {
          for (const v of val) addCandidate(v);
        } else {
          addCandidate(val);
        }
      }
    }
    const uniqueIds = [...new Set(candidateIds)];
    if (uniqueIds.length === 0) return records; // blob-only / empty → zero cost

    // One batched sys_file read. `__expandRead` mirrors the lookup-expansion
    // sub-read (a system-built marker, never from client input). Metadata only —
    // byte-download authorization is enforced at the /files/:fileId resolver.
    let fileRows: any[] = [];
    try {
      fileRows = (await this.find(
        'sys_file',
        { where: { id: { $in: uniqueIds } }, context: { ...(execCtx ?? {}), __expandRead: true } as ExecutionContextInput },
      )) ?? [];
    } catch {
      return records; // sys_file unregistered / unreadable — leave ids as-is
    }

    const fileMap = new Map<string, any>();
    for (const row of fileRows) {
      if (row?.id != null && row.status === 'committed') fileMap.set(String(row.id), row);
    }
    if (fileMap.size === 0) return records;

    const toValue = (row: any) => ({
      id: String(row.id),
      ...(row.name != null ? { name: row.name } : {}),
      ...(row.size != null ? { size: row.size } : {}),
      ...(row.mime_type != null ? { mimeType: row.mime_type } : {}),
      url: `/api/v1/storage/files/${row.id}`,
    });

    for (const record of records) {
      for (const fieldName of fileFields) {
        const val = record[fieldName];
        if (val == null) continue;
        if (Array.isArray(val)) {
          record[fieldName] = val.map((v: any) =>
            typeof v === 'string' && fileMap.has(v) ? toValue(fileMap.get(v)) : v);
        } else if (typeof val === 'string' && fileMap.has(val)) {
          record[fieldName] = toValue(fileMap.get(val));
        }
      }
    }
    return records;
  }

  // ============================================
  // Data Access Methods (IDataEngine Interface)
  // ============================================

  /**
   * Expand `{filter-placeholder}` values in a read AST's `where` against the
   * request (framework#3582).
   *
   * Filters travel as JSON, so a time- or user-scoped slice authored in a view,
   * dashboard, related list or REST query writes `'{current_year_start}'` /
   * `'{current_user_id}'` rather than a literal. Until now nothing on the server
   * substituted them: the placeholder reached the driver as a string, compared
   * as text, and matched nothing — an empty grid with no error anywhere.
   *
   * The engine is the right seam because it is the ONE gate every server-side
   * read passes through (REST, SDK, related lists, flow `find_records`, sharing
   * graph reads), so a surface that follows the filter contract works the day it
   * ships instead of waiting for its own resolver. It runs BEFORE the middleware
   * chain so only author-supplied filters are inspected; the RLS/sharing filters
   * injected downstream are built from concrete context values and carry no
   * placeholders.
   *
   * Cheap by construction: {@link resolveFilterTokens} returns the input by
   * reference when the tree holds no placeholder, which is every internal query.
   * An unresolvable placeholder throws (see the resolver's module doc) — the one
   * outcome an author can act on.
   */
  private resolveWhereTokens(ast: QueryAST | undefined, execCtx?: ExecutionContextInput): void {
    if (!ast || ast.where == null) return;
    ast.where = resolveFilterTokens(ast.where, filterTokenContextFrom(execCtx));
  }

  /**
   * The write-path counterpart of {@link resolveWhereTokens}: return `options`
   * with `where` placeholders expanded (#3810).
   *
   * Returns the SAME object when nothing resolved — `resolveFilterTokens`
   * returns its input by reference on a placeholder-free tree, so the common
   * path allocates nothing. When something does resolve, a shallow copy is
   * made rather than assigning through: `options` belongs to the caller, and
   * writing back would bake one request's user id into a filter object the
   * caller may reuse (view metadata and flow node config both get reused).
   */
  private withResolvedWhere<T extends { where?: unknown; context?: ExecutionContextInput } | undefined>(
    options: T,
  ): T {
    if (!options || options.where == null) return options;
    const resolved = resolveFilterTokens(options.where, filterTokenContextFrom(options.context));
    return resolved === options.where ? options : ({ ...options, where: resolved } as T);
  }

  /**
   * ADR-0061: expand `search` into a server-resolved cross-field `$or` of
   * `$contains`, AND it with any caller `where`, then strip the search keys off
   * the AST.
   *
   * Shared by `find` and `findOne` (#4419). It lived inline in `find` and
   * nowhere else, while `ENGINE_FIND_OPTION_KEYS` — the one legal-key set BOTH
   * methods are checked against (see {@link ENGINE_OPTION_KEY_SETS}) — declares
   * `search`/`searchFields` for both. So `findOne({ search })` passed the gate,
   * rode onto the AST verbatim, and reached a driver: no driver reads
   * `ast.search` (the expansion is the engine's job by ADR-0061), so the
   * predicate vanished and the forced `limit: 1` turned it into the first row of
   * the WHOLE object — a real, plausible-looking record unrelated to the search.
   * That is #4419's reported failure exactly, under a different key than the
   * `filter` #4346 closed; one expander, called from both, is what stops the
   * pair drifting again.
   *
   * Field resolution is server-side (declared `searchableFields` →
   * auto-default); the optional `searchFields` override is intersected with the
   * allowed set, never widened. All drivers already execute `$or`/`$contains`,
   * so this needs no driver changes.
   *
   * The keys are deleted whether or not anything expanded — leaving them on
   * would hand the driver a key it does not read, which is the same silent drop
   * one layer down.
   */
  private expandSearchOnAst(ast: QueryAST, schema: ServiceObject | undefined): void {
    // The `$search`/`$searchFields` OData spellings are NOT read here: the
    // protocol layer normalizes them to the bare keys before the engine
    // (protocol.ts findData), and a direct engine call carrying one is an
    // unknown option — rejected at the entry point, not silently dropped
    // (#4371).
    const raw = (ast as any).search;
    if (raw != null && schema?.fields) {
      const requestedFields = (ast as any).searchFields
        ?? (typeof raw === 'object' ? raw?.fields : undefined);
      const searchFilter = expandSearchToFilter(raw, {
        fields: schema.fields as any,
        searchableFields: (schema as any).searchableFields,
        requestedFields,
        // [ADR-0079] `nameField` is the canonical primary-title pointer;
        // `displayNameField` is the deprecated alias (still honored).
        displayField: (schema as any).nameField ?? (schema as any).displayNameField,
      });
      if (searchFilter) {
        ast.where = ast.where ? { $and: [ast.where, searchFilter] } : searchFilter;
      }
    }
    delete (ast as any).search;
    delete (ast as any).searchFields;
  }

  /**
   * Refuse a `findOne` that selects nothing in particular (#4419).
   *
   * The AST reaching here is the CALLER's own intent: aliases folded, unknown
   * keys refused, `search` expanded — but the security/sharing middlewares have
   * not run yet, and that ordering is the point. An injected RLS predicate
   * narrows *which* rows are visible; it does not make "whichever of them comes
   * first" a thing the caller asked for. Judging the post-middleware AST would
   * pass every query on a scoped object and leave the hole open where it is
   * most expensive.
   *
   * "Selects nothing" is read the same way #3896 read an empty sharing
   * criteria: absent, `null`, or `{}` — the three shapes that mean "match every
   * row". A `where` that is not a plain object (an expression tree) is the
   * driver's to interpret, and counts as a predicate; this guard closes the one
   * case that is unambiguously match-everything, not everything it cannot
   * prove.
   *
   * `orderBy` is the other way to be specific, and a legitimate one — "the
   * newest", "the highest priority". It is honored on this path by every
   * driver, so it is a real answer and not a second silent drop.
   */
  private requireFindOnePredicate(object: string, ast: QueryAST): void {
    const where = ast.where as unknown;
    const hasPredicate =
      where != null &&
      (typeof where !== 'object' || Array.isArray(where) || Object.keys(where).length > 0);
    if (hasPredicate) return;
    if (Array.isArray(ast.orderBy) && ast.orderBy.length > 0) return;
    throw new Error(
      `findOne('${object}') selects no particular record: 'where' is absent or empty ` +
      `and the query carries no 'orderBy'. findOne applies limit: 1, so this would return an ` +
      `ARBITRARY row — a real, plausible-looking record unrelated to what was asked for, which ` +
      `no caller's null-check can catch (#4419). Pass 'where' (or a 'search' that resolves to ` +
      `one) to select the record; pass 'orderBy' if you mean "the first record in THIS order"; ` +
      `or call find('${object}', { limit: 1 }) if any row will genuinely do.`,
    );
  }

  async find(object: string, query?: EngineQueryOptions, options?: EngineReadOptions): Promise<any[]> {
    object = this.resolveObjectName(object);
    // Normalize the alias spellings (`filter`→`where`, `top`→`limit`) by the
    // spec's slot table — the driver AST only understands the canonical keys,
    // so an unfolded `{ filter }` would match ALL rows (silent over-grant —
    // surfaced by ADR-0057's sharing/graph read path). Same fold, same
    // conflict rule on every engine entry point (#4346). The wire-only
    // spellings (`sort`/`select`/`skip`/`populate`) are REJECTED instead of
    // folded — a direct call carrying one used to have it silently dropped
    // (#4371, three shipped instances in #4370).
    query = foldEngineOptionAliases(object, 'find', query, ENGINE_QUERY_SLOTS, ENGINE_WIRE_ONLY_SLOTS);
    rejectUnknownEngineOptions(object, 'find', query, ENGINE_FIND_OPTION_KEYS);
    this.logger.debug('Find operation starting', { object, query });
    const driver = this.getDriver(object);
    // `object` LAST: the resolved name must win. Spread-first used to let a
    // stray `query.object` overwrite it, splitting the AST's object from the
    // table actually queried (#4371 option-2 survey) — every middleware and
    // hook reading `ast.object` would have been lied to.
    const ast: QueryAST = { ...query, object };
    // Remove context from the AST — it's not a driver concern
    delete (ast as any).context;

    // Plan formula projection: rewrite ast.fields to drop virtual formula
    // names and inject their dependencies, so the driver returns the raw
    // fields needed to compute the formulas after fetch.
    const _findSchema = this._registry.getObject(object);

    this.expandSearchOnAst(ast, _findSchema);
    const _findFormula = planFormulaProjection(_findSchema, ast.fields);
    if (_findFormula.projected) ast.fields = _findFormula.projected;

    // Drop any requested field that doesn't exist on the schema. Without
    // this, drivers (notably SqlDriver) emit `SELECT unknown_col FROM ...`
    // which the DB rejects ("no such column") — and SqlDriver swallows
    // that error and returns `[]`, making a frontend bug (e.g. a generic
    // view requesting `name`/`due_date` on every object) look like "no
    // records exist". Silently filtering matches the existing OData
    // tolerance and Salesforce/Postgres behavior of `SELECT *` semantics.
    if (_findSchema?.fields && Array.isArray(ast.fields) && ast.fields.length > 0) {
      const known = new Set(Object.keys(_findSchema.fields));
      // Always allow the primary key + audit columns even if not present in
      // schema.fields. Without this, callers requesting `select=id,name`
      // silently get the `id` projected away, breaking record navigation.
      known.add('id');
      known.add('created_at');
      known.add('updated_at');
      const filtered = ast.fields.filter(f => {
        // Keep relationship paths like `owner.name` — the engine will
        // resolve those via populate; only validate top-level segment.
        const head = f.split('.')[0];
        return known.has(head);
      });
      // Guard against an empty projection — fall back to `*` so the
      // request still returns rows. An empty SELECT list would either
      // 400 in Postgres or silently project nothing.
      ast.fields = filtered.length > 0 ? filtered : undefined;
    }

    const opCtx: OperationContext = {
      object,
      operation: 'find',
      ast,
      options: query,
      context: mergeReadContext(query?.context, options?.context),
    };
    this.resolveWhereTokens(opCtx.ast as QueryAST, opCtx.context);

    await this.executeWithMiddleware(opCtx, async () => {
      const hookContext: HookContext = {
          object,
          event: 'beforeFind',
          input: { ast: opCtx.ast, options: opCtx.options },
          session: this.buildSession(opCtx.context),
          provenance: this.buildProvenance(opCtx.context),
          user: this.buildUser(opCtx.context),
          api: this.buildHookApi(opCtx.context),
          transaction: opCtx.context?.transaction,
          ql: this
      };
      await this.triggerHooks('beforeFind', hookContext);
      hookContext.input.options = this.buildDriverOptions(object, opCtx.context, hookContext.input.options as any);

      try {
          let result = await driver.find(object, hookContext.input.ast as QueryAST, hookContext.input.options as any);

          // Post-process: evaluate formula virtual fields against the raw rows
          if (Array.isArray(result)) applyFormulaPlan(_findFormula.plan, result, opCtx.context);

          // Post-process: expand related records if expand is requested
          if (ast.expand && Object.keys(ast.expand).length > 0 && Array.isArray(result)) {
            result = await this.expandRelatedRecords(object, result, ast.expand, 0, opCtx.context);
          }

          // Post-process: resolve file-field id references to their expanded
          // FileValueSchema form (ADR-0104 D3). Always-on but free unless a file
          // field holds an id string; dual-mode-safe (blobs pass through).
          if (Array.isArray(result)) {
            result = await this.resolveFileReferences(object, result, opCtx.context);
          }

          hookContext.event = 'afterFind';
          hookContext.result = result;
          await this.triggerHooks('afterFind', hookContext);

          // Never let secret-field plaintext (or its ref) leave through the
          // generic read path — mask after hooks run. Privileged consumers use
          // resolveSecret() against the stored ref instead.
          this.maskSecretFields(object, hookContext.result);

          return hookContext.result;
      } catch (e) {
          this.logger.error('Find operation failed', e as Error, { object });
          throw e;
      }
    });

    return opCtx.result as any[];
  }

  /**
   * Read the ONE record the query selects, or `null`.
   *
   * `findOne` applies `limit: 1` by contract — so unlike `find`, the query's
   * predicate is the only thing standing between the caller and *an arbitrary
   * row*. A query that selects nothing in particular does not return nothing;
   * it returns the object's first row, which is a real, plausible-looking
   * record that no caller's `if (!row)` check can catch, and that propagates
   * into whatever is computed next (#4419). So this method REQUIRES the caller
   * to say which record it wants:
   *
   * - `where` (or the `filter` alias, folded here), or a `search` that expands
   *   to one — the record is selected by predicate.
   * - `orderBy` — "the FIRST record in this order" (the newest, the highest
   *   priority). Deterministic without a predicate, and honored by every
   *   driver on this path.
   *
   * Neither → throws. If any row genuinely will do, that is
   * `find(object, { limit: 1 })`, which says so at the call site.
   *
   * No ordering is IMPOSED when the caller supplies none: `ORDER BY <pk> LIMIT
   * 1` makes a planner abandon the predicate's own index (objectstack#4363, and
   * see `SqlDriver.findRows`' `singleRowLookup`). `findOne` promises *a*
   * matching record, never a position in a sequence.
   *
   * Fires the same `beforeFind`/`afterFind` hooks as `find` (#3195).
   */
  async findOne(objectName: string, query?: EngineQueryOptions, options?: EngineReadOptions): Promise<any> {
    objectName = this.resolveObjectName(objectName);
    // Same alias fold as find() (#4346). Without it, `findOne({ filter })`
    // matched the first row of the WHOLE table rather than the predicate.
    // `top` folds into `limit` here too, but findOne is single-row by
    // contract, so the literal `limit: 1` below wins over both spellings.
    // Wire-only spellings are rejected, same as find() (#4371) — `sort`
    // matters here too: findOne({ sort }) means "first row of THIS order".
    query = foldEngineOptionAliases(objectName, 'findOne', query, ENGINE_QUERY_SLOTS, ENGINE_WIRE_ONLY_SLOTS);
    rejectUnknownEngineOptions(objectName, 'findOne', query, ENGINE_FIND_OPTION_KEYS);
    this.logger.debug('FindOne operation', { objectName });
    const driver = this.getDriver(objectName);
    // `object` after the spread for the same reason as find(); `limit: 1`
    // last — findOne is single-row by contract.
    const ast: QueryAST = { ...query, object: objectName, limit: 1 };
    // Remove context from the AST — it's not a driver concern
    delete (ast as any).context;

    // Plan formula projection (same as find): rewrite ast.fields so the driver
    // returns the raw dependency fields, then evaluate formulas after fetch.
    const _findOneSchema = this._registry.getObject(objectName);
    // Before the guard below, so a `search` that resolves to a real filter
    // counts as the predicate it is (#4419).
    this.expandSearchOnAst(ast, _findOneSchema);
    this.requireFindOnePredicate(objectName, ast);
    const _findOneFormula = planFormulaProjection(_findOneSchema, ast.fields);
    if (_findOneFormula.projected) ast.fields = _findOneFormula.projected;

    // Drop unknown fields — see equivalent block in `find()` for rationale.
    if (_findOneSchema?.fields && Array.isArray(ast.fields) && ast.fields.length > 0) {
      const known = new Set(Object.keys(_findOneSchema.fields));
      // Always allow the primary key + audit columns even if not present
      // in schema.fields (matches `find()` behavior).
      known.add('id');
      known.add('created_at');
      known.add('updated_at');
      const filtered = ast.fields.filter(f => known.has(f.split('.')[0]));
      ast.fields = filtered.length > 0 ? filtered : undefined;
    }

    const opCtx: OperationContext = {
      object: objectName,
      operation: 'findOne',
      ast,
      options: query,
      context: mergeReadContext(query?.context, options?.context),
    };
    this.resolveWhereTokens(opCtx.ast as QueryAST, opCtx.context);

    await this.executeWithMiddleware(opCtx, async () => {
      // [#3195] `findOne` fires the SAME `beforeFind`/`afterFind` hooks as
      // `find` — the read event attaches to record materialization, not to the
      // engine method, so one subscription covers every read shape (there is no
      // separate `beforeFindOne`/`afterFindOne`). Mirrors `find()` above.
      const hookContext: HookContext = {
          object: objectName,
          event: 'beforeFind',
          input: { ast: opCtx.ast, options: opCtx.options },
          session: this.buildSession(opCtx.context),
          provenance: this.buildProvenance(opCtx.context),
          user: this.buildUser(opCtx.context),
          api: this.buildHookApi(opCtx.context),
          transaction: opCtx.context?.transaction,
          ql: this
      };
      await this.triggerHooks('beforeFind', hookContext);
      hookContext.input.options = this.buildDriverOptions(objectName, opCtx.context, hookContext.input.options as any);

      let result = await driver.findOne(objectName, hookContext.input.ast as QueryAST, hookContext.input.options as any);

      // Post-process: evaluate formula virtual fields against the raw row
      if (result != null) applyFormulaPlan(_findOneFormula.plan, [result], opCtx.context);

      // Post-process: expand related records if expand is requested
      if (ast.expand && Object.keys(ast.expand).length > 0 && result != null) {
        const expanded = await this.expandRelatedRecords(objectName, [result], ast.expand, 0, opCtx.context);
        result = expanded[0];
      }

      // Post-process: resolve file-field id references (ADR-0104 D3).
      if (result != null) {
        const resolved = await this.resolveFileReferences(objectName, [result], opCtx.context);
        result = resolved[0];
      }

      hookContext.event = 'afterFind';
      hookContext.result = result;
      await this.triggerHooks('afterFind', hookContext);

      // Mask secret fields — plaintext never leaves through the read path.
      this.maskSecretFields(objectName, hookContext.result);

      return hookContext.result;
    });

    return opCtx.result;
  }

  /**
   * Insert one record or an array of records.
   *
   * At-least-once hook semantics (framework#3152): when this call is driven by
   * `bulkWrite` (seed / import), a transient batch retry or a per-row
   * degradation re-runs the whole insert, so a `beforeInsert` hook may fire
   * MORE THAN ONCE for the same input row (a first attempt that failed in
   * validation, then the degraded retry). Side-effecting `beforeInsert` hooks
   * (notifications, external calls, counters) must therefore be idempotent.
   * `afterInsert` hooks fire only on a successful write, so they are not
   * re-run by a validation failure. Autonumbers are assigned only after
   * validation passes, so a doomed attempt no longer consumes a sequence value
   * (no number-range gaps from a rejected batch).
   */
  // [#3407] `WriteObservabilityOptions.onFieldsDropped` is accepted for
  // signature symmetry with `update()` but never fires here: INSERT is
  // deliberately exempt from the readonly/readonlyWhen strips (a create may
  // legitimately seed read-only columns), and the FLS write gate throws
  // instead of stripping. If insert ever gains a silent strip, wire the
  // listener at that strip site — do not let it go silent.
  async insert(object: string, data: any | any[], options?: DataEngineInsertOptions & WriteObservabilityOptions): Promise<any> {
    object = this.resolveObjectName(object);
    this.logger.debug('Insert operation starting', { object, isBatch: Array.isArray(data) });
    this.assertWriteAllowed(object, 'insert');
    const driver = this.getDriver(object);

    const opCtx: OperationContext = {
      object,
      operation: 'insert',
      data,
      options,
      context: options?.context,
    };

    await this.executeWithMiddleware(opCtx, async () => {
      // Resolve field `defaultValue`s (including the `current_user` token)
      // BEFORE the beforeInsert hook runs, so a hook that DERIVES one field
      // from another can read the defaulted value instead of a stale `null`
      // (#2703). The hook still has final say — it runs after and may override
      // any defaulted field. `applyFieldDefaults` returns a fresh copy and only
      // fills fields left `undefined`, so client-supplied values are untouched.
      const nowSnap = new Date();
      const isBatch = Array.isArray(opCtx.data);
      const defaultedData = isBatch
        ? (opCtx.data as any[]).map((row) =>
            this.applyFieldDefaults(object, row as Record<string, unknown>, opCtx.context, nowSnap),
          )
        : this.applyFieldDefaults(object, opCtx.data as Record<string, unknown>, opCtx.context, nowSnap);

      // Batch inserts trigger beforeInsert/afterInsert PER ROW, each with the
      // exact single-record context shape (`input.data` = one row, `result` =
      // its returned record). A single array-shaped context broke every
      // consumer built for the single shape — the flat-input proxy read
      // `undefined`s, declarative `condition`s evaluated against an array,
      // audit rows and flow-trigger contexts came out mangled (#2922).
      const rowHookContexts: HookContext[] = (isBatch ? (defaultedData as any[]) : [defaultedData]).map(
        (row) => ({
          object,
          event: 'beforeInsert',
          input: { data: row, options: opCtx.options },
          session: this.buildSession(opCtx.context),
          provenance: this.buildProvenance(opCtx.context),
          user: this.buildUser(opCtx.context),
          api: this.buildHookApi(opCtx.context),
          transaction: opCtx.context?.transaction,
          ql: this,
        }),
      );
      for (const rowCtx of rowHookContexts) {
        await this.triggerHooks('beforeInsert', rowCtx);
      }
      // Thread the open transaction (if any) into the driver-facing
      // options so that knex's `.transacting(trx)` is honoured. Without
      // this, calls inside a `engine.transaction(...)` block would deadlock
      // on SQLite's single-connection pool. Also propagates tenantId so
      // the driver can enforce per-tenant isolation.
      // Base the merge on the first row context's options: hooks share the
      // same underlying options object (in-place mutations are visible), and
      // for single inserts this is exactly the pre-#2922 behaviour.
      const driverOptions = this.buildDriverOptions(object, opCtx.context, rowHookContexts[0]?.input.options as any);
      for (const rowCtx of rowHookContexts) {
        rowCtx.input.options = driverOptions;
      }

      try {
        let result: any;
        const schemaForValidation = this._registry.getObject(object);
        // When the driver generates autonumbers natively (persistent SQL
        // sequence), the engine defers to it — see #1603.
        const driverOwnsAutonumber = (driver as any)?.supports?.autonumber === true;
        // Defaults are already resolved above (pre-hook, #2703); a hook may
        // have overridden fields or replaced `input.data` — take its data as-is.
        const rows = rowHookContexts.map((rowCtx) => rowCtx.input.data as Record<string, unknown>);
        // Partial-success mode (framework#3172, entered via insertMany): a row
        // that fails validation is culled and reported per-row instead of
        // aborting the whole batch — so a bulkWrite caller never needs the
        // whole-batch degradation that re-runs beforeInsert hooks on the good
        // rows. rowErrors[i] set = row i is dead; only live rows reach the
        // driver / afterInsert / summaries.
        const partialMode = isBatch && (options as any)?.__partialRowErrors === true;
        const rowErrors: (unknown | undefined)[] = new Array(rows.length);
        for (let i = 0; i < rows.length; i++) {
          try {
            await this.encryptSecretFields(object, rows[i], opCtx.context, driverOptions);
          } catch (e) {
            if (!partialMode) throw e;
            rowErrors[i] = e;
          }
        }
        // Resolved once for the whole batch; dormant unless the object declares
        // a media field, and memoized after the first object that does.
        const mediaValueShapeStrict = await this.mediaValueShapeStrictFor(schemaForValidation);
        const valueShapeStrict = await this.valueShapeStrictFor(schemaForValidation);
        // Locale + translation hooks for the rejection messages (#3957) —
        // resolved once for the batch, identical for every row.
        const msgCtx = this.validationMessageContext(object, opCtx.context);
        // [#4441] The RAW caller payload per row — before `applyFieldDefaults`
        // resolved any `defaultValue` / `current_user` token and before the
        // beforeInsert hooks stamped `owner_id` / `organization_id` /
        // `created_by`. The reference check consults it to decide WHAT THE
        // CALLER ACTUALLY SENT, so neither a platform stamp nor a backfilled
        // default is ever reported as the caller's bad reference.
        const suppliedPerRow: Array<Record<string, unknown>> =
          (isBatch ? (opCtx.data as any[]) : [opCtx.data]).map(
            (row) => (row ?? {}) as Record<string, unknown>,
          );
        for (let i = 0; i < rows.length; i++) {
          if (rowErrors[i] !== undefined) continue;
          try {
            normalizeMultiValueFields(schemaForValidation, rows[i]);
            validateRecord(schemaForValidation, rows[i], 'insert', { mediaValueShapeStrict, valueShapeStrict, messages: msgCtx });
            evaluateValidationRules(schemaForValidation as any, rows[i], 'insert', { logger: this.logger, currentUser: this.buildEvalUser(opCtx.context), skipStateMachine: shouldSkipStateMachine(opCtx.context), messages: msgCtx });
            await this.assertReferencesResolve(
              schemaForValidation, rows[i], suppliedPerRow[i], opCtx.context, msgCtx,
            );
          } catch (e) {
            if (!partialMode) throw e;
            rowErrors[i] = e;
          }
        }
        // Autonumbers are assigned AFTER validation (framework#3152): in the
        // batch path a bulkWrite retry / per-row degradation re-runs the whole
        // engine.insert, and a batch that dies in validation would otherwise
        // have already consumed a sequence value for every good row on the
        // failed attempt, leaving gaps. Required-validation exempts autonumber
        // fields either way (they are engine-assigned), and a driver that owns
        // autonumber assigns nothing here — so no validation rule can depend on
        // the value, making this reorder safe. In partial mode dead rows are
        // skipped, so they never consume a sequence value either.
        for (let i = 0; i < rows.length; i++) {
          if (rowErrors[i] !== undefined) continue;
          try {
            await this.applyAutonumbers(object, rows[i], opCtx.context, driverOwnsAutonumber);
          } catch (e) {
            if (!partialMode) throw e;
            rowErrors[i] = e;
          }
        }
        // Live rows = the ones that survived per-row preparation. In
        // non-partial mode rowErrors is all-empty, so this is exactly `rows`.
        const liveIndexes: number[] = [];
        for (let i = 0; i < rows.length; i++) if (rowErrors[i] === undefined) liveIndexes.push(i);
        const liveRows = liveIndexes.map((i) => rows[i]);
        if (isBatch) {
          if (liveRows.length === 0) {
               result = [];
          } else if (driver.bulkCreate) {
               result = await driver.bulkCreate(object, liveRows, driverOptions);
          } else {
               // Fallback loop
               result = await Promise.all(liveRows.map((item) => driver.create(object, item, driverOptions)));
          }
        } else {
          result = await driver.create(object, liveRows[0], driverOptions);
        }

        // Driver-result contract guard (framework#3151): a batch write must
        // return one record per input row. A short / non-array return would
        // otherwise be padded with `undefined` below and fed to afterInsert
        // hooks (`ctx.result === undefined`) and back to the caller as phantom
        // records — corrupting seed externalId→id maps and import undo logs.
        // Refuse it: throw so the caller sees a real failure rather than
        // silent data loss. (Every driver in this repo already returns
        // one-per-row in order; this defends against third-party drivers.)
        if (isBatch && (!Array.isArray(result) || result.length !== liveRows.length)) {
          throw Object.assign(
            new Error(
              `bulkCreate for '${object}' returned ${
                Array.isArray(result) ? `${result.length} record(s)` : String(typeof result)
              } for ${liveRows.length} input row(s) — refusing to fabricate afterInsert contexts`,
            ),
            { code: 'ERR_BULK_RESULT_MISMATCH' },
          );
        }

        // Coerce `boolean` fields (SQLite/libsql return 0/1) to real booleans on
        // the after-hook view so flow trigger conditions (`record.is_escalated
        // != true`) and `{record.<bool>}` interpolation see JS booleans, not
        // ints. A shallow copy — the value returned to the caller is untouched.
        // Only live rows have results — dead (partial-mode) rows never reach
        // afterInsert.
        const resultRows: any[] = isBatch ? (Array.isArray(result) ? result : [result]) : [result];
        for (let k = 0; k < liveIndexes.length; k++) {
          const rowCtx = rowHookContexts[liveIndexes[k]];
          rowCtx.event = 'afterInsert';
          rowCtx.result = coerceBooleanFields(schemaForValidation as any, resultRows[k] as any);
          await this.triggerHooks('afterInsert', rowCtx);
        }

        // Roll-up: recompute parent summary fields that aggregate this object.
        const summaryFailures = await this.recomputeSummaries(object, result, null, opCtx.context);

        // Publish one data.record.created DataEvent per written record (#4626).
        // A batch insert is N record events, not one event about N records —
        // `DataEvent.recordId` is per record.
        if (this.realtimeService) {
          const createdRows: any[] = Array.isArray(result) ? result : [result];
          for (const record of createdRows) {
            await this.publishDataEvent('created', object, {
              recordId: record?.id,
              after: record,
              context: opCtx.context,
            });
          }
        }

        // Return the (possibly hook-mutated) after-view: the array of per-row
        // results for batch, the single record otherwise. In partial mode the
        // batch return is instead one outcome PER INPUT ROW ({ok,record} /
        // {ok:false,error}), in input order (framework#3172).
        const written = isBatch
          ? (partialMode
              ? rows.map((_r, i) => (rowErrors[i] === undefined
                  ? { ok: true as const, record: rowHookContexts[i].result }
                  : { ok: false as const, error: rowErrors[i] }))
              : rowHookContexts.map((rowCtx) => rowCtx.result))
          : rowHookContexts[0].result;
        // Records ARE written; a summary that could not be recomputed after
        // retries must not be silent (framework#3147). Thrown after realtime
        // publish, carrying the written records so a bulk caller (seed/import)
        // can treat it as a warning rather than re-writing.
        if (summaryFailures.length > 0) throw new SummaryRecomputeError(summaryFailures, written);
        return written;
      } catch (e) {
        this.logger.error('Insert operation failed', e as Error, { object });
        throw e;
      }
    });

    return opCtx.result;
  }

  /**
   * Batch insert with PARTIAL SUCCESS (framework#3172): unlike
   * `insert(object, rows[])` — which aborts the whole batch when any row
   * fails validation — this culls the bad rows after beforeInsert, writes the
   * survivors in one driver batch, and returns one outcome per input row in
   * input order. beforeInsert hooks therefore run exactly ONCE per row even
   * when the batch contains bad rows (no whole-batch degradation re-run), and
   * dead rows never consume an autonumber sequence value. afterInsert hooks,
   * summary recompute, and realtime events fire only for the written rows.
   *
   * A summary-recompute failure after retries still throws
   * {@link SummaryRecomputeError} (framework#3147) with `written` set to the
   * outcome array — the records ARE written.
   */
  async insertMany(object: string, rows: any[], options?: DataEngineInsertOptions): Promise<InsertManyRowOutcome[]> {
    if (!Array.isArray(rows)) throw new Error('insertMany expects an array of rows');
    return this.insert(object, rows, { ...(options ?? {}), __partialRowErrors: true } as any);
  }

  async update(object: string, data: any, options?: EngineUpdateOptions & WriteObservabilityOptions): Promise<any> {
     object = this.resolveObjectName(object);
     this.logger.debug('Update operation starting', { object });
     this.assertWriteAllowed(object, 'update');
     const driver = this.getDriver(object);

     // Fold the `filter` alias into `where` FIRST (#4346): everything below —
     // token resolution, the by-id fast path, the #2982 AST seeding — reads
     // `options.where` only, so an unfolded `{ filter }` left the AST with no
     // predicate at all and a `multi: true` update rewrote EVERY row.
     options = foldEngineOptionAliases(object, 'update', options, ENGINE_WHERE_SLOTS);
     rejectUnknownEngineOptions(object, 'update', options, ENGINE_UPDATE_OPTION_KEYS);

     // Expand `{filter-placeholder}` values BEFORE the id is extracted (#3810).
     // The read path resolves them; without the same call here the SAME filter
     // selected different rows depending on the verb — `find({owner:
     // '{current_user_id}'})` matched the signed-in user's rows while
     // `update`/`delete` compared the literal token text and matched none. That
     // is the #3106 shape one layer down: the evaluator existed, but only some
     // call sites reached it.
     //
     // Ordering matters: a scalar `where.id` becomes the by-id fast path below,
     // so an unresolved `{current_user_id}` would be bound as the primary key
     // itself. Resolve first, then extract.
     options = this.withResolvedWhere(options);

     // 1. Extract ID from data or where if it's a single update by ID.
     //    Only a SCALAR `where.id` means "update one row by primary key". An
     //    operator object ({ $in: [...] }, { $ne: ... }, …) is a multi-row
     //    predicate — treating it as an id would bind the object literally
     //    (e.g. `WHERE id = {"$in":[...]}`, which SQLite rejects). Leave `id`
     //    undefined in that case so the call routes to updateMany (requires
     //    options.multi=true), where applyFilters compiles the operator.
     let id = data.id;
     if (!id && options?.where && typeof options.where === 'object' && 'id' in options.where) {
         const whereId = (options.where as Record<string, unknown>).id;
         const t = typeof whereId;
         if (whereId !== null && (t === 'string' || t === 'number' || t === 'bigint')) {
             id = whereId;
         }
     }

     const opCtx: OperationContext = {
       object,
       operation: 'update',
       data,
       options,
       context: options?.context,
     };

     // [#2982] A no-single-id update routes to `driver.updateMany` below with
     // an AST that used to be REBUILT from `options.where` AFTER the
     // middleware chain — so row-scoping filters a middleware AND-composed
     // onto `opCtx.ast` (RLS write policies, the sharing plugin's
     // editable-rows filter) never reached the driver, and a bulk write hit
     // every matching row regardless of ownership. Seed the AST with the
     // caller's predicate BEFORE the chain runs — the same seam the read path
     // uses — and let the multi branch consume the composed result. Keyed on
     // the SAME falsy-`id` test the multi branch dispatches on (below), so the
     // seed and the branch never disagree. `where` is included only when
     // supplied, mirroring the read path's AST shape.
     if (!id) {
       opCtx.ast = { object, ...(options?.where !== undefined ? { where: options.where } : {}) } as QueryAST;
     }

     // [#2948] Snapshot the keys the CALLER supplied, BEFORE any middleware /
     // beforeUpdate hook stamps server-managed columns (owner/tenant stamp,
     // `updated_by`/`updated_at`). The static-`readonly` strip below drops only
     // caller-supplied read-only writes, so hook/middleware stamps survive.
     const suppliedKeys: ReadonlySet<string> = new Set(
       Object.keys((opCtx.data ?? {}) as Record<string, unknown>),
     );

     // [#3407] Structured strip observability. The readonly/readonlyWhen strips
     // below are LEGAL semantics (the write still succeeds without the locked
     // fields), but until now the only trace was a server-side logger warn — a
     // caller that reports success per requested field (a flow's `update_record`
     // step) saw a clean success while the DB value never changed. When the
     // caller registers `onFieldsDropped`, report each strip pass's dropped
     // keys back with its reason. Diffing before/after key sets is exact here:
     // every strip helper returns the SAME reference when nothing was dropped,
     // else a shallow copy with keys removed. A listener fault must never break
     // the write.
     const onFieldsDropped = options?.onFieldsDropped;
     const reportDroppedFields = (
       before: Record<string, unknown> | null | undefined,
       after: Record<string, unknown> | null | undefined,
       reason: DroppedFieldsEvent['reason'],
     ): void => {
       if (!onFieldsDropped || before === after || !before) return;
       const afterObj = (after ?? {}) as Record<string, unknown>;
       const fields = Object.keys(before).filter((k) => !(k in afterObj));
       if (fields.length === 0) return;
       try {
         onFieldsDropped({ object, fields, reason });
       } catch (err) {
         this.logger.warn('onFieldsDropped listener threw — ignored', { object, error: err });
       }
     };

     await this.executeWithMiddleware(opCtx, async () => {
       const hookContext: HookContext = {
          object,
          event: 'beforeUpdate',
          input: { id, data: opCtx.data, options: opCtx.options },
          session: this.buildSession(opCtx.context),
          provenance: this.buildProvenance(opCtx.context),
          user: this.buildUser(opCtx.context),
          api: this.buildHookApi(opCtx.context),
          transaction: opCtx.context?.transaction,
          ql: this
       };
       await this.triggerHooks('beforeUpdate', hookContext);
       hookContext.input.options = this.buildDriverOptions(object, opCtx.context, hookContext.input.options as any);

       try {
           let result;
           // [#4639] Which event contract this write reports under. A predicate
           // write publishes the aggregate `data.records.updated`; anything else
           // publishes the per-record `data.record.updated`. Recorded at the
           // branch that made the choice rather than re-derived at the publish
           // site from an absent id — the two are the same today, and a future
           // driver returning a row from `updateMany` would silently reroute a
           // bulk write onto the per-record contract if we inferred it.
           let isPredicateWrite = false;
           // Pre-update snapshot. Exposed to after-hooks via `hookContext.previous`
           // (the HookContext contract documents `previous` for update/delete) and
           // reused for object-level validation rules. Fetched once, only for
           // single-id updates, when either a rule needs it (ADR-0020:
           // state_machine / cross_field / script — a PATCH carries only changed
           // fields) OR an afterUpdate hook is registered. The latter is what makes
           // record-change flow triggers work: their start-condition gate reads
           // `previous.*` (e.g. `status == "done" && previous.status != "done"`),
           // which silently fails when `previous` is absent.
           //
           // [#4784] It is ALSO what supplies the `previous` binding to a
           // declarative hook `condition` (`hook-wrappers.ts`), which is how a
           // TRANSITION is expressed there: `previous.done != true &&
           // record.done == true`. Note this needs NO second demand-driven
           // fetch — the existing gate already fetches whenever an afterUpdate
           // hook exists, and afterUpdate is the event whose context carries
           // `previous`. Deliberately: adding a "does the condition reference
           // `previous`?" analysis on top would be dead code today. If this
           // gate is ever NARROWED (e.g. scoped per object), hook conditions
           // reading `previous` must be counted into the new demand test —
           // pinned by `hook-condition-previous-scope.test.ts`.
           let priorRecord: Record<string, unknown> | null = null;
           const updateSchema = this._registry.getObject(object);
           const mediaValueShapeStrict = await this.mediaValueShapeStrictFor(updateSchema);
           const valueShapeStrict = await this.valueShapeStrictFor(updateSchema);
           const updateMsgCtx = this.validationMessageContext(object, opCtx.context);
           if (hookContext.input.id) {
               await this.encryptSecretFields(object, hookContext.input.data as Record<string, unknown>, opCtx.context, hookContext.input.options);
               normalizeMultiValueFields(updateSchema, hookContext.input.data as Record<string, unknown>);
               validateRecord(updateSchema, hookContext.input.data as Record<string, unknown>, 'update', { mediaValueShapeStrict, valueShapeStrict, messages: updateMsgCtx });
               if (needsPriorRecord(updateSchema as any) || (this.hooks.get('afterUpdate')?.length ?? 0) > 0) {
                   const priorAst: QueryAST = { object, where: { id: hookContext.input.id }, limit: 1 };
                   priorRecord = await driver.findOne(object, priorAst, hookContext.input.options as any);
               }
               // B2: drop writes to fields locked by a TRUE `readonlyWhen` — the
               // field is read-only for this record's state, so the incoming
               // change is ignored (the persisted value is kept).
               const preRoWhen = hookContext.input.data as Record<string, unknown>;
               hookContext.input.data = stripReadonlyWhenFields(updateSchema as any, preRoWhen, priorRecord, this.logger) as any;
               reportDroppedFields(preRoWhen, hookContext.input.data as Record<string, unknown>, 'readonly_when');
               // [#2948] Enforce STATIC `readonly` on the write path for
               // non-system callers (system writes legitimately set read-only
               // columns and are exempt). Runs AFTER hooks/middleware stamped
               // their columns; `suppliedKeys` ensures only caller-forged
               // read-only writes are dropped, never the server stamps.
               if (!opCtx.context?.isSystem) {
                   const preRo = hookContext.input.data as Record<string, unknown>;
                   hookContext.input.data = stripReadonlyFields(updateSchema as any, preRo, suppliedKeys, this.logger, { preserveAudit: opCtx.context?.preserveAudit === true }) as any;
                   reportDroppedFields(preRo, hookContext.input.data as Record<string, unknown>, 'readonly');
               }
               evaluateValidationRules(updateSchema as any, hookContext.input.data as Record<string, unknown>, 'update', { previous: priorRecord, logger: this.logger, currentUser: this.buildEvalUser(opCtx.context), skipStateMachine: shouldSkipStateMachine(opCtx.context), messages: updateMsgCtx });
               // [#4441] A repoint is as capable of dangling as an initial link.
               await this.assertReferencesResolve(
                 updateSchema, hookContext.input.data as Record<string, unknown>,
                 opCtx.data as Record<string, unknown>, opCtx.context, updateMsgCtx,
               );
               result = await driver.update(object, hookContext.input.id as string, hookContext.input.data as Record<string, unknown>, hookContext.input.options as any);
           } else if (options?.multi && driver.updateMany) {
               await this.encryptSecretFields(object, hookContext.input.data as Record<string, unknown>, opCtx.context, hookContext.input.options);
               normalizeMultiValueFields(updateSchema, hookContext.input.data as Record<string, unknown>);
               validateRecord(updateSchema, hookContext.input.data as Record<string, unknown>, 'update', { mediaValueShapeStrict, valueShapeStrict, messages: updateMsgCtx });
               // [#2982] Consume the middleware-composed AST seeded above, so
               // the injected row-scoping (RLS write filter, sharing's
               // editable-rows filter) actually binds the driver operation. Fail
               // CLOSED if it is somehow absent — rebuilding `{ object, where }`
               // here would silently drop every composed filter and reopen the
               // unscoped-bulk-write hole this fix closes (AGENTS.md PD #12: no
               // lenient fallback that tolerates the broken invariant).
               const ast = opCtx.ast;
               if (!ast) {
                   throw new Error(
                     `[Security] Refusing bulk update on '${object}': row-scoping AST was not seeded ` +
                       `(a hook cleared the target id after the security filter was composed).`,
                   );
               }
               // [#3106] Validation rules, `requiredWhen` and per-option
               // `visibleWhen` are PER ROW on a bulk update, exactly like the
               // `readonlyWhen` strip below: one payload, N prior states. Read
               // the row-scoped match set ONCE with the SAME AST the write binds
               // (shared with the [#3042] strip), and only when the schema
               // actually needs prior state — `needsPriorRecord` subsumes
               // `hasReadonlyWhenInPayload` (readonlyWhen fields count toward
               // it), so a rule-free schema still pays nothing here.
               const rulesNeedRows = needsPriorRecord(updateSchema as any);
               const payloadHasReadonlyWhen = hasReadonlyWhenInPayload(updateSchema as any, hookContext.input.data as Record<string, unknown>);
               let priorRows: Record<string, unknown>[] | null = null;
               if (rulesNeedRows || payloadHasReadonlyWhen) {
                   priorRows = await driver.find(object, ast, hookContext.input.options as any) as Record<string, unknown>[];
               }
               // [#3042] Enforce conditional `readonlyWhen` on the bulk path too.
               // Unlike static `readonly` (below), a `readonlyWhen` lock is PER
               // ROW — drop any field locked in ≥1 matched row: a bulk write
               // can't keep it for some rows and drop it for others, so a field
               // locked in any target row is fail-safe-dropped for all (narrow
               // `where` to reach the unlocked rows). Symmetric with the
               // single-id `stripReadonlyWhenFields`; INSERT stays exempt.
               if (payloadHasReadonlyWhen) {
                   const preRoWhenMulti = hookContext.input.data as Record<string, unknown>;
                   hookContext.input.data = stripReadonlyWhenFieldsMulti(updateSchema as any, preRoWhenMulti, priorRows, this.logger) as any;
                   reportDroppedFields(preRoWhenMulti, hookContext.input.data as Record<string, unknown>, 'readonly_when');
               }
               // [#2948] Same static-`readonly` write guard on the bulk path —
               // a forged read-only column in a multi-row update is dropped for
               // non-system callers (a foreign `organization_id` is additionally
               // rejected upstream by the tenant write wall, #2946).
               if (!opCtx.context?.isSystem) {
                   const preRoMulti = hookContext.input.data as Record<string, unknown>;
                   hookContext.input.data = stripReadonlyFields(updateSchema as any, preRoMulti, suppliedKeys, this.logger, { preserveAudit: opCtx.context?.preserveAudit === true }) as any;
                   reportDroppedFields(preRoMulti, hookContext.input.data as Record<string, unknown>, 'readonly');
               }
               // [#3106] Same enforcement the single-id branch runs at its
               // `evaluateValidationRules` call, applied per matched row: any
               // error-severity violation rejects the WHOLE batch before
               // `updateMany` writes anything (all-or-nothing, like the strip's
               // locked-in-any-row rule). Runs on the stripped payload, in the
               // single-id branch's order. Warning-severity violations may log
               // once per matched row — accepted. With no prior-dependent rules
               // the payload-only evaluation covers format / json_schema /
               // non-prior conditional at zero fetch cost.
               const bulkEvalUser = this.buildEvalUser(opCtx.context);
               if (rulesNeedRows) {
                   for (const row of priorRows ?? []) {
                       try {
                           evaluateValidationRules(updateSchema as any, hookContext.input.data as Record<string, unknown>, 'update', { previous: row, logger: this.logger, currentUser: bulkEvalUser, skipStateMachine: shouldSkipStateMachine(opCtx.context), messages: updateMsgCtx });
                       } catch (err) {
                           if (err instanceof ValidationError && row?.id != null) {
                               throw new ValidationError(err.fields.map((f) => ({ ...f, message: `${f.message} (record ${String(row.id)})` })));
                           }
                           throw err;
                       }
                   }
               } else {
                   evaluateValidationRules(updateSchema as any, hookContext.input.data as Record<string, unknown>, 'update', { previous: null, logger: this.logger, currentUser: bulkEvalUser, skipStateMachine: shouldSkipStateMachine(opCtx.context), messages: updateMsgCtx });
               }
               // [#4441] The bulk call site too — a guard wired into single-id
               // writes only is still a hole one call site over (AGENTS.md
               // PD #10's own worked example, #3106).
               await this.assertReferencesResolve(
                 updateSchema, hookContext.input.data as Record<string, unknown>,
                 opCtx.data as Record<string, unknown>, opCtx.context, updateMsgCtx,
               );
               result = await driver.updateMany(object, ast, hookContext.input.data as Record<string, unknown>, hookContext.input.options as any);
               isPredicateWrite = true;
           } else {
               throw new Error('Update requires an ID or options.multi=true');
           }

           hookContext.event = 'afterUpdate';
           // Coerce boolean fields (SQLite 0/1 → JS bool) on the after-hook view
           // of both the new row and the prior row, so flow conditions comparing
           // `record.is_escalated`/`previous.status` against booleans behave.
           // Shallow copies — the value returned to the caller is untouched.
           hookContext.result = Array.isArray(result)
             ? result.map((r) => coerceBooleanFields(updateSchema as any, r as any))
             : coerceBooleanFields(updateSchema as any, result as any);
           if (priorRecord) hookContext.previous = coerceBooleanFields(updateSchema as any, priorRecord as any);
           await this.triggerHooks('afterUpdate', hookContext);

           // Roll-up: recompute parent summaries; pass priorRecord too so a child
           // that moved to a different parent updates BOTH old and new parent.
           const summaryFailures = await this.recomputeSummaries(object, result, priorRecord, opCtx.context);

           // Publish the update event under whichever contract this write can
           // honour: per-record `data.record.updated` (#4626), or the aggregate
           // `data.records.updated` for a predicate write, whose driver call
           // returns an affected count and names no row (#4639).
           if (this.realtimeService) {
             if (isPredicateWrite) {
               await this.publishBulkDataEvent('updated', object, {
                 matched: result,
                 context: opCtx.context,
               });
             } else {
               const resultId = (typeof result === 'object' && result && 'id' in result) ? (result as any).id : undefined;
               await this.publishDataEvent('updated', object, {
                 recordId: hookContext.input.id ?? resultId,
                 changes: hookContext.input.data,
                 after: result,
                 context: opCtx.context,
               });
             }
           }

           // The record IS updated; a summary that could not recompute after
           // retries must surface, not stay silent (framework#3147).
           if (summaryFailures.length > 0) throw new SummaryRecomputeError(summaryFailures, hookContext.result);
           return hookContext.result;
       } catch (e) {
          this.logger.error('Update operation failed', e as Error, { object });
          throw e;
       }
     });

     return opCtx.result;
  }

  /**
   * Apply referential delete behavior for relations pointing AT this record,
   * before it is removed. For every registered object with a `master_detail`
   * or `lookup` field referencing `object`, honor the field's `deleteBehavior`:
   *   - `cascade`  → delete the dependent rows (recursively, so grandchildren
   *                  are handled by each child's own delete),
   *   - `set_null` → clear the foreign key,
   *   - `restrict` → refuse the delete when dependents exist.
   * `master_detail` defaults to `cascade` (the parent owns the child
   * lifecycle); `lookup` defaults to `set_null` — except a `set_null` default
   * on a REQUIRED lookup escalates to `restrict` (you can't null a NOT NULL
   * FK; restricting with a clear dependent-count message beats a misleading
   * "<field> is required" 400 from the child). Only runs for single-id
   * deletes — multi/predicate deletes skip cascade (logged).
   */
  private async cascadeDeleteRelations(
    object: string,
    id: string | number,
    context?: ExecutionContextInput,
    depth = 0,
  ): Promise<void> {
    if (id == null || depth >= ObjectQL.MAX_CASCADE_DEPTH) return;
    let objects: ServiceObject[];
    try {
      objects = this._registry.getAllObjects();
    } catch {
      return;
    }
    for (const child of objects) {
      const childName = (child as any)?.name as string | undefined;
      const fields = (child as any)?.fields as Record<string, any> | undefined;
      if (!childName || !fields) continue;
      for (const [fieldName, fdef] of Object.entries(fields)) {
        if (!fdef || (fdef.type !== 'master_detail' && fdef.type !== 'lookup')) continue;
        const ref = fdef.reference;
        if (!ref) continue;
        // Match the target object by raw or resolved name.
        let resolvedRef: string | undefined;
        try { resolvedRef = this.resolveObjectName(ref); } catch { resolvedRef = undefined; }
        if (ref !== object && resolvedRef !== object) continue;

        // A master-detail parent owns its children: cascade by default (the
        // child FK is typically required, so set_null would be invalid). Only
        // an explicit `restrict` deviates. A plain lookup honors its
        // configured deleteBehavior (default set_null).
        let behavior: string =
          fdef.type === 'master_detail'
            ? (fdef.deleteBehavior === 'restrict' ? 'restrict' : 'cascade')
            : (fdef.deleteBehavior || 'set_null');

        // A REQUIRED foreign key cannot be nulled — set_null would issue an
        // UPDATE clearing the FK, which the child's required-field validator
        // rejects with a misleading "<field> is required" 400 (the field isn't
        // even on the object being deleted). That's a contradiction, not the
        // author's intent: a required FK means the child can't exist detached,
        // so deleting the parent must be RESTRICTed (SQL's default for a
        // NOT NULL FK). Authors who want the children gone set
        // deleteBehavior:'cascade' explicitly. This only escalates the
        // *defaulted* set_null; an explicit cascade/restrict is untouched.
        if (behavior === 'set_null' && fdef.required === true) {
          behavior = 'restrict';
        }

        let dependents: any[];
        try {
          dependents = await this.find(childName, { where: { [fieldName]: id }, context } as any);
        } catch {
          continue;
        }
        if (!dependents || dependents.length === 0) continue;

        if (behavior === 'restrict') {
          const reason = fdef.deleteBehavior !== 'restrict' && fdef.required === true
            ? ` (${fieldName} is required, so it cannot be cleared)`
            : '';
          const err: any = new Error(
            `Cannot delete ${object} (${id}): ${dependents.length} dependent ${childName} record(s) reference it via ${fieldName}${reason}. ` +
            `Delete or reassign them first, or set deleteBehavior:'cascade' on ${childName}.${fieldName}.`,
          );
          err.code = 'DELETE_RESTRICTED';
          err.status = 409;
          err.object = object;
          err.dependentObject = childName;
          err.dependentCount = dependents.length;
          throw err;
        }

        for (const dep of dependents) {
          const depId = dep?.id;
          if (depId == null) continue;
          if (behavior === 'cascade') {
            // Recurse via the public delete so the child's own cascade,
            // hooks and events fire.
            await this.delete(childName, { where: { id: depId }, context } as any);
          } else {
            // [#3023] Clear the FK as an engine-internal referential-integrity
            // write, tagged so plugin-security's ownership-anchor guard (#3004)
            // treats an `owner_id = null` cascade as integrity maintenance, not
            // a user-initiated disown — otherwise deleting the referenced record
            // trips the transfer guard and aborts the cascade mid-way. Marker
            // rides a server-DERIVED context (set here, never from client input
            // — same trust model as `__expandRead`), so it cannot be forged from
            // a request to bypass the guard on an ordinary write.
            const referentialCtx = { ...(context ?? {}), __referentialFieldClear: true } as ExecutionContextInput;
            await this.update(childName, { id: depId, [fieldName]: null }, { context: referentialCtx } as any);
          }
        }
      }
    }
  }

  async delete(object: string, options?: EngineDeleteOptions): Promise<any> {
    object = this.resolveObjectName(object);
    this.logger.debug('Delete operation starting', { object });
    this.assertWriteAllowed(object, 'delete');
    const driver = this.getDriver(object);

    // Fold the `filter` alias into `where` first — same reasoning as update()
    // above (#4346): unfolded, a `multi: true` delete with `{ filter }` had no
    // predicate on its AST and emptied the table.
    options = foldEngineOptionAliases(object, 'delete', options, ENGINE_WHERE_SLOTS);
    rejectUnknownEngineOptions(object, 'delete', options, ENGINE_DELETE_OPTION_KEYS);

    // Expand `{filter-placeholder}` values before the id is extracted — same
    // reasoning as update() above (#3810).
    options = this.withResolvedWhere(options);

    // Extract ID logic mirroring update(): only a SCALAR `where.id` means
    // "delete one row by primary key". An operator object ({ $in: [...] }, …)
    // is a multi-row predicate — treating it as an id would bind the object
    // literally (driver.delete(object, {$in:[…]})) and both skip the #2982 AST
    // seeding below AND bypass the by-id RLS pre-image check. Leave `id`
    // undefined so the call routes to deleteMany with the scoped AST.
    let id: any = undefined;
    if (options?.where && typeof options.where === 'object' && 'id' in options.where) {
        const whereId = (options.where as Record<string, unknown>).id;
        const t = typeof whereId;
        if (whereId !== null && (t === 'string' || t === 'number' || t === 'bigint')) {
            id = whereId;
        }
    }

    const opCtx: OperationContext = {
      object,
      operation: 'delete',
      options,
      context: options?.context,
    };

    // [#2982] Same seam as update: a no-single-id delete routes to
    // `driver.deleteMany` with an AST that used to be rebuilt from
    // `options.where` after the middleware chain, discarding any row-scoping a
    // middleware composed onto `opCtx.ast`. Seed the caller's predicate before
    // the chain, keyed on the SAME falsy-`id` test the multi branch dispatches
    // on; the multi branch consumes the composed result.
    if (!id) {
      opCtx.ast = { object, ...(options?.where !== undefined ? { where: options.where } : {}) } as QueryAST;
    }

    await this.executeWithMiddleware(opCtx, async () => {
      const hookContext: HookContext = {
          object,
          event: 'beforeDelete',
          input: { id, options: opCtx.options },
          session: this.buildSession(opCtx.context),
          provenance: this.buildProvenance(opCtx.context),
          user: this.buildUser(opCtx.context),
          api: this.buildHookApi(opCtx.context),
          transaction: opCtx.context?.transaction,
          ql: this
      };
      await this.triggerHooks('beforeDelete', hookContext);
      hookContext.input.options = this.buildDriverOptions(object, opCtx.context, hookContext.input.options as any);

      try {
          let result;
          // [#4639] See update()'s twin: recorded at the branch that chose the
          // driver call, not inferred later from a missing id.
          let isPredicateWrite = false;
          // Capture the row's FK values BEFORE deletion so roll-up summaries can
          // recompute the (now-orphaned) parent. Only when a summary aggregates
          // this object — avoids an extra read on every delete.
          let summaryPrev: any = null;
          if (hookContext.input.id && this.getSummaryDescriptors(object).length > 0) {
            try {
              summaryPrev = await this.findOne(object, { where: { id: hookContext.input.id }, context: opCtx.context } as any);
            } catch { /* best-effort */ }
          }
          if (hookContext.input.id) {
              // Honor referential delete behavior (cascade/set_null/restrict)
              // for relations pointing at this record before removing it.
              await this.cascadeDeleteRelations(object, hookContext.input.id as string | number, opCtx.context);
              result = await driver.delete(object, hookContext.input.id as string, hookContext.input.options as any);
          } else if (options?.multi && driver.deleteMany) {
               // [#2982] Consume the middleware-composed AST seeded above so the
               // injected row-scoping binds the bulk delete. Fail CLOSED if it
               // is absent rather than rebuilding an unscoped `{ object, where }`
               // (AGENTS.md PD #12).
               const ast = opCtx.ast;
               if (!ast) {
                   throw new Error(
                     `[Security] Refusing bulk delete on '${object}': row-scoping AST was not seeded ` +
                       `(a hook cleared the target id after the security filter was composed).`,
                   );
               }
               result = await driver.deleteMany(object, ast, hookContext.input.options as any);
               isPredicateWrite = true;
          } else {
               throw new Error('Delete requires an ID or options.multi=true');
          }

          hookContext.event = 'afterDelete';
          hookContext.result = result;
          await this.triggerHooks('afterDelete', hookContext);

          // Roll-up: recompute the parent summary now that the child is gone.
          const summaryFailures = summaryPrev
            ? await this.recomputeSummaries(object, null, summaryPrev, opCtx.context)
            : [];

          // Same split as update(): per-record `data.record.deleted` (#4626),
          // or the aggregate `data.records.deleted` when the delete was by
          // predicate (#4639).
          if (this.realtimeService) {
            if (isPredicateWrite) {
              await this.publishBulkDataEvent('deleted', object, {
                matched: result,
                context: opCtx.context,
              });
            } else {
              const resultId = (typeof result === 'object' && result && 'id' in result) ? (result as any).id : undefined;
              await this.publishDataEvent('deleted', object, {
                recordId: hookContext.input.id ?? resultId,
                context: opCtx.context,
              });
            }
          }

          // The record IS deleted; a summary that could not recompute after
          // retries must surface, not stay silent (framework#3147).
          if (summaryFailures.length > 0) throw new SummaryRecomputeError(summaryFailures, hookContext.result);
          return hookContext.result;
      } catch (e) {
          this.logger.error('Delete operation failed', e as Error, { object });
          throw e;
      }
    });

    return opCtx.result;
  }

  async count(object: string, query?: EngineCountOptions, options?: EngineReadOptions): Promise<number> {
     object = this.resolveObjectName(object);
     // Fold the `filter` alias into `where` (#4346) — the AST below reads
     // `query.where` only, so an unfolded `{ filter }` counted the whole table.
     query = foldEngineOptionAliases(object, 'count', query, ENGINE_WHERE_SLOTS);
     rejectUnknownEngineOptions(object, 'count', query, ENGINE_COUNT_OPTION_KEYS);
     const driver = this.getDriver(object);

     // The AST must ride on the opCtx so the security/sharing middlewares can
     // inject their read filters (RLS, OWD/sharing scope) into `ast.where` —
     // exactly like find(). Building it locally inside the executor (#2737)
     // discarded every injected filter: `total` counted the RAW table while
     // `records` were scoped, leaking invisible-row counts and breaking
     // pagination.
     const opCtx: OperationContext = {
       object,
       operation: 'count',
       ast: { object, where: query?.where },
       options: query,
       context: mergeReadContext(query?.context, options?.context),
     };
     this.resolveWhereTokens(opCtx.ast as QueryAST, opCtx.context);
     // The caller's own `where`, placeholders expanded — captured BEFORE the
     // middleware chain scopes `opCtx.ast.where`, so the find() fallback below
     // still passes the unscoped filter (find() applies the read filters itself).
     const callerWhere = (opCtx.ast as QueryAST).where;

     await this.executeWithMiddleware(opCtx, async () => {
       const countOpts = this.buildDriverOptions(object, opCtx.context);
       if (driver.count) {
           return driver.count(object, opCtx.ast as QueryAST, countOpts);
       }
       // Fallback to find().length — find() applies the read filters itself,
       // so pass the caller's original where, not the already-scoped ast.
       const res = await this.find(object, { where: callerWhere, fields: ['id'], context: opCtx.context });
       return res.length;
     });

     return opCtx.result as number;
  }

  /**
   * Fail-closed guard (ADR-0100 / #3171): refuse to aggregate over a credential
   * field. `secret`/`password` values are masked on the generic read path so
   * plaintext never leaves the engine, but `aggregate()` has no equivalent mask
   * — a GROUP BY / MIN / MAX / array_agg over such a column would surface the
   * stored `secret:<id>` ref or the password value, and post-hoc masking would
   * corrupt group keys. So we reject instead. The check is unconditional
   * (ignores `managedBy`): aggregating a credential is never legitimate, even on
   * a better-auth object, where it would be an inference oracle over hashes.
   *
   * Only the two output-bearing positions on `EngineAggregateOptions` carry
   * field names: `aggregations[].field` (skip COUNT(*) — undefined or '*') and
   * `groupBy[]` (a string, or a `{ field }` bucket object).
   */
  private rejectCredentialAggregation(object: string, query: EngineAggregateOptions): void {
    const schema = this._registry.getObject(object);
    const credentialFields = collectCredentialFields(schema);
    if (credentialFields.length === 0) return;

    const referenced = new Set<string>();
    for (const agg of query?.aggregations ?? []) {
      const field = (agg as { field?: string })?.field;
      if (field && field !== '*') referenced.add(field);
    }
    for (const g of (query?.groupBy as unknown[]) ?? []) {
      const field = typeof g === 'string' ? g : (g as { field?: string })?.field;
      if (field) referenced.add(field);
    }

    const hit = credentialFields.filter((f) => referenced.has(f));
    if (hit.length > 0) {
      throw new Error(
        `Cannot aggregate credential field(s) ${hit.map((f) => `"${object}.${f}"`).join(', ')}: `
          + 'secret/password fields are masked on read so plaintext never leaves the engine, and '
          + 'aggregating them (group-by, min/max, array_agg, …) would surface the stored value. '
          + 'Refusing (fail-closed) — see ADR-0100 / #3171.',
      );
    }
  }

  async aggregate(object: string, query: EngineAggregateOptions, options?: EngineReadOptions): Promise<any[]> {
      object = this.resolveObjectName(object);
      // Fold the `filter` alias into `where` (#4346) — the AST below reads
      // `query.where` only, so an unfolded `{ filter }` aggregated every row.
      query = foldEngineOptionAliases(object, 'aggregate', query, ENGINE_WHERE_SLOTS);
      rejectUnknownEngineOptions(object, 'aggregate', query, ENGINE_AGGREGATE_OPTION_KEYS);
      this.rejectCredentialAggregation(object, query);
      const driver = this.getDriver(object);
      this.logger.debug(`Aggregate on ${object} using ${driver.name}`, query);

      const opCtx: OperationContext = {
        object,
        operation: 'aggregate',
        // On the opCtx so middleware-injected read filters (RLS/sharing) land
        // in `ast.where` — same #2737 hole as count(): a locally-built AST
        // aggregated over rows the caller may not see.
        ast: {
            object,
            where: query.where,
            groupBy: query.groupBy as any,
            aggregations: query.aggregations,
            // ENFORCED since #4286 (step 3). On the ast so the FLS predicate
            // guard walks its references (predicate-guard.ts) and a future
            // native pushdown has it in hand; the engine's post-aggregation
            // applyHaving() below is authoritative either way.
            having: query.having,
        },
        options: query,
        context: mergeReadContext(query?.context, options?.context),
      };
      this.resolveWhereTokens(opCtx.ast as QueryAST, opCtx.context);

      await this.executeWithMiddleware(opCtx, async () => {
        const ast = opCtx.ast as QueryAST;

        // Prefer driver.aggregate() when available — driver.find() in many
        // drivers (e.g. driver-sql) does not honor `groupBy` / `aggregations`
        // and would silently return ungrouped raw rows. Fall back to find()
        // for drivers that handle aggregations through their query AST.
        const drv = driver as any;
        // Structured groupBy items ({field, dateGranularity}) require the
        // driver to advertise per-granularity native bucket support via
        // `supports.queryDateGranularity[g]`. If every structured item is
        // supported we can push the aggregate down to the driver; otherwise
        // we fall back to driver.find() + in-memory bucketing so the result
        // remains correct on partial-support dialects (e.g. SQLite + week).
        const groupByItems = Array.isArray(query.groupBy) ? (query.groupBy as any[]) : [];
        const granularityCaps: Record<string, boolean> | undefined =
            drv?.supports?.queryDateGranularity;
        const structuredItems = groupByItems.filter((g) => typeof g !== 'string');
        const allStructuredSupported = structuredItems.every((g: any) => {
            if (!g?.dateGranularity) return true; // plain {field} object is fine
            return granularityCaps?.[g.dateGranularity] === true;
        });
        // ADR-0053 Phase 2 (D2): native driver date bucketing (`date_trunc`) is
        // UTC-only — SQLite has no tz database and MySQL needs tz tables loaded,
        // so pushing tz-aware bucketing down splits boundaries per dialect. When
        // a non-UTC reference timezone is in play we therefore force the
        // in-memory path: the date-range `where` still goes to the driver (only
        // matching rows are fetched), but bucketing runs uniformly in JS so a
        // row near a tz day-boundary lands identically on every driver.
        const tz = query.timezone;
        const hasDateBucket = structuredItems.some((g: any) => !!g?.dateGranularity);
        const tzRequiresInMemory = !!tz && tz !== 'UTC' && hasDateBucket;
        if (typeof drv.aggregate === 'function' && allStructuredSupported && !tzRequiresInMemory) {
            // HAVING is engine-owned (#4286): applied AFTER aggregation, over
            // the aggregated row's own columns (aggregation aliases + groupBy
            // projections), identically on both paths. No driver implements it
            // natively today; one that grows native support must advertise a
            // capability flag, at which point this post-filter becomes the
            // fallback tier — the dateGranularity two-tier pattern.
            const aggregated = await drv.aggregate(object, ast, this.buildDriverOptions(object, opCtx.context));
            return applyHaving(aggregated, ast.having);
        }
        // In-memory fallback path: ask the driver for raw rows, then bucket +
        // aggregate here. This guarantees `groupBy` (incl. structured items
        // carrying `dateGranularity`) and `aggregations` always work even on
        // drivers that have no native aggregation support (driver-rest,
        // driver-memory, partial SQL drivers), and is the path that honours a
        // non-UTC reference timezone.
        const raw = await driver.find(object, ast, this.buildDriverOptions(object, opCtx.context));
        return applyHaving(applyInMemoryAggregation(raw, ast, tz), ast.having);
      });

      return opCtx.result as any[];
  }
  
  /**
   * Run raw driver-specific commands (SQL for SqlDriver, REST for RestDriver, …).
   *
   * ⚠️ **Tenant isolation bypass.** Raw `execute()` does NOT thread the
   * caller's `ExecutionContext.tenantId` into a `WHERE organization_id`
   * predicate — drivers see the command verbatim. Callers MUST inline the
   * tenant filter themselves, or restrict raw execution to genuinely global
   * statements (schema migrations, sys_* / control-plane tables).
   *
   * Prefer the typed entry points (`find`, `update`, `delete`, `count`, …)
   * whenever feasible — they auto-apply tenancy + soft-delete + audit warnings.
   */
  async execute(command: any, options?: Record<string, any>): Promise<any> {
      // Driver selection priority:
      //   1. options.object  → route via getDriver(objectName)
      //   2. options.datasource → explicit driver name
      //   3. default driver (set via datasourceMapping or defaultDriver)
      // This lets system services (e.g. PackageService, AuditService) issue raw
      // SQL against the control-plane / default DB without having to know the
      // object name behind every CREATE TABLE / SELECT statement.
      let driver: IDataDriver | undefined;
      if (options?.object) {
          driver = this.getDriver(options.object);
      } else if (options?.datasource && this.drivers.has(options.datasource)) {
          driver = this.drivers.get(options.datasource);
      } else if (this.defaultDriver && this.drivers.has(this.defaultDriver)) {
          driver = this.drivers.get(this.defaultDriver);
      } else if (this.drivers.size === 1) {
          // Single registered driver — unambiguously the right one.
          driver = this.drivers.values().next().value;
      }

      if (!driver) {
          throw new Error(
              'Execute requires options.object to select a driver, or a default driver to be configured. ' +
              'Configure datasourceMapping with `default: true` or pass `{ object }` / `{ datasource }` in options.',
          );
      }
      if (!driver.execute) {
          throw new Error('Selected driver does not implement execute()');
      }

      // Support both call shapes:
      //   execute('SELECT ...', { args: [...] })
      //   execute({ sql: 'SELECT ...', args: [...] })
      let rawCommand: any = command;
      let params: any[] | undefined = options?.args ?? options?.params;
      if (command && typeof command === 'object' && !Array.isArray(command) && 'sql' in command) {
          rawCommand = command.sql;
          if (params === undefined) {
              params = command.args ?? command.params;
          }
      }

      return driver.execute(rawCommand, params, options);
  }

  /**
   * Execute a callback inside a database transaction.
   *
   * The callback receives a context object that should be passed to all
   * downstream `engine.insert/update/delete/find/findOne` calls (as
   * `{ context: trxCtx }`). The transaction handle threads through
   * `OperationContext.context.transaction` and the SQL driver's per-builder
   * `.transacting(trx)` call.
   *
   * - If the default driver does not support `beginTransaction`, the callback
   *   runs directly with the supplied base context (no rollback). This keeps
   *   the API safe to call on drivers without ACID support (e.g. the
   *   in-memory driver in tests).
   * - On callback success the transaction is committed; on any thrown error
   *   it is rolled back and the original error is re-thrown.
   *
   * Use case: multi-step operations that must be atomic (e.g. CRM
   * `convertLead`, which creates an account + contact + opportunity + flips
   * the lead in a single unit of work).
   */
  async transaction<T>(
    callback: (trxCtx: any) => Promise<T>,
    baseContext?: any,
  ): Promise<T> {
    // ADR-0067 D2 — JOIN an already-open ambient transaction instead of
    // opening a nested driver transaction. A nested begin would acquire a
    // second connection (a deadlock on single-connection pools like the
    // SQLite knex pool) and would NOT be covered by the outer rollback —
    // exactly the half-landing this join prevents: an outer batch
    // transaction (e.g. `publishPackageDrafts`) must own the one-and-only
    // commit/rollback for every write made through nested helpers (the
    // sys-metadata repository's `withTxn`, hook-driven writes, …).
    const ambient = this.txStore.getStore();
    if (ambient?.transaction) {
      return callback({ ...(baseContext ?? {}), transaction: ambient.transaction });
    }
    const driver = this.defaultDriver ? this.drivers.get(this.defaultDriver) : undefined;
    const drv = driver as any;
    if (!drv?.beginTransaction) {
      return callback(baseContext);
    }
    const trx = await drv.beginTransaction();
    const trxCtx = { ...(baseContext ?? {}), transaction: trx };
    try {
      // Run the callback inside the ambient transaction store so internal
      // queries during writes reuse this transaction's connection (ADR-0034).
      const result = await this.txStore.run({ transaction: trx }, () => callback(trxCtx));
      if (drv.commit) await drv.commit(trx);
      else if (drv.commitTransaction) await drv.commitTransaction(trx);
      return result;
    } catch (err) {
      try {
        if (drv.rollback) await drv.rollback(trx);
        else if (drv.rollbackTransaction) await drv.rollbackTransaction(trx);
      } catch {
        // swallow rollback failures so the original error surfaces
      }
      throw err;
    }
  }

  // ============================================
  // Compatibility / Convenience API
  // ============================================
  // These methods provide a higher-level API matching the @objectql/core
  // ObjectQL interface, enabling painless migration from the legacy layer.

  /**
   * Register a single object definition.
   * 
   * Proxies to SchemaRegistry.registerObject() with sensible defaults.
   * Fields without a `name` property are auto-assigned from their key.
   */
  registerObject(
    schema: ServiceObject,
    packageId: string = '__runtime__',
    namespace?: string
  ): string {
    // Auto-assign field names from keys
    if (schema.fields) {
      for (const [key, field] of Object.entries(schema.fields)) {
        if (field && typeof field === 'object' && !('name' in field)) {
          (field as any).name = key;
        }
      }
    }
    return this._registry.registerObject(schema, packageId, namespace);
  }

  /**
   * Unregister a single object by name.
   */
  unregisterObject(name: string, packageId?: string): void {
    if (packageId) {
      this._registry.unregisterObjectsByPackage(packageId);
    } else {
      // Remove from generic metadata as fallback
      this._registry.unregisterItem('object', name);
    }
  }

  /**
   * Get an object definition by name.
   * Alias for getSchema() — matches @objectql/core API.
   */
  getObject(name: string): ServiceObject | undefined {
    return this.getSchema(name);
  }

  /**
   * Get all registered object configs as a name→config map.
   * Matches @objectql/core getConfigs() API.
   */
  getConfigs(): Record<string, ServiceObject> {
    const result: Record<string, ServiceObject> = {};
    const objects = this._registry.getAllObjects();
    for (const obj of objects) {
      if (obj.name) {
        result[obj.name] = obj;
      }
    }
    return result;
  }

  /**
   * Get a registered driver by datasource name.
   * 
   * Unlike the private getDriver() (which resolves by object name),
   * this method directly looks up a driver by its registered name.
   */
  getDriverByName(name: string): IDataDriver | undefined {
    return this.drivers.get(name);
  }

  /**
   * Introspect a datasource's live remote schema (ADR-0015).
   *
   * Resolves the driver registered under `datasource` and delegates to its
   * `introspectSchema()` capability. Used by the external-datasource service
   * (and CLI/REST) to list remote tables and validate federated objects.
   *
   * @throws if the datasource has no registered driver, or the driver does
   *   not support introspection.
   */
  async introspectDatasource(datasource: string): Promise<unknown> {
    const driver = this.drivers.get(datasource) as any;
    if (!driver) {
      throw new Error(`[ObjectQL] Datasource '${datasource}' has no registered driver to introspect.`);
    }
    if (typeof driver.introspectSchema !== 'function') {
      throw new Error(`[ObjectQL] Driver for datasource '${datasource}' does not support introspectSchema().`);
    }
    return driver.introspectSchema();
  }

  /**
   * Get the driver responsible for the given object.
   *
   * Resolves datasource binding from the object's schema definition,
   * falling back to the default driver. This is a public version of
   * the internal getDriver() used by CRUD operations.
   *
   * @param objectName - FQN or short name of the registered object.
   * @returns The resolved IDataDriver, or undefined if no driver is available.
   */
  getDriverForObject(objectName: string): IDataDriver | undefined {
    try {
      return this.getDriver(objectName);
    } catch {
      return undefined;
    }
  }

  /**
   * Sync all registered object schemas to their respective drivers.
   * Call this after dynamically registering new objects at runtime
   * (e.g. after template seeding) to ensure tables/collections exist
   * before inserting seed data.
   */
  async syncSchemas(): Promise<void> {
    const allObjects = this._registry.getAllObjects();
    for (const obj of allObjects) {
      const driver = this.getDriverForObject(obj.name);
      if (!driver) continue;
      const tableName = StorageNameMapping.resolveTableName(obj);
      if (typeof (driver as any).syncSchemasBatch === 'function' && (driver as any).supports?.batchSchemaSync) {
        // Already handled per-driver below; skip individual call
      }
      if (typeof (driver as any).syncSchema === 'function') {
        try {
          await (driver as any).syncSchema(tableName, obj);
        } catch (e: unknown) {
          // #4632 — this catch used to be empty, with the comment "log
          // suppressed to avoid noise on already-synced tables". Suppressing an
          // already-synced no-op is not what it did: `syncSchema` is required to
          // be idempotent (see this method's doc comment), so a driver that
          // reaches this catch did NOT sync. The only callers are runtime
          // installs — marketplace plugin install, template seeding — which go
          // on to INSERT into a table this failure means does not exist, and
          // then report the install as successful. Nothing that claims to be
          // persisted afterwards is.
          this.logger.error(
            `Schema sync FAILED for object '${obj.name}' — its table/collection was NOT created or altered, yet the object is ` +
              `registered and will be written to: those writes will fail, or drop the columns that were never created. ` +
              `Any seeding or install step that continues past this point is not durable. ` +
              `Fix the driver error below, then re-run the install/sync.`,
            e as Error,
            { object: obj.name, tableName, driver: (driver as any)?.name },
          );
        }
      }
    }
  }

  /**
   * Sync a SINGLE object's physical storage (create/alter its table) on
   * demand. Boot-time {@link syncSchemas} runs once at startup, so an object
   * that becomes live at runtime (e.g. publishing a drafted object) has a
   * registry entry but no table — data CRUD then fails with "no such table"
   * until the next restart. Calling this right after the object is registered
   * makes it immediately usable. Idempotent: the SQL driver only creates the
   * table when absent (and alters to add new columns).
   */
  async syncObjectSchema(objectName: string): Promise<void> {
    const obj = this._registry.getObject(objectName) as any;
    if (!obj) return;
    const driver = this.getDriverForObject(objectName);
    if (!driver) return;
    // Federated (external) object (ADR-0015): register read metadata WITHOUT DDL
    // (its remote schema is owned externally). This is what an app's onEnable
    // calls after registering a late external driver so coercion maps + the
    // physical-table mapping exist for queries. See SqlDriver.registerExternalObject.
    if (obj.external != null) {
      if (typeof (driver as any).registerExternalObject === 'function') {
        await (driver as any).registerExternalObject(obj);
      }
      return;
    }
    if (typeof (driver as any).syncSchema !== 'function') return;
    const tableName = StorageNameMapping.resolveTableName(obj);
    await (driver as any).syncSchema(tableName, obj);
  }

  /**
   * Drop the physical storage (table/collection) backing an object — the
   * inverse of {@link syncObjectSchema}. DESTRUCTIVE: deletes all rows in the
   * table. Used by the protocol delete path when the caller explicitly opts
   * into storage teardown (e.g. discarding an object that was published only
   * to preview it). No-op when the object's driver does not expose `dropTable`.
   * Resolves the physical table name from the registered definition, falling
   * back to the bare name if the def was already removed.
   */
  async dropObjectSchema(objectName: string): Promise<void> {
    const obj = this._registry.getObject(objectName) as any;
    const driver = this.getDriverForObject(objectName);
    if (!driver || typeof (driver as any).dropTable !== 'function') return;
    const tableName = StorageNameMapping.resolveTableName(obj ?? ({ name: objectName } as any));
    await (driver as any).dropTable(tableName);
  }

  /**
   * Get a registered driver by datasource name.
   * Alias matching @objectql/core datasource() API.
   *
   * @throws Error if the datasource is not found
   */
  datasource(name: string): IDataDriver {
    const driver = this.drivers.get(name);
    if (!driver) {
      throw new Error(`[ObjectQL] Datasource '${name}' not found`);
    }
    return driver;
  }

  /**
   * Register a hook handler.
   * Convenience alias for registerHook() matching @objectql/core on() API.
   * 
   * Usage:
   *   ql.on('beforeInsert', 'user', async (ctx) => { ... });
   */
  on(
    event: string,
    objectName: string,
    handler: (ctx: HookContext) => Promise<void> | void,
    packageId?: string
  ): void {
    this.registerHook(event, handler, { object: objectName, packageId });
  }

  /**
   * Remove all hooks, actions, and objects contributed by a package.
   */
  removePackage(packageId: string): void {
    // Remove hooks
    for (const [key, handlers] of this.hooks.entries()) {
      const filtered = handlers.filter(h => h.packageId !== packageId);
      if (filtered.length !== handlers.length) {
        this.hooks.set(key, filtered);
      }
    }
    // Remove actions
    this.removeActionsByPackage(packageId);
    // Remove objects
    this._registry.unregisterObjectsByPackage(packageId, true);
  }

  /**
   * Gracefully shut down the engine, disconnecting all drivers.
   * Alias for destroy() — matches @objectql/core close() API.
   */
  async close(): Promise<void> {
    return this.destroy();
  }

  /**
   * Create a scoped execution context bound to this engine.
   * 
   * Usage:
   *   const ctx = engine.createContext({ userId: '...', tenantId: '...' });
   *   const users = ctx.object('user');
   *   await users.find({ where: { status: 'active' } });
   */
  createContext(ctx: Partial<ExecutionContext>): ScopedContext {
    return new ScopedContext(
      ExecutionContextSchema.parse(ctx),
      this
    );
  }

  /**
   * Static factory: create a fully configured ObjectQL instance.
   * 
   * Matches @objectql/core's `new ObjectQL(config)` pattern but also
   * registers drivers and objects, then calls init().
   * 
   * Usage:
   *   const ql = await ObjectQL.create({
   *     datasources: { default: myDriver },
   *     objects: { user: { name: 'user', fields: { ... } } }
   *   });
   */
  static async create(config: {
    datasources?: Record<string, IDataDriver>;
    objects?: Record<string, ServiceObject>;
    hooks?: Array<{ event: string; object: string; handler: (ctx: HookContext) => Promise<void> | void }>;
  }): Promise<ObjectQL> {
    const ql = new ObjectQL();

    // Register drivers
    if (config.datasources) {
      for (const [name, driver] of Object.entries(config.datasources)) {
        // Set driver name if not already set
        if (!driver.name) {
          (driver as any).name = name;
        }
        ql.registerDriver(driver, name === 'default');
      }
    }

    // Register objects
    if (config.objects) {
      for (const [_key, schema] of Object.entries(config.objects)) {
        ql.registerObject(schema);
      }
    }

    // Register hooks
    if (config.hooks) {
      for (const hook of config.hooks) {
        ql.on(hook.event, hook.object, hook.handler);
      }
    }

    // Initialize (connect drivers)
    await ql.init();

    return ql;
  }
}

/**
 * Repository scoped to a single object, bound to an execution context.
 *
 * Provides both IDataEngine-style methods (find, insert, update, delete)
 * and convenience aliases (create, updateById, deleteById) matching
 * the @objectql/core ObjectRepository API.
 */
export class ObjectRepository {
  constructor(
    private objectName: string,
    private context: ExecutionContextInput,
    private engine: IDataEngine & { executeAction?: (o: string, a: string, c: any) => Promise<any> }
  ) {}

  async find(query: any = {}): Promise<any[]> {
    return this.engine.find(this.objectName, {
      ...query,
      context: this.context,
    });
  }

  async findOne(query: any = {}): Promise<any> {
    return this.engine.findOne(this.objectName, {
      ...query,
      context: this.context,
    });
  }

  async insert(data: any): Promise<any> {
    return this.engine.insert(this.objectName, data, {
      context: this.context,
    });
  }

  /** Alias for insert() — matches @objectql/core convention */
  async create(data: any): Promise<any> {
    return this.insert(data);
  }

  async update(data: any, options: any = {}): Promise<any> {
    return this.engine.update(this.objectName, data, {
      ...options,
      context: this.context,
    });
  }

  /** Update a single record by ID */
  async updateById(id: string | number, data: any): Promise<any> {
    return this.engine.update(this.objectName, { ...data, id: id }, {
      where: { id: id },
      context: this.context,
    });
  }

  async delete(options: any = {}): Promise<any> {
    return this.engine.delete(this.objectName, {
      ...options,
      context: this.context,
    });
  }

  /** Delete a single record by ID */
  async deleteById(id: string | number): Promise<any> {
    return this.engine.delete(this.objectName, {
      where: { id: id },
      context: this.context,
    });
  }

  async count(query: any = {}): Promise<number> {
    return this.engine.count(this.objectName, {
      ...query,
      context: this.context,
    });
  }

  /** Aggregate query */
  async aggregate(query: any = {}): Promise<any[]> {
    return this.engine.aggregate(this.objectName, {
      ...query,
      context: this.context,
    });
  }

  /** Execute a named action registered on this object */
  async execute(actionName: string, params?: any): Promise<any> {
    if (this.engine.executeAction) {
      return this.engine.executeAction(this.objectName, actionName, {
        ...params,
        userId: this.context.userId,
        tenantId: this.context.tenantId,
        roles: this.context.positions,
      });
    }
    throw new Error(`Actions not supported by engine`);
  }
}

/**
 * Scoped execution context with object() accessor.
 * 
 * Provides identity (userId, tenantId/spaceId, roles),
 * repository access via object(), privilege escalation via sudo(),
 * and transactional execution via transaction().
 */
export class ScopedContext {
  constructor(
    private executionContext: ExecutionContextInput,
    private engine: IDataEngine
  ) {}

  /** Get a repository scoped to this context */
  object(name: string): ObjectRepository {
    return new ObjectRepository(name, this.executionContext, this.engine as any);
  }

  /** Create an elevated (system) context */
  sudo(): ScopedContext {
    return new ScopedContext(
      { ...this.executionContext, isSystem: true },
      this.engine
    );
  }

  /**
   * Execute a callback within a database transaction.
   *
   * The callback receives a new ScopedContext whose operations
   * share the same transaction handle. If the callback throws,
   * the transaction is rolled back; otherwise it is committed.
   *
   * Falls back to non-transactional execution if the driver
   * does not support transactions.
   */
  async transaction(callback: (trxCtx: ScopedContext) => Promise<any>): Promise<any> {
    const engine = this.engine as any;

    // Find the default driver for transaction support
    const driver = engine.defaultDriver
      ? engine.drivers?.get(engine.defaultDriver)
      : undefined;

    if (!driver?.beginTransaction) {
      // No transaction support — execute directly
      return callback(this);
    }

    const trx = await driver.beginTransaction();
    const trxCtx = new ScopedContext(
      { ...this.executionContext, transaction: trx },
      this.engine
    );
    // Share the engine's ambient transaction store so internal queries during
    // writes reuse this transaction's connection (ADR-0034).
    const txStore = (this.engine as any)?.txStore as
      | { run<R>(s: { transaction: unknown }, fn: () => R): R }
      | undefined;
    const runIn = <R>(fn: () => Promise<R>): Promise<R> =>
      txStore ? txStore.run({ transaction: trx }, fn) : fn();

    try {
      const result = await runIn(() => callback(trxCtx));
      if (driver.commit) await driver.commit(trx);
      else if (driver.commitTransaction) await driver.commitTransaction(trx);
      return result;
    } catch (error) {
      if (driver.rollback) await driver.rollback(trx);
      else if (driver.rollbackTransaction) await driver.rollbackTransaction(trx);
      throw error;
    }
  }

  /**
   * Resolve the default driver, if it exposes transaction primitives.
   * Shared by {@link transaction} and the discrete begin/commit/rollback trio.
   */
  private txDriver(): any | undefined {
    const engine = this.engine as any;
    const driver = engine.defaultDriver
      ? engine.drivers?.get(engine.defaultDriver)
      : undefined;
    return driver?.beginTransaction ? driver : undefined;
  }

  /**
   * Discrete transaction primitives — `begin` / `commit` / `rollback` as three
   * separate calls, in contrast to {@link transaction}'s single-callback form.
   *
   * This trio exists for callers that cannot keep a JS closure on the stack for
   * the lifetime of the transaction — chiefly the sandbox runner, where the
   * hook/action body's `ctx.api.transaction(fn)` is driven across many host
   * event-loop turns via deferred promises. Across those `setImmediate`
   * boundaries the engine's ambient `txStore` (AsyncLocalStorage) does NOT
   * survive, so the transaction handle is threaded **explicitly**: `begin`
   * returns a child ScopedContext carrying `transaction: trx` in its execution
   * context, and `resolveTx` honors that explicit handle ahead of the ambient
   * store. Every `object(...)` op on the returned context therefore reuses the
   * one connection without relying on ALS.
   *
   * Returns `null` when the driver has no transaction support — the caller then
   * runs non-transactionally against `this` (same graceful degrade as
   * {@link transaction}).
   */
  async beginTransaction(): Promise<{ ctx: ScopedContext; handle: unknown } | null> {
    const driver = this.txDriver();
    if (!driver) return null;
    const trx = await driver.beginTransaction();
    const ctx = new ScopedContext(
      { ...this.executionContext, transaction: trx },
      this.engine
    );
    return { ctx, handle: trx };
  }

  /** Commit a handle obtained from {@link beginTransaction}. */
  async commitTransaction(handle: unknown): Promise<void> {
    const driver = this.txDriver();
    if (!driver) return;
    if (driver.commit) await driver.commit(handle);
    else if (driver.commitTransaction) await driver.commitTransaction(handle);
  }

  /** Roll back a handle obtained from {@link beginTransaction}. */
  async rollbackTransaction(handle: unknown): Promise<void> {
    const driver = this.txDriver();
    if (!driver) return;
    if (driver.rollback) await driver.rollback(handle);
    else if (driver.rollbackTransaction) await driver.rollbackTransaction(handle);
  }

  get userId() { return this.executionContext.userId; }
  get tenantId() { return this.executionContext.tenantId; }
  /** Alias for tenantId — matches ObjectQLContext.spaceId convention */
  get spaceId() { return this.executionContext.tenantId; }
  get roles() { return this.executionContext.positions; }
  get isSystem() { return this.executionContext.isSystem; }

  /** Internal: expose the transaction handle for driver-level access */
  get transactionHandle() { return this.executionContext.transaction; }
}