@@ -83,6 +83,27 @@
  * bytes, indentation included, because they are the statement's own text.
  * `sql-driver.ts` no longer reads `recoverUnencodedJsonText`.
  *
+ * [#25521 cut 17] The movable members of the class's data-lifecycle rotation
+ * sub-banner joined them the same way: `ensureRotation`, which inventories a
+ * rotation-managed table's shards, adopts a legacy base table, column-syncs
+ * the retained shards, drops the expired ones and rebuilds the read view, with
+ * `rotationShardKey` and `rotationColumnList`, which it calls;
+ * `warnBuiltinColumnCollisions`, the warning the create, alter and shard paths
+ * share when a declared field collides with a builtin column; and
+ * `registerExternalObject`, the DDL-free registration of a federated object.
+ * The class keeps all five methods, and `SchemaSyncHost` grew by exactly the
+ * members these bodies read that it did not list yet; the fields among them
+ * go as getters too. `rotationShardKey` and `rotationColumnList` read no member
+ * of the driver, so their methods call them directly. `rotateShards`,
+ * `ensureShardTable`, `aliasShardBookkeeping` and the two tenant-field
+ * resolvers stay in the class. The module-level declarations these bodies read
+ * are imported from where they are declared: `declaredValueShapes` (which now
+ * carries `export`), `NUMERIC_SCALAR_TYPES` and `isMultiValuedColumn` from
+ * `sql-driver.ts`, the attribute check and its formatter from
+ * `builtin-column-collision.ts`, and the field-type sets and autonumber
+ * parsers from `@objectstack/spec/data` and `@objectstack/core`.
+ * `sql-driver.ts` no longer reads `formatAttribute` or `UndeliveredAttribute`.
+ *
  * A MOVE, not a redesign. Each function below is the body of the `SqlDriver`
  * method of the same name as it stood, re-indented one level and otherwise
  * unedited: the same dialect statements, the same rebuild procedure, the same
@@ -118,7 +139,10 @@ import { carriesNullSafeTenantKeyPart, ceremonyOwnedDropFinding, describeMissing
 import { isJsonColumnType, mediaColumnMoveDialect, mediaColumnMovePlan, type MediaColumnMovePlan, type MediaColumnMoveRefusal, type MediaColumnMoveScan } from './media-column-move.js';
 import { StandardErrorCode } from '@objectstack/spec/api';
 import { recoverUnencodedJsonText } from './unencoded-json-text.js';
-import { AUDIT_TIMESTAMP_COLUMNS, type IntrospectedColumn, isMultiValuedColumn, LOCK_WAIT_REFUSAL, type SqlDriver, SQLITE_NON_TEMPORAL_TEXT_REFS, SQLITE_TIME_EXPR_REFS } from './sql-driver.js';
+import { aggregandColumnClass } from '@objectstack/core';
+import { FILE_REFERENCE_TYPES, NUMERIC_VALUE_TYPES, parseAutonumberFormat, resolveAutonumberFormat, type AutonumberToken } from '@objectstack/spec/data';
+import { formatAttribute, undeliveredStorageAttributes, type UndeliveredAttribute } from './builtin-column-collision.js';
+import { AUDIT_TIMESTAMP_COLUMNS, declaredValueShapes, type IntrospectedColumn, isMultiValuedColumn, LOCK_WAIT_REFUSAL, NUMERIC_SCALAR_TYPES, type SqlDriver, SQLITE_NON_TEMPORAL_TEXT_REFS, SQLITE_TIME_EXPR_REFS } from './sql-driver.js';
 
 /**
  * The `SqlDriver` members the functions below read — every one of them, and
@@ -129,33 +153,43 @@ import { AUDIT_TIMESTAMP_COLUMNS, type IntrospectedColumn, isMultiValuedColumn,
  * for which members are getters.
  */
 export interface SchemaSyncHost {
+  aliasShardBookkeeping: SqlDriver['aliasShardBookkeeping'];
   applyDeclaredColumnDefault: SqlDriver['applyDeclaredColumnDefault'];
   applyDriftOpInPlace: SqlDriver['applyDriftOpInPlace'];
   applyIndexDriftOp: SqlDriver['applyIndexDriftOp'];
   applyMigrationEntries: SqlDriver['applyMigrationEntries'];
   applyNullSafeUniquePreflight: SqlDriver['applyNullSafeUniquePreflight'];
   assertSchemaMutable: SqlDriver['assertSchemaMutable'];
   readonly autoMigrate: SqlDriver['autoMigrate'];
+  readonly autoNumberFields: SqlDriver['autoNumberFields'];
+  readonly booleanFields: SqlDriver['booleanFields'];
   buildRebuiltColumn: SqlDriver['buildRebuiltColumn'];
   readonly canonicalDatetimeFields: SqlDriver['canonicalDatetimeFields'];
   readonly canonicalTimeFields: SqlDriver['canonicalTimeFields'];
+  readonly columnFieldByObject: SqlDriver['columnFieldByObject'];
+  computeAndRecordTenantField: SqlDriver['computeAndRecordTenantField'];
   convergeJsonColumn: SqlDriver['convergeJsonColumn'];
   countNonTemporalTextRows: SqlDriver['countNonTemporalTextRows'];
   createHashShadowUniqueIndex: SqlDriver['createHashShadowUniqueIndex'];
   createNullSafeUniqueIndex: SqlDriver['createNullSafeUniqueIndex'];
+  readonly dateFields: SqlDriver['dateFields'];
   readonly datetimeFields: SqlDriver['datetimeFields'];
   detectTableDrift: SqlDriver['detectTableDrift'];
   detectTableIndexDrift: SqlDriver['detectTableIndexDrift'];
   readonly dialectName: SqlDriver['dialectName'];
   readonly driftWarned: SqlDriver['driftWarned'];
   dropIndexIfExists: SqlDriver['dropIndexIfExists'];
   dropOrphanedHashShadowColumn: SqlDriver['dropOrphanedHashShadowColumn'];
+  ensureShardTable: SqlDriver['ensureShardTable'];
   explainUnkeyableTextColumn: SqlDriver['explainUnkeyableTextColumn'];
+  readonly fieldColumnByObject: SqlDriver['fieldColumnByObject'];
   readonly fileColumnsMoved: SqlDriver['fileColumnsMoved'];
   readonly flushingDeferredDdl: SqlDriver['flushingDeferredDdl'];
+  readonly fractionalNumericFields: SqlDriver['fractionalNumericFields'];
   getExistingIndexNames: SqlDriver['getExistingIndexNames'];
   introspectColumns: SqlDriver['introspectColumns'];
   introspectIndexes: SqlDriver['introspectIndexes'];
+  isJsonField: SqlDriver['isJsonField'];
   readonly isMysql: SqlDriver['isMysql'];
   readonly isPostgres: SqlDriver['isPostgres'];
   isProductionEnv: SqlDriver['isProductionEnv'];
@@ -169,18 +203,28 @@ export interface SchemaSyncHost {
   readonly managedObjectIndexes: SqlDriver['managedObjectIndexes'];
   readonly mediaFields: SqlDriver['mediaFields'];
   nowColumnDefault: SqlDriver['nowColumnDefault'];
+  readonly numericFields: SqlDriver['numericFields'];
+  readonly numericValueFields: SqlDriver['numericValueFields'];
+  readonly objectByPhysicalTable: SqlDriver['objectByPhysicalTable'];
+  readonly physicalSchemaByObject: SqlDriver['physicalSchemaByObject'];
+  readonly physicalTableByObject: SqlDriver['physicalTableByObject'];
   probeNullSafeUniqueDuplicates: SqlDriver['probeNullSafeUniqueDuplicates'];
   rebuildSqliteTablePatched: SqlDriver['rebuildSqliteTablePatched'];
   resolveTenantField: SqlDriver['resolveTenantField'];
   restoreBareIndexAfterFailedTighten: SqlDriver['restoreBareIndexAfterFailedTighten'];
+  rotationColumnList: SqlDriver['rotationColumnList'];
+  rotationShardKey: SqlDriver['rotationShardKey'];
+  readonly rotationStateByTable: SqlDriver['rotationStateByTable'];
   readonly runtimeCreatedIndexes: SqlDriver['runtimeCreatedIndexes'];
   readonly schemaMode: SqlDriver['schemaMode'];
   sqliteCanonicalDatetimeSql: SqlDriver['sqliteCanonicalDatetimeSql'];
   sqliteCanonicalTimeSql: SqlDriver['sqliteCanonicalTimeSql'];
   sqliteNonTemporalTextSql: SqlDriver['sqliteNonTemporalTextSql'];
   syncDeclaredIndexes: SqlDriver['syncDeclaredIndexes'];
   syncTableIndexes: SqlDriver['syncTableIndexes'];
+  readonly tenantFieldByTable: SqlDriver['tenantFieldByTable'];
   readonly timeFields: SqlDriver['timeFields'];
+  readonly valueShapeFields: SqlDriver['valueShapeFields'];
   varcharColumnChars: SqlDriver['varcharColumnChars'];
   withPinnedSession: SqlDriver['withPinnedSession'];
 }
@@ -1651,3 +1695,220 @@ export async function legacyMysqlTimeColumns(this: Pick<SchemaSyncHost, 'isMysql
     }))
     .filter((c) => c.name && candidates.has(c.name));
 }
+
+/** The body of `SqlDriver#rotationShardKey`; the method's docblock there is its contract. */
+export function rotationShardKey(nowMs: number, unit: 'day' | 'week' | 'month'): string {
+  const ymd = (ms: number) => new Date(ms).toISOString().slice(0, 10).replace(/-/g, '');
+  if (unit === 'month') return ymd(nowMs).slice(0, 6);
+  if (unit === 'week') {
+    const dow = (new Date(nowMs).getUTCDay() + 6) % 7; // Monday = 0
+    return ymd(nowMs - dow * 86_400_000);
+  }
+  return ymd(nowMs);
+}
+
+/** The body of `SqlDriver#ensureRotation`; the method's docblock there is its contract. */
+export async function ensureRotation(this: Pick<SchemaSyncHost, 'aliasShardBookkeeping' | 'ensureShardTable' | 'knex' | 'logger' | 'rotationColumnList' | 'rotationShardKey' | 'rotationStateByTable'>, tableName: string, obj: { name: string; fields?: Record<string, any>; tenancy?: any; indexes?: any[] }, policy: { shards: number; unit: 'day' | 'week' | 'month' }, nowMs: number = Date.now()): Promise<{ object: string; current: string; shards: string[]; dropped: string[] }> {
+  const current = `${tableName}__r${this.rotationShardKey(nowMs, policy.unit)}`;
+
+  // Physical inventory: the base name (table before adoption, view after)
+  // and every existing shard.
+  const esc = tableName.replace(/[\\%_]/g, '\\$&');
+  const raw: any = await this.knex.raw(
+    `SELECT name, type FROM sqlite_master WHERE name = ? OR (name LIKE ? ESCAPE '\\')`,
+    [tableName, `${esc}\\_\\_r%`],
+  );
+  const rows: Array<{ name: string; type: string }> = Array.isArray(raw) ? raw : raw?.rows ?? [];
+  const baseType = rows.find((r) => r.name === tableName)?.type as 'table' | 'view' | undefined;
+  const shardNames = new Set(
+    rows.map((r) => r.name).filter((n) => n !== tableName && /__r\d{6,8}$/.test(n)),
+  );
+
+  // Adopt a legacy pre-rotation table: its whole history becomes the
+  // current shard (coarse, but safe — it then ages out of the window).
+  if (baseType === 'table') {
+    if (!shardNames.has(current)) {
+      await this.knex.schema.renameTable(tableName, current);
+    } else {
+      // Partial-failure recovery: both exist — merge, then drop the base.
+      await this.knex.raw(`INSERT INTO "${current}" SELECT * FROM "${tableName}"`);
+      await this.knex.schema.dropTable(tableName);
+    }
+    shardNames.add(current);
+  }
+  shardNames.add(current);
+
+  // Time-based window: retain shards whose period falls inside the last
+  // `shards × unit` (ends at the current period); everything older is the
+  // O(1) reclaim. Count-based retention would silently stretch the window
+  // when rotation cadence has gaps.
+  const unitMs = { day: 86_400_000, week: 7 * 86_400_000, month: 30 * 86_400_000 }[policy.unit];
+  const oldestRetainedKey = this.rotationShardKey(nowMs - (Math.max(1, policy.shards) - 1) * unitMs, policy.unit);
+  const parsed = [...shardNames].sort((a, b) => b.localeCompare(a));
+  const keyOf = (n: string) => n.slice(tableName.length + 3);
+  const retained = parsed.filter((n) => keyOf(n) >= oldestRetainedKey);
+  const dropped = parsed.filter((n) => keyOf(n) < oldestRetainedKey);
+
+  // Column-sync every retained shard (creates the current one; adds any
+  // newly declared columns to older shards so the UNION stays uniform).
+  for (const shard of retained) {
+    await this.ensureShardTable(shard, obj, tableName);
+    this.aliasShardBookkeeping(tableName, shard);
+  }
+
+  for (const d of dropped) {
+    await this.knex.schema.dropTableIfExists(d);
+  }
+
+  // Rebuild the read view over the retained set, newest shard first. An
+  // explicit column list (not `*`) keeps the view stable when old shards
+  // carry orphaned columns.
+  const cols = this.rotationColumnList(obj);
+  await this.knex.raw(`DROP VIEW IF EXISTS "${tableName}"`);
+  await this.knex.raw(
+    `CREATE VIEW "${tableName}" AS ` + retained.map((s) => `SELECT ${cols} FROM "${s}"`).join(' UNION ALL '),
+  );
+
+  this.rotationStateByTable.set(tableName, { shards: retained, current });
+  if (dropped.length > 0) {
+    (this.logger as { info?: (msg: string) => void }).info?.(
+      `[sql-driver] rotated ${tableName}: ${retained.length} shard(s) live, dropped ${dropped.join(', ')}`,
+    );
+  }
+  return { object: tableName, current, shards: retained, dropped };
+}
+
+/** The body of `SqlDriver#warnBuiltinColumnCollisions`; the method's docblock there is its contract. */
+export function warnBuiltinColumnCollisions(this: Pick<SchemaSyncHost, 'logger'>, tableName: string, fields: Record<string, any> | undefined, builtinColumns: ReadonlySet<string>, phase: 'create' | 'alter' | 'shard'): string[] {
+  const where = {
+    create: `while creating table "${tableName}"`,
+    alter: `while syncing existing table "${tableName}"`,
+    shard: `while syncing shard "${tableName}"`,
+  }[phase];
+
+  const warned: string[] = [];
+  for (const [field, declaration] of Object.entries(fields ?? {})) {
+    if (!builtinColumns.has(field)) continue;
+    const undelivered: UndeliveredAttribute[] = undeliveredStorageAttributes(
+      field,
+      declaration as Record<string, unknown>,
+    );
+    // Declared, colliding, and yet nothing was lost — the honoured half only.
+    if (undelivered.length === 0) continue;
+
+    this.logger.warn(
+      `[sql-driver] ${where}: declared field '${field}' asks for storage the platform's own ` +
+        `'${field}' column does not provide — ${undelivered.map(formatAttribute).join('; ')}. ` +
+        `The platform emits id/created_at/updated_at itself, so THOSE attributes are not applied; ` +
+        `the rest of the declaration (label, help text, the ADR-0113 write contract, and everything ` +
+        `other layers read) is honoured as written. Drop the storage attribute(s) named above, or ` +
+        `rename the field if you meant a column of your own.`,
+      { table: tableName, field, phase, undelivered: undelivered.map((a) => a.key) },
+    );
+    warned.push(field);
+  }
+  return warned;
+}
+
+/** The body of `SqlDriver#rotationColumnList`; the method's docblock there is its contract. */
+export function rotationColumnList(obj: { fields?: Record<string, any> }): string {
+  const builtin = ['id', 'created_at', 'updated_at'];
+  const declared = Object.keys(obj.fields ?? {}).filter((f) => !builtin.includes(f));
+  return [...builtin, ...declared].map((c) => `"${c}"`).join(', ');
+}
+
+/** The body of `SqlDriver#registerExternalObject`; the method's docblock there is its contract. */
+export function registerExternalObject(this: Pick<SchemaSyncHost, 'autoNumberFields' | 'booleanFields' | 'columnFieldByObject' | 'computeAndRecordTenantField' | 'dateFields' | 'datetimeFields' | 'fieldColumnByObject' | 'fractionalNumericFields' | 'isJsonField' | 'isSqlite' | 'jsonFields' | 'logger' | 'mediaFields' | 'numericFields' | 'numericValueFields' | 'objectByPhysicalTable' | 'physicalSchemaByObject' | 'physicalTableByObject' | 'tenantFieldByTable' | 'timeFields' | 'valueShapeFields'>, schema: {
+  name: string;
+  fields?: Record<string, any>;
+  tenancy?: any;
+  external?: { remoteName?: string; remoteSchema?: string; columnMap?: Record<string, string> };
+}): void {
+  const key = schema.name;
+  const remoteName = schema.external?.remoteName || schema.name;
+  const remoteSchema = schema.external?.remoteSchema;
+  this.physicalTableByObject[key] = remoteName;
+  this.objectByPhysicalTable[remoteName] = key;
+  if (remoteSchema) {
+    if (this.isSqlite) {
+      this.logger.warn(
+        `[sql-driver] external object "${key}" declares remoteSchema="${remoteSchema}" but SQLite has no schema namespace; ignoring (treating "${remoteName}" as a bare table).`,
+      );
+    } else {
+      this.physicalSchemaByObject[key] = remoteSchema;
+    }
+  }
+
+  // External columnMap (ADR-0015) is declared as { remoteColumn -> localField }.
+  // Keep it for read-output remap, and invert to { localField -> remoteColumn }
+  // for WHERE/ORDER BY/write translation. Absent => managed-identical behavior.
+  const columnMap = schema.external?.columnMap;
+  if (columnMap && typeof columnMap === 'object' && Object.keys(columnMap).length > 0) {
+    const fieldToCol: Record<string, string> = {};
+    const colToField: Record<string, string> = {};
+    for (const [remoteCol, localField] of Object.entries(columnMap)) {
+      if (typeof localField === 'string' && localField) {
+        fieldToCol[localField] = remoteCol;
+        colToField[remoteCol] = localField;
+      }
+    }
+    this.fieldColumnByObject[key] = fieldToCol;
+    this.columnFieldByObject[key] = colToField;
+  }
+
+  const jsonCols: string[] = [];
+  const mediaCols: string[] = [];
+  const booleanCols: string[] = [];
+  const numericCols: string[] = [];
+  const numericValueCols: string[] = [];
+  const fractionalCols: string[] = [];
+  const dateCols: string[] = [];
+  const datetimeCols: string[] = [];
+  const timeCols: string[] = [];
+  const autoNumberCols: Array<{ name: string; format: string; tokens: AutonumberToken[]; tenantField: string | null }> = [];
+
+  const tenantField = this.computeAndRecordTenantField(key, schema);
+  if (schema.fields) {
+    for (const [name, field] of Object.entries<any>(schema.fields)) {
+      const type = field.type || 'string';
+      if (this.isJsonField(type, field)) jsonCols.push(name);
+      // Unconditional, on BOTH arms — see {@link mediaFields}. The read-side
+      // legacy-encoding repair runs on a deployment that has not moved too.
+      if (!isMultiValuedColumn(type, field) && FILE_REFERENCE_TYPES.has(type)) mediaCols.push(name);
+      // [commit d46deba19] SCALAR only — the house spelling its three neighbours in
+      // this block already carry, and this line was the single omission. See
+      // {@link booleanFields}: every reader of this registry presents its
+      // entry as ONE JS boolean, which for a multi-valued (JSON) column
+      // collapses the parsed array to `true`. [#17469] The exclusion asks
+      // {@link isMultiValuedColumn}, so "excluded from the scalar registries"
+      // and "stored as a JSON column" stay the SAME population.
+      if ((type === 'boolean' || type === 'toggle') && !isMultiValuedColumn(type, field)) booleanCols.push(name);
+      if (NUMERIC_SCALAR_TYPES.has(type) && !isMultiValuedColumn(type, field)) numericCols.push(name);
+      // [#16318] The authorable half only — see {@link numericValueFields}.
+      if (NUMERIC_VALUE_TYPES.has(type) && !isMultiValuedColumn(type, field)) numericValueCols.push(name);
+      // [#20387, #21042] See {@link fractionalNumericFields}: the predicate's
+      // `'fractional'` class, scalar only by the predicate's own reading.
+      if (aggregandColumnClass({ type, multiple: field?.multiple }) === 'fractional') fractionalCols.push(name);
+      if (type === 'date') dateCols.push(name);
+      if (type === 'datetime') datetimeCols.push(name);
+      if (type === 'time') timeCols.push(name);
+      if (type === 'auto_number' || type === 'autonumber') {
+        const fmt = resolveAutonumberFormat(field);
+        autoNumberCols.push({ name, format: fmt, tokens: parseAutonumberFormat(fmt), tenantField });
+      }
+    }
+  }
+  this.jsonFields[key] = jsonCols;
+  // [#20444] The declared value shapes `$empty` expands — see {@link valueShapeFields}.
+  this.valueShapeFields[key] = declaredValueShapes(schema.fields);
+  this.mediaFields[key] = mediaCols;
+  this.booleanFields[key] = booleanCols;
+  this.numericFields[key] = numericCols;
+  this.numericValueFields[key] = numericValueCols;
+  this.fractionalNumericFields[key] = fractionalCols;
+  this.autoNumberFields[key] = autoNumberCols;
+  this.tenantFieldByTable[key] = tenantField;
+  if (dateCols.length) this.dateFields[key] = new Set(dateCols);
+  if (datetimeCols.length) this.datetimeFields[key] = new Set(datetimeCols);
+  if (timeCols.length) this.timeFields[key] = new Set(timeCols);
+}