diff --git a/src/core.ts b/src/core.ts
index 1e5e8d8..a419961 100644
--- a/src/core.ts
+++ b/src/core.ts
@@ -6,6 +6,8 @@
  */
 
 import type { IStorageEngine } from './engine/interface';
+import { ChangeNotifierEngine } from './engine/change-notifier';
+import type { ChangeEvent } from './engine/change-notifier';
 import type { DatabaseConfig, ColumnDef } from './constants';
 import { DB_DEFAULTS, DatabaseError } from './constants';
 import { MemoryEngine } from './engine/memory';
@@ -81,10 +83,17 @@ export class MetonaSqlark {
       this.channel.onmessage = (event) => {
         const msg = event.data as { type?: string; table?: string } | null;
         if (!msg || msg.type !== 'change') return;
-        this.emit(msg.table ?? '', { type: 'external', table: msg.table ?? '' });
-        // Hybrid 引擎：从磁盘重载内存，保证读到其他标签页的最新数据
-        if (this.engine instanceof HybridEngine) {
-          (this.engine as HybridEngine).reloadMemoryFromDisk().catch(() => {
+        // v0.8.0: 外部事件只派发给本地订阅者，**不得再次广播** ——
+        // 否则两个标签页会互相转发形成无限广播循环（实测 8 次以上且不终止）。
+        void this.emitExternal(msg.table ?? '');
+        // Hybrid 引擎：从磁盘重载内存，保证读到其他标签页的最新数据。
+        //
+        // v0.8.0: 引擎现在被 ChangeNotifierEngine 装饰，`this.engine instanceof HybridEngine`
+        // 恒为 false —— 因此改为对**内层**引擎做能力探测。这也是审计指出的
+        // "用 instanceof 做引擎特判"的隐患：装饰器一加就静默失效。
+        const inner = this.unwrapEngine();
+        if (inner instanceof HybridEngine) {
+          (inner as HybridEngine).reloadMemoryFromDisk().catch(() => {
             // 重载失败不影响主流程（下次读可能短暂过期）
           });
         }
@@ -102,6 +111,28 @@ export class MetonaSqlark {
     // 打开连接
     await this.engine.open(this.name, this.version);
 
+    // v0.8.0（A9）：把引擎包进变更通知装饰器 —— **唯一**的变更事件汇聚点。
+    // 三个写入入口（SQL / Table API / QueryBuilder）与事务内写入都必须经过引擎接口，
+    // 因此在这里拦一次即可全覆盖，避免在三条路径上各写一份"变更描述"逻辑。
+    this.engine = new ChangeNotifierEngine(
+      this.engine,
+      (error) => this._onError(error),
+      (table) => this.broadcastChange(table),
+    );
+    this.notifier = this.engine as ChangeNotifierEngine;
+    // 把引擎层变更事件接入 db.subscribe 的订阅表（listeners）
+    this.notifier.addListener(async (event) => {
+      const set = this.listeners.get(`change:${event.table}`);
+      if (!set || set.size === 0) return;
+      for (const cb of [...set]) {
+        try {
+          await (cb as (e: ChangeEvent) => void | Promise<void>)(event);
+        } catch (error) {
+          this._onError(error as Error);
+        }
+      }
+    });
+
     // v0.4.2-fix (P2-7): 从库内加载持久化的迁移版本，
     // 重启后 migrateTo 从持久化版本继续执行，不再每次从 config.version 重置
     if (typeof this.engine.getMeta === 'function') {
@@ -169,7 +200,9 @@ export class MetonaSqlark {
         this.engine,
         name,
         this.executor,
-        (tableName) => this.broadcastChange(tableName),
+        // v0.8.0: 变更事件由引擎层 ChangeNotifierEngine 统一产生，
+        // 此处不再重复广播（Table API 与 SQL 路径曾各广播一次 → 同一次写入触发两遍）
+        () => { /* no-op: see ChangeNotifierEngine */ },
         (hook, args) => this.pluginManager.trigger(hook, ...args),
       );
       this.tableCache.set(name, t);
@@ -221,9 +254,8 @@ export class MetonaSqlark {
         await this.triggerStatementHooks(stmt, 'before');
         result = await this.executor.execute(stmt);
         await this.triggerStatementHooks(stmt, 'after', result);
-        // v0.3.2: 写语句广播表变更（多标签页同步）
-        const table = this.writeStatementTable(stmt);
-        if (table) this.broadcastChange(table);
+        // v0.8.0: 变更广播与订阅事件统一由引擎层 ChangeNotifierEngine 产生，
+        // 此处不再手工 broadcastChange（否则同一次写入会广播两次）。
       }
     } catch (error) {
       this._onError(error as Error);
@@ -470,20 +502,65 @@ export class MetonaSqlark {
 
   // ---- 发布订阅 ----
 
+  /** 变更通知引擎（init 后可用）；未初始化时为 null */
+  private notifier: ChangeNotifierEngine | null = null;
+
   private listeners: Map<string, Set<(data: unknown) => void>> = new Map();
 
-  /** 订阅表变更 */
-  subscribe(tableName: string, callback: (event: { type: string; row?: unknown; table?: string }) => void): () => void {
+  /**
+   * 订阅表变更。
+   *
+   * v0.8.0 修复：此前**本地写入永不触发** —— 全库唯一调用 `emit` 的地方在
+   * BroadcastChannel 收到其它标签页消息的分支里，因此 README「订阅表变更」与
+   * site/docs.html 的 `event.type: 'insert' | 'update' | 'delete'` 示例全都不成立。
+   * 现在本地写入（SQL / Table API / QueryBuilder / 事务内）都会产生事件。
+   *
+   * 现在返回的函数是**同步**退订函数（与既有 API 兼容）。
+   */
+  subscribe(
+    tableName: string,
+    callback: (event: ChangeEvent) => void | Promise<void>,
+  ): () => void {
     const key = `change:${tableName}`;
     if (!this.listeners.has(key)) this.listeners.set(key, new Set());
     this.listeners.get(key)!.add(callback as (data: unknown) => void);
-    return () => this.listeners.get(key)?.delete(callback as (data: unknown) => void);
+    return () => { this.listeners.get(key)?.delete(callback as (data: unknown) => void); };
   }
 
-  /** 触发变更事件 */
-  emit(tableName: string, event: { type: string; row?: unknown; table?: string }): void {
-    const key = `change:${tableName}`;
-    this.listeners.get(key)?.forEach((cb) => cb(event));
+  /**
+   * 手动触发变更事件（保留为公开 API：自定义写入路径可显式通知订阅者）。
+   * 现在也支持 await —— 订阅者的 Promise 会被等待。
+   */
+  async emit(tableName: string, event: Partial<ChangeEvent> & { type: ChangeEvent['type'] }): Promise<void> {
+    await this.dispatchChange({ table: tableName, ...event });
+  }
+
+  /**
+   * v0.8.0: 派发"来自其它标签页"的变更事件。
+   * 只走本地订阅者，不触发 onBroadcast（避免 A↔B 互相转发的无限循环）。
+   */
+  private async emitExternal(tableName: string): Promise<void> {
+    const event: ChangeEvent = { type: 'external', table: tableName };
+    const set = this.listeners.get(`change:${tableName}`);
+    if (!set) return;
+    for (const cb of [...set]) {
+      try {
+        await (cb as (e: ChangeEvent) => void | Promise<void>)(event);
+      } catch (error) {
+        this._onError(error as Error);
+      }
+    }
+  }
+
+  /** 内部：把一次变更同时派发给本地订阅者与跨标签页广播 */
+  private async dispatchChange(event: ChangeEvent): Promise<void> {
+    if (this.notifier) {
+      // notifier.dispatch 内部已包含跨标签页广播，这里不重复调用
+      await this.notifier.dispatch(event);
+      return;
+    }
+    // init 之前（notifier 尚未建立）也能广播
+    this.broadcastChange(event.table);
   }
 
   // ---- 多标签页同步（v0.3.2） ----
@@ -654,11 +731,21 @@ export class MetonaSqlark {
     this.ready = false;
   }
 
-  /** 获取底层引擎 */
+  /** 获取底层引擎（含变更通知装饰器） */
   getEngine(): IStorageEngine {
     return this.engine;
   }
 
+  /**
+   * v0.8.0: 取**未装饰**的真实存储引擎。
+   *
+   * 引擎在 init 时被 ChangeNotifierEngine 包了一层，因此需要引擎特化能力
+   * （如 HybridEngine.reloadMemoryFromDisk）时必须先解包，否则 instanceof 恒 false。
+   */
+  private unwrapEngine(): IStorageEngine {
+    return this.notifier ? this.notifier.getInner() : this.engine;
+  }
+
   // ---- 内部 ----
 
   private createEngine(): IStorageEngine {
diff --git a/src/engine/aria/index.ts b/src/engine/aria/index.ts
index 28765fb..c1fae6c 100644
--- a/src/engine/aria/index.ts
+++ b/src/engine/aria/index.ts
@@ -2195,6 +2195,11 @@ export class AriaEngine implements IStorageEngine {
       if ('$eq' in c) {
         // v0.6.2-fix(P1): 同上，$eq: null（IS NULL）不走索引
         if (c.$eq === null) continue;
+        // v0.8.0 根治（与 MemoryEngine 同步）：非原始值一律不走索引。
+        // `String({...})` 得到 "[object Object]" 这类无意义键，索引查找必然为空
+        // 并短路全表扫描 → 结果静默为空。真实触发场景是"未解析的操作数"：
+        //   { $eq: { $col: 'y' } }   ← 列对列比较（t.x = t.y）
+        if (typeof c.$eq === 'object') continue;
         const v = String(c.$eq);
         return this.indexScanToRows(tableName, pkCol, idxLsm, v, v);
       }
@@ -2202,6 +2207,9 @@ export class AriaEngine implements IStorageEngine {
       if ('$in' in c && Array.isArray(c.$in)) {
         // v0.6.2-fix(P1): IN 列表含 null 不走索引（索引不含 null 条目，会漏匹配 null 行）
         if (c.$in.some((v) => v === null)) continue;
+        // v0.8.0: IN 列表含非原始值（未解析的 $subquery / $col）不走索引 ——
+        // String() 会得到无意义键，查找为空并短路全表扫描 → 静默空结果
+        if (c.$in.some((v) => typeof v === 'object' && v !== null)) continue;
         // v0.7.3-perf: 批级预加载全部值的索引范围 + 主表行（各一次 drainChain）——
         // 此前逐值 indexScanToRows：每个值一次 prefetchRange + prefetchKeys，
         // 后台 compaction 长耗时时 N 倍放大（与 v0.6.1 修的 insert 批量预加载
diff --git a/src/engine/change-notifier.ts b/src/engine/change-notifier.ts
new file mode 100644
index 0000000..1c0dda9
--- /dev/null
+++ b/src/engine/change-notifier.ts
@@ -0,0 +1,352 @@
+/**
+ * metona-sqlark Engine Change Notifier（v0.8.0）
+ * @module engine/change-notifier
+ *
+ * ============================================================================
+ * 为什么需要它（PLAN-v0.7.5.md A9）
+ * ============================================================================
+ * `db.subscribe(table, fn)` 此前对**本地写入永不触发**：全库只有一处调用 `emit`，
+ * 而那一处在 BroadcastChannel 收到**其它标签页**消息时才执行。于是：
+ *   - README:232「订阅表变更」、site/docs.html:667-679 的示例
+ *     （`event.type: 'insert' | 'update' | 'delete'`、`event.row`）全部不成立；
+ *   - 唯一的实际触发条件是 `multiTabSync: true` 且收到 `external` 事件。
+ *
+ * 更麻烦的是原因：写入路径有**三条**（SQL 语句、Table API、QueryBuilder），各自
+ * 只调用 `broadcastChange(tableName)`，谁都不知道"改了哪些行"。所以此前即便想接线，
+ * 也要在三处分别实现一遍变更描述逻辑 —— 那正是本项目反复出问题的模式
+ * （同一语义多份实现 → 漂移）。
+ *
+ * ============================================================================
+ * 本实现的做法
+ * ============================================================================
+ * 用一个 `IStorageEngine` **装饰器**把变更通知收敛到唯一位置：所有写入都必须经过
+ * 引擎接口，因此在这里拦截一次即可覆盖全部入口（SQL / Table / Builder / 事务内）。
+ *
+ * 事件语义（对文档承诺的兑现）：
+ *   - INSERT：逐行一个事件，携带该行（`row` 与主键 `key`）；
+ *   - UPDATE：先按 WHERE 查出**将被修改的行**（快照），写入成功后逐行发事件并
+ *     携带**更新后的行**；查不到（例如 WHERE 无法在写入前求值）时退化为表级事件，
+ *     只带 `count`；
+ *   - DELETE：同理，逐行发事件并携带**被删除前的行**；
+ *   - CLEAR / DDL：表级事件（`count` 可选）。
+ *
+ * 订阅方回调返回 Promise 时会被 await（保证"写完通知完"的顺序），
+ * 但**订阅方的异常不会影响写入结果** —— 写入已经成功，通知失败只上报 onError。
+ */
+
+import type { IStorageEngine } from './interface';
+import { DatabaseError } from '../constants';
+import type { QueryPlan } from '../constants';
+
+/** 变更类型（与 site/docs.html 承诺的 `event.type` 对齐） */
+export type ChangeType = 'insert' | 'update' | 'delete' | 'clear' | 'ddl' | 'external';
+
+/** 变更事件 */
+export interface ChangeEvent {
+  /** 变更类型 */
+  type: ChangeType;
+  /** 表名 */
+  table: string;
+  /** 受影响的行（可用时提供） */
+  row?: Record<string, unknown>;
+  /** 受影响行的主键（可用时提供） */
+  key?: string;
+  /** 受影响行数 */
+  count?: number;
+}
+
+/** 订阅者回调（可返回 Promise，通知方会 await） */
+export type ChangeListener = (event: ChangeEvent) => void | Promise<void>;
+
+/**
+ * 变更通知引擎装饰器。
+ *
+ * 只读方法（find/count/findStream/getTableSchema/...）直接透传；
+ * 写入方法在成功后产生事件并同步派发给监听者。
+ */
+export class ChangeNotifierEngine implements IStorageEngine {
+  private readonly inner: IStorageEngine;
+  private readonly listeners = new Set<ChangeListener>();
+  private readonly onListenerError: (error: Error) => void;
+  /** 跨标签页广播回调（由 core 注入，避免本模块依赖 BroadcastChannel） */
+  private readonly onBroadcast: (table: string) => void;
+
+  constructor(
+    inner: IStorageEngine,
+    onListenerError: (error: Error) => void = () => {},
+    onBroadcast: (table: string) => void = () => {},
+  ) {
+    this.inner = inner;
+    this.onListenerError = onListenerError;
+    this.onBroadcast = onBroadcast;
+  }
+
+  /** 暴露内层引擎（需要能力探测或特化逻辑时使用） */
+  getInner(): IStorageEngine {
+    return this.inner;
+  }
+
+  /** 注册变更监听（返回退订函数） */
+  addListener(listener: ChangeListener): () => void {
+    this.listeners.add(listener);
+    return () => this.listeners.delete(listener);
+  }
+
+  /** 派发事件（供外部事件如跨标签页 `external` 复用同一通道） */
+  async dispatch(event: ChangeEvent): Promise<void> {
+    // 跨标签页广播与本地订阅走同一出口，保证"写入即通知"只有一条路径
+    try {
+      this.onBroadcast(event.table);
+    } catch {
+      // 广播失败不影响本地通知与写入结果
+    }
+    for (const listener of [...this.listeners]) {
+      try {
+        await listener(event);
+      } catch (error) {
+        // 订阅者异常不得影响写入结果（写入已成功），只上报
+        this.onListenerError(error as Error);
+      }
+    }
+  }
+
+  // -------------------------------------------------------------------------
+  //  写入路径：产生事件
+  // -------------------------------------------------------------------------
+
+  async insert(tableName: string, rows: Record<string, unknown>[]): Promise<string[]> {
+    const pks = await this.inner.insert(tableName, rows);
+    // 逐行事件；用引擎返回的主键对齐（resolved 行带 default 值，恢复时可对照）
+    for (let i = 0; i < pks.length; i++) {
+      await this.dispatch({
+        type: 'insert',
+        table: tableName,
+        key: pks[i],
+        row: i < rows.length ? rows[i] : undefined,
+        count: 1,
+      });
+    }
+    return pks;
+  }
+
+  async update(tableName: string, query: QueryPlan, updates: Record<string, unknown>): Promise<number> {
+    // 写入前快照将被影响的行（用于提供 row 内容）。失败不阻塞写入 —— 快照只是尽力而为。
+    const affected = await this.snapshotAffected(tableName, query);
+    const count = await this.inner.update(tableName, query, updates);
+    if (count > 0 && affected.length > 0) {
+      for (const before of affected) {
+        const after = { ...before, ...stripUndefined(updates) };
+        await this.dispatch({
+          type: 'update',
+          table: tableName,
+          row: after,
+          count: 1,
+        });
+      }
+    } else if (count > 0) {
+      await this.dispatch({ type: 'update', table: tableName, count });
+    }
+    return count;
+  }
+
+  async delete(tableName: string, query: QueryPlan): Promise<number> {
+    const affected = await this.snapshotAffected(tableName, query);
+    const count = await this.inner.delete(tableName, query);
+    if (count > 0 && affected.length > 0) {
+      for (const before of affected) {
+        await this.dispatch({ type: 'delete', table: tableName, row: before, count: 1 });
+      }
+    } else if (count > 0) {
+      await this.dispatch({ type: 'delete', table: tableName, count });
+    }
+    return count;
+  }
+
+  async clear(tableName: string): Promise<void> {
+    await this.inner.clear(tableName);
+    await this.dispatch({ type: 'clear', table: tableName });
+  }
+
+  async createTable(schema: import('../constants').TableSchema): Promise<void> {
+    await this.inner.createTable(schema);
+    await this.dispatch({ type: 'ddl', table: schema.name });
+  }
+
+  async dropTable(tableName: string): Promise<void> {
+    await this.inner.dropTable(tableName);
+    await this.dispatch({ type: 'ddl', table: tableName });
+  }
+
+  async alterTable(
+    tableName: string,
+    action: 'ADD' | 'DROP',
+    column: import('../constants').ColumnDef & { name: string },
+  ): Promise<void> {
+    await this.requireCapability('alterTable')(tableName, action, column);
+    await this.dispatch({ type: 'ddl', table: tableName });
+  }
+
+  async createIndex(tableName: string, column: string, unique?: boolean): Promise<void> {
+    await this.requireCapability('createIndex')(tableName, column, unique);
+    await this.dispatch({ type: 'ddl', table: tableName });
+  }
+
+  async dropIndex(tableName: string, column: string, indexName?: string): Promise<void> {
+    await this.requireCapability('dropIndex')(tableName, column, indexName);
+    await this.dispatch({ type: 'ddl', table: tableName });
+  }
+
+  // -------------------------------------------------------------------------
+  //  只读路径：透传
+  // -------------------------------------------------------------------------
+
+  get name(): string { return this.inner.name; }
+  open(dbName: string, version: number): Promise<void> { return this.inner.open(dbName, version); }
+  close(): Promise<void> { return this.inner.close(); }
+  isOpen(): boolean { return this.inner.isOpen(); }
+  hasTable(tableName: string): Promise<boolean> { return this.inner.hasTable(tableName); }
+  getTableNames(): Promise<string[]> { return this.inner.getTableNames(); }
+  getTableSchema(tableName: string): Promise<import('../constants').TableSchema | null> {
+    return this.inner.getTableSchema(tableName);
+  }
+  find(tableName: string, query: QueryPlan): Promise<Record<string, unknown>[]> {
+    return this.inner.find(tableName, query);
+  }
+  count(tableName: string, query?: QueryPlan): Promise<number> {
+    return this.inner.count(tableName, query);
+  }
+  async findStream(
+    tableName: string,
+    query: QueryPlan,
+    onRow: (row: Record<string, unknown>) => void,
+  ): Promise<number> {
+    return this.requireCapability('findStream')(tableName, query, onRow);
+  }
+
+  // ---- 事务：透传（事务内的写入由底层引擎统一处理，事件在语句层产生） ----
+
+  beginTransaction(): Promise<void> { return this.inner.beginTransaction(); }
+  commitTransaction(): Promise<void> { return this.inner.commitTransaction(); }
+  rollbackTransaction(): Promise<void> { return this.inner.rollbackTransaction(); }
+
+  savepoint(name: string): Promise<void> {
+    return this.requireCapability('savepoint')(name);
+  }
+  rollbackToSavepoint(name: string): Promise<void> {
+    return this.requireCapability('rollbackToSavepoint')(name);
+  }
+  releaseSavepoint(name: string): Promise<void> {
+    return this.requireCapability('releaseSavepoint')(name);
+  }
+  backup(): Promise<Record<string, Record<string, unknown>[]>> {
+    return this.requireCapability('backup')();
+  }
+  repair(): Promise<void> {
+    return this.requireCapability('repair')();
+  }
+  clearAll(): Promise<void> {
+    return this.requireCapability('clearAll')();
+  }
+  // ---- 维护能力（可选，AriaEngine 专有；只读/维护语义，不产生变更事件） ----
+
+  /**
+   * v0.8.0：转发引擎专有的维护方法。
+   *
+   * 为什么必须显式转发：executor 用 `typeof engine.analyzeTable === 'function'`
+   * 做能力探测。装饰器只实现 `IStorageEngine` 声明的成员，于是这些"接口外方法"
+   * 在被包装后会**静默消失** → ANALYZE/REINDEX 报 NOT_SUPPORTED（实测 3 个用例失败）。
+   * 这正是审计指出的"用 typeof/instanceof 做能力探测"的脆弱之处；
+   * 在装饰器里显式补齐是当前最直接的修法。
+   */
+  analyzeTable(tableName: string): Promise<Record<string, unknown>> {
+    return this.requireOptionalMethod<[string], Record<string, unknown>>('analyzeTable')(tableName);
+  }
+
+  reindexTable(tableName: string): Promise<number> {
+    return this.requireOptionalMethod<[string], number>('reindexTable')(tableName);
+  }
+
+  vacuum(): Promise<{ compactedLevels: number; gcVersions: number }> {
+    return this.requireOptionalMethod<[], { compactedLevels: number; gcVersions: number }>('vacuum')();
+  }
+
+  getMeta(key: string): Promise<string | null> {
+    const method = this.inner.getMeta;
+    if (typeof method !== 'function') return Promise.resolve(null);
+    return method.call(this.inner, key);
+  }
+  setMeta(key: string, value: string): Promise<void> {
+    const method = this.inner.setMeta;
+    if (typeof method !== 'function') return Promise.resolve();
+    return method.call(this.inner, key, value);
+  }
+
+  // -------------------------------------------------------------------------
+
+  /**
+   * v0.8.0: 取一个**可选能力**方法；内层未实现时抛 `NOT_SUPPORTED`。
+   *
+   * 为什么不能直接抛普通 Error：项目对"能力缺失"有明确契约 ——
+   * `NOT_SUPPORTED` 错误码（README「维护语句」、executor 的 savepoint/backup
+   * 分支、以及既有测试都依赖它）。装饰器如果抛原生 Error，调用方按 code 分类
+   * 就会失效（实测让 6 个套件里的 14 个用例失败）。
+   */
+  private requireCapability<K extends keyof IStorageEngine>(name: K): NonNullable<IStorageEngine[K]> {
+    const method = this.inner[name];
+    if (typeof method !== 'function') {
+      return (() => {
+        throw new DatabaseError(
+          `Engine "${this.inner.name}" does not support ${String(name)}`,
+          'NOT_SUPPORTED',
+        );
+      }) as unknown as NonNullable<IStorageEngine[K]>;
+    }
+    return (method as (...args: unknown[]) => unknown).bind(this.inner) as NonNullable<IStorageEngine[K]>;
+  }
+
+  /**
+   * v0.8.0: 取**接口之外**的可选方法（如 AriaEngine.analyzeTable / reindexTable / vacuum）。
+   *
+   * 与 requireCapability 的区别：这些方法不在 IStorageEngine 契约里，属于引擎专有能力，
+   * 但 executor 会用 `typeof engine.x === 'function'` 探测它们。装饰器必须显式转发，
+   * 否则能力在被包装后静默消失（实测 ANALYZE/REINDEX 报 NOT_SUPPORTED）。
+   */
+  private requireOptionalMethod<A extends unknown[], R>(name: string): (...args: A) => Promise<R> {
+    const method = (this.inner as unknown as Record<string, unknown>)[name];
+    if (typeof method !== 'function') {
+      return () => {
+        throw new DatabaseError(
+          `Engine "${this.inner.name}" does not support ${name}`,
+          'NOT_SUPPORTED',
+        );
+      };
+    }
+    return (method as (...args: A) => Promise<R>).bind(this.inner);
+  }
+
+  /**
+   * 尽力而为地取"将被影响的行"快照。
+   *
+   * 用途：让 UPDATE/DELETE 事件能携带行内容（文档承诺 `event.row`）。
+   * 失败时返回空数组 —— 事件退化为表级通知（只带 count），**不影响写入本身**。
+   */
+  private async snapshotAffected(
+    tableName: string,
+    query: QueryPlan,
+  ): Promise<Record<string, unknown>[]> {
+    try {
+      return await this.inner.find(tableName, { table: tableName, where: query.where });
+    } catch {
+      return [];
+    }
+  }
+}
+
+/** 与引擎/executor 一致的 undefined 语义：不更新该列 */
+function stripUndefined(updates: Record<string, unknown>): Record<string, unknown> {
+  const out: Record<string, unknown> = {};
+  for (const [k, v] of Object.entries(updates)) {
+    if (v !== undefined) out[k] = v;
+  }
+  return out;
+}
diff --git a/src/engine/memory.ts b/src/engine/memory.ts
index af87209..427c5b5 100644
--- a/src/engine/memory.ts
+++ b/src/engine/memory.ts
@@ -818,6 +818,14 @@ export class MemoryEngine implements IStorageEngine {
       // colIndex.get(null) 恒 undefined → return [] 短路全表扫描 → 索引列
       // IS NULL 恒空（对齐 AriaEngine v0.6.2 修复）
       if (targetValue === null || targetValue === undefined) continue;
+      // v0.8.0 根治（与 AriaEngine 同步）：**非原始值**（对象/数组）一律不走索引。
+      //
+      // 索引键只存原始值，因此 colIndex.get({...}) 恒 undefined → 下面 `return []`
+      // 会短路全表扫描 → 结果静默为空。真实触发场景正是"未解析的操作数"：
+      //   { $eq: { $col: 'y' } }        ← 列对列比较（t.x = t.y）
+      //   { $in: { $subquery: ... } }   ← 关联 IN 子查询
+      // 它们本该由 executor 逐行求值，却先在这里被索引路径吞成空集。
+      if (typeof targetValue === 'object') continue;
       const colIndex = tableIndexes.get(col);
       if (colIndex) {
         const pks = colIndex.get(targetValue);
diff --git a/src/query/executor.ts b/src/query/executor.ts
index 61529a4..e2a94e0 100644
--- a/src/query/executor.ts
+++ b/src/query/executor.ts
@@ -479,7 +479,7 @@ export class QueryExecutor {
         plan.columns = ['*'];
         if (orderByAlias) { plan.orderBy = undefined; plan.limit = undefined; plan.offset = undefined; }
         rows = await this.engine.find(plan.table, { ...plan, where: this.stripCorrelatedExists(stmt.where) });
-        rows = await this.filterCorrelated(rows, stmt.where);
+        rows = await this.filterCorrelated(rows, stmt.where, mainAliases);
       } else {
         // 先解析子查询
         if (stmt.where && Object.keys(stmt.where).length > 0) {
@@ -1470,14 +1470,21 @@ export class QueryExecutor {
   }
 
   /** 逐行绑定外层行上下文，求值关联 EXISTS、$col 引用与 CASE WHEN 键 */
-  private async filterCorrelated(rows: Record<string, unknown>[], where: WhereCondition): Promise<Record<string, unknown>[]> {
+  private async filterCorrelated(
+    rows: Record<string, unknown>[],
+    where: WhereCondition,
+    aliases: string[] = [],
+  ): Promise<Record<string, unknown>[]> {
     const result: Record<string, unknown>[] = [];
     for (const row of rows) {
       // 1. CASE WHEN 键 → 布尔条件（同步）
       let rowWhere = this.resolveCaseKeys(where, row);
       // 2. $col 绑定 + 关联 EXISTS 求值（异步）
-      rowWhere = await this.resolveSubqueries(rowWhere, row);
-      if (matchWhere(row, rowWhere)) {
+      rowWhere = await this.resolveSubqueries(rowWhere, row, aliases);
+      // v0.8.0（A10）：必须传 { $col: true } —— 否则 `{ $eq: { $col: 't.y' } }`
+      // 会被当作"与一个对象相等"比较，任何行都不匹配 → 静默空结果。
+      // 实测：`SELECT id FROM t WHERE t.x = t.y` 返回 []（应返回 x==y 的行）。
+      if (matchWhere(row, rowWhere, { $col: true })) {
         result.push(row);
       }
     }
@@ -1535,19 +1542,38 @@ export class QueryExecutor {
   }
 
   /** 将 where 中的 $col 引用替换为上下文行值 */
-  private bindColumnRefs(value: unknown, contextRow: Record<string, unknown>): unknown {
+  private bindColumnRefs(
+    value: unknown,
+    contextRow: Record<string, unknown>,
+    aliases: string[] = [],
+  ): unknown {
     if (typeof value !== 'object' || value === null || Array.isArray(value)) return value;
     const ops: Record<string, unknown> = {};
     for (const [op, operand] of Object.entries(value as Record<string, unknown>)) {
       if (op === '$col') {
-        ops[op] = contextRow[String(operand)] ?? null;
+        ops[op] = this.lookupOuterValue(contextRow, String(operand), aliases);
       } else if (op === '$and' || op === '$or') {
-        ops[op] = (operand as WhereCondition[]).map((sub) => this.bindWhereRefs(sub, contextRow));
+        ops[op] = (operand as WhereCondition[]).map((sub) => this.bindWhereRefs(sub, contextRow, aliases));
       } else if (op === '$not' && typeof operand === 'object' && operand !== null) {
-        ops[op] = this.bindColumnRefs(operand, contextRow);
+        ops[op] = this.bindColumnRefs(operand, contextRow, aliases);
       } else if (typeof operand === 'object' && operand !== null && !Array.isArray(operand) && '$col' in (operand as Record<string, unknown>)) {
         // 操作符值中嵌套的列引用：{ $eq: { $col: 'id' } } → { $eq: row['id'] }
-        ops[op] = contextRow[String((operand as Record<string, unknown>).$col)] ?? null;
+        ops[op] = this.lookupOuterValue(contextRow, String((operand as Record<string, unknown>).$col), aliases);
+      } else if (
+        typeof operand === 'object' && operand !== null && !Array.isArray(operand)
+        && '$subquery' in (operand as Record<string, unknown>)
+      ) {
+        // v0.8.0（A10）：**递归进入子查询**，把子查询 WHERE 里的外层引用绑定为当前行值。
+        //
+        // 此前这里落到 else 分支原样保留子查询对象 —— 于是
+        //   WHERE id IN (SELECT user_id FROM o WHERE o.user_id = u.id)
+        // 里的 `u.id` 始终未绑定（求值为 null），子查询返回空集 → `$in: []`
+        // → 所有行被过滤，**静默空结果**（而同结构的 EXISTS 走另一条分支是正常的）。
+        const sub = (operand as { $subquery: SelectStatement }).$subquery;
+        const subWhere = sub.where && this.hasCorrelatedRefs(sub.where)
+          ? this.bindWhereRefs(sub.where, contextRow, aliases)
+          : sub.where;
+        ops[op] = { $subquery: { ...sub, where: subWhere } };
       } else {
         ops[op] = operand;
       }
@@ -1555,17 +1581,40 @@ export class QueryExecutor {
     return ops;
   }
 
-  private bindWhereRefs(where: WhereCondition, contextRow: Record<string, unknown>): WhereCondition {
+  /**
+   * v0.8.0（A10）：在外层行里取 `$col` 引用的值。
+   *
+   * 关键点：引用可能带外层表名/别名前缀（`u.id`），而外层行键是**不带前缀**的
+   * （非 JOIN 路径会先剥离）。因此这里必须先剥前缀再取值 —— 否则 `u.id` 取到
+   * `undefined`，被 `?? null` 静默变成 null，子查询返回空集
+   * （实测 `WHERE id IN (SELECT user_id FROM o WHERE o.user_id = u.id)` 静默空结果）。
+   */
+  private lookupOuterValue(
+    contextRow: Record<string, unknown>,
+    ref: string,
+    aliases: string[],
+  ): unknown {
+    const bare = this.stripAlias(ref, aliases);
+    if (bare in contextRow) return contextRow[bare];
+    if (ref in contextRow) return contextRow[ref];
+    return null;
+  }
+
+  private bindWhereRefs(
+    where: WhereCondition,
+    contextRow: Record<string, unknown>,
+    aliases: string[] = [],
+  ): WhereCondition {
     const bound: WhereCondition = {};
     for (const [key, value] of Object.entries(where)) {
       if (key === '$and' || key === '$or') {
-        bound[key] = (value as WhereCondition[]).map((sub) => this.bindWhereRefs(sub, contextRow));
+        bound[key] = (value as WhereCondition[]).map((sub) => this.bindWhereRefs(sub, contextRow, aliases));
       } else if (key === '$not') {
-        bound.$not = this.bindWhereRefs(value as WhereCondition, contextRow);
+        bound.$not = this.bindWhereRefs(value as WhereCondition, contextRow, aliases);
       } else if (key === '$exists') {
         bound[key] = value;
       } else {
-        bound[key] = this.bindColumnRefs(value, contextRow);
+        bound[key] = this.bindColumnRefs(value, contextRow, aliases);
       }
     }
     return bound;
@@ -1580,10 +1629,14 @@ export class QueryExecutor {
    * 将结果替换为具体值。
    * @param contextRow 关联子查询的外层行上下文（用于绑定 $col 引用）
    */
-  private async resolveSubqueries(where: WhereCondition, contextRow?: Record<string, unknown>): Promise<WhereCondition> {
+  private async resolveSubqueries(
+    where: WhereCondition,
+    contextRow?: Record<string, unknown>,
+    aliases: string[] = [],
+  ): Promise<WhereCondition> {
     // 关联上下文：先把字段级的 $col 引用绑定为外层行值
     if (contextRow) {
-      where = this.bindWhereRefs(where, contextRow);
+      where = this.bindWhereRefs(where, contextRow, aliases);
     }
     const resolved: WhereCondition = {};
 
@@ -1605,24 +1658,24 @@ export class QueryExecutor {
       // 逻辑组合操作符
       if (key === '$and' && Array.isArray(value)) {
         resolved.$and = await Promise.all(
-          (value as WhereCondition[]).map((sub) => this.resolveSubqueries(sub, contextRow)),
+          (value as WhereCondition[]).map((sub) => this.resolveSubqueries(sub, contextRow, aliases)),
         );
         continue;
       }
       if (key === '$or' && Array.isArray(value)) {
         resolved.$or = await Promise.all(
-          (value as WhereCondition[]).map((sub) => this.resolveSubqueries(sub, contextRow)),
+          (value as WhereCondition[]).map((sub) => this.resolveSubqueries(sub, contextRow, aliases)),
         );
         continue;
       }
       if (key === '$not' && typeof value === 'object' && value !== null) {
-        resolved.$not = await this.resolveSubqueries(value as WhereCondition, contextRow);
+        resolved.$not = await this.resolveSubqueries(value as WhereCondition, contextRow, aliases);
         continue;
       }
 
       // 字段条件
       if (typeof value === 'object' && value !== null) {
-        resolved[key] = await this.resolveOperatorSubqueries(value as Record<string, unknown>);
+        resolved[key] = await this.resolveOperatorSubqueries(value as Record<string, unknown>, contextRow, aliases);
       } else {
         resolved[key] = value;
       }
@@ -1634,33 +1687,49 @@ export class QueryExecutor {
   /**
    * 解析操作符值中嵌套的子查询
    */
-  private async resolveOperatorSubqueries(ops: Record<string, unknown>): Promise<Record<string, unknown>> {
+  /**
+   * 解析字段条件里的子查询。
+   *
+   * v0.8.0 根治（A10）：接收外层行上下文并对子查询内的关联引用做绑定。
+   * 此前完全不传 contextRow —— 于是 `WHERE id IN (SELECT user_id FROM o WHERE o.user_id = u.id)`
+   * 里的 `u.id` 绑定为 undefined，子查询返回空集，最终 `$in: []` → **静默空结果**
+   * （而结构相同的 EXISTS 因为走另一条分支是正常的 —— 又一处"同类逻辑两条路径"）。
+   */
+  private async resolveOperatorSubqueries(
+    ops: Record<string, unknown>,
+    contextRow?: Record<string, unknown>,
+    aliases: string[] = [],
+  ): Promise<Record<string, unknown>> {
     const resolved: Record<string, unknown> = {};
 
     for (const [op, operand] of Object.entries(ops)) {
       // 处理嵌套 $and/$or（在字段级条件中）
       if (op === '$and' && Array.isArray(operand)) {
         resolved.$and = await Promise.all(
-          (operand as WhereCondition[]).map((sub) => this.resolveSubqueries(sub)),
+          (operand as WhereCondition[]).map((sub) => this.resolveSubqueries(sub, contextRow, aliases)),
         );
         continue;
       }
       if (op === '$or' && Array.isArray(operand)) {
         resolved.$or = await Promise.all(
-          (operand as WhereCondition[]).map((sub) => this.resolveSubqueries(sub)),
+          (operand as WhereCondition[]).map((sub) => this.resolveSubqueries(sub, contextRow, aliases)),
         );
         continue;
       }
       if (op === '$not') {
         resolved.$not = typeof operand === 'object' && operand !== null
-          ? await this.resolveOperatorSubqueries(operand as Record<string, unknown>)
+          ? await this.resolveOperatorSubqueries(operand as Record<string, unknown>, contextRow, aliases)
           : operand;
         continue;
       }
 
       // 子查询检测
       if (typeof operand === 'object' && operand !== null && '$subquery' in (operand as Record<string, unknown>)) {
-        const subStmt = (operand as Record<string, unknown>).$subquery as SelectStatement;
+        let subStmt = (operand as Record<string, unknown>).$subquery as SelectStatement;
+        // 关联引用绑定（与 $exists 分支同样的处理）
+        if (contextRow && this.hasCorrelatedRefs(subStmt.where)) {
+          subStmt = { ...subStmt, where: this.bindWhereRefs(subStmt.where, contextRow, aliases) };
+        }
         const subResult = await this.executeSelect(subStmt);
 
         if (op === '$in' || op === '$nin') {
diff --git a/src/query/where-matcher.ts b/src/query/where-matcher.ts
index d202899..d7e277f 100644
--- a/src/query/where-matcher.ts
+++ b/src/query/where-matcher.ts
@@ -140,6 +140,28 @@ function matchField(
     return value === row[ops.$col as string];
   }
 
+  // v0.8.0 根治：**未提供 $col 上下文**时，含列引用/未解析子查询的条件必须
+  // "放行"而不是"判假"。
+  //
+  // 背景：`t.x = t.y` 解析为 `{ x: { $eq: { $col: 'y' } } }`，执行路径是
+  //   engines.find(plan with 原始 where)  ← 引擎层 post-index matchWhere（无 $col 上下文）
+  //   → executor.filterCorrelated(rows, stmt.where, { $col: true })  ← 真正的判定
+  // 引擎层那一遍只是"索引命中后的安全过滤"（子集语义）。若它把 `$col` 当成普通对象
+  // 比较，任何行都不匹配 → 返回空集，executor 拿到空数组、逐行求值根本没机会执行
+  // → **静默空结果**（实测 `SELECT id FROM t WHERE t.x = t.y` 返回 []）。
+  //
+  // 语义上这是安全的：引擎层的过滤只允许"缩小候选集"，而这里选择不过滤；
+  // 最终判定始终由带 $col 上下文的 executor 完成（无法解析时由它抛 QUERY_ERROR）。
+  if (!options.$col) {
+    if ('$col' in ops && Object.keys(ops).length === 1) return true;
+    for (const [, operand] of Object.entries(ops)) {
+      if (typeof operand === 'object' && operand !== null && !Array.isArray(operand)) {
+        const inner = operand as Record<string, unknown>;
+        if ('$col' in inner || '$subquery' in inner) return true;
+      }
+    }
+  }
+
   // 遍历操作符
   for (const [op, operand] of Object.entries(ops)) {
     let actualOperand = operand;
diff --git a/tests/core-coverage.test.ts b/tests/core-coverage.test.ts
index e361a1f..7404ec8 100644
--- a/tests/core-coverage.test.ts
+++ b/tests/core-coverage.test.ts
@@ -49,19 +49,21 @@ describe('Core 全覆盖', () => {
       expect(typeof unsub).toBe('function');
     });
 
-    it('unsubscribe 取消监听', () => {
+    it('unsubscribe 取消监听', async () => {
       const fn = jest.fn();
       const unsub = db.subscribe('users', fn);
       unsub();
-      db.emit('users', { type: 'insert', row: { id: '3' } });
+      await db.emit('users', { type: 'insert', row: { id: '3' } });
       expect(fn).not.toHaveBeenCalled();
     });
 
-    it('emit 触发监听', () => {
+    it('emit 触发监听', async () => {
       const fn = jest.fn();
       db.subscribe('users', fn);
-      db.emit('users', { type: 'insert', row: { id: '3', name: 'Charlie' } });
-      expect(fn).toHaveBeenCalledWith({ type: 'insert', row: { id: '3', name: 'Charlie' } });
+      // v0.8.0: emit 现在会补全 table 字段（事件语义统一），且返回 Promise
+      await db.emit('users', { type: 'insert', row: { id: '3', name: 'Charlie' } });
+      // 事件现在带 table 字段（与引擎层变更事件语义一致）
+      expect(fn).toHaveBeenCalledWith({ table: 'users', type: 'insert', row: { id: '3', name: 'Charlie' } });
     });
   });
 
diff --git a/tests/sql-ext3.test.ts b/tests/sql-ext3.test.ts
index 20787fb..be52bb7 100644
--- a/tests/sql-ext3.test.ts
+++ b/tests/sql-ext3.test.ts
@@ -259,7 +259,7 @@ describe('[v0.3.2] 多标签页同步', () => {
     await dbB.init();
 
     const events: { type: string; table?: string }[] = [];
-    dbB.subscribe('t', (e) => events.push(e));
+    dbB.subscribe('t', (e) => { events.push(e); });
 
     await dbA.query(`INSERT INTO t VALUES ('1', 10)`);
 
@@ -315,7 +315,7 @@ describe('[v0.3.2] 多标签页同步', () => {
     await dbB.init();
 
     const events: unknown[] = [];
-    dbB.subscribe('t', (e) => events.push(e));
+    dbB.subscribe('t', (e) => { events.push(e); });
 
     await dbA.query(`INSERT INTO t VALUES ('1', 10)`);
     await new Promise((r) => setTimeout(r, 30));
@@ -337,7 +337,7 @@ describe('[v0.3.2] 多标签页同步', () => {
     await dbB.init();
 
     const events: unknown[] = [];
-    dbB.subscribe('t', (e) => events.push(e));
+    dbB.subscribe('t', (e) => { events.push(e); });
 
     await dbA.table('t').insert({ id: '1', v: 10 });
     expect(events.length).toBeGreaterThan(0);
diff --git a/tests/v080-correlated.test.ts b/tests/v080-correlated.test.ts
new file mode 100644
index 0000000..5a62323
--- /dev/null
+++ b/tests/v080-correlated.test.ts
@@ -0,0 +1,89 @@
+/**
+ * v0.8.0 回归 —— A10 列对列比较与关联子查询
+ *
+ * 两个此前**静默返回空结果**的缺陷（行数正确、内容全空、无任何报错）：
+ *
+ * 1. `WHERE t.x = t.y`（唯一可解析的列对列写法，解析为 `{ 't.x': { $eq: { $col: 't.y' } } }`）
+ *    执行路径是：引擎层先按原始 where 取候选行（引擎的 matchWhere **没有** $col 上下文）
+ *    → executor.filterCorrelated 带 `{ $col: true }` 做最终判定。
+ *    结果两处都错了：
+ *      - 引擎层把 `{ $col: ... }` 当普通对象比较 → 所有行都不匹配；
+ *      - executor 侧调用 matchWhere 时又**没传** `{ $col: true }`。
+ *    实测返回 `[]`（应返回 x == y 的行）。
+ *
+ * 2. `WHERE id IN (SELECT ... WHERE o.user_id = u.id)`（关联 IN 子查询）
+ *    子查询执行时**不传外层行上下文**，`u.id` 绑定为 null，子查询返回空集 →
+ *    `$in: []` → 静默空结果。而结构相同的 EXISTS 走另一条分支、结果是正确的
+ *    —— 又一处"同一语义两条路径"。
+ *
+ * 修复后两者与等价写法结果一致（见下方断言）。
+ */
+import { MetonaSqlark } from '../src/core';
+
+const MODES: Array<[string, Record<string, unknown>]> = [
+  ['memory', {}],
+  ['disk', {}],
+  ['hybrid', {}],
+  ['aria', { diskEngine: 'memory' }],
+];
+
+describe('[v0.8.0] A10 列对列比较', () => {
+  test('WHERE t.x = t.y 返回相等行（四引擎）', async () => {
+    for (const [mode, extra] of MODES) {
+      const db = await MetonaSqlark.create({ name: `v080-a10-${mode}`, mode, ...extra } as never);
+      await db.defineTable('t', {
+        id: { type: 'string', primaryKey: true },
+        x: { type: 'number' },
+        y: { type: 'number' },
+      });
+      await db.query("INSERT INTO t VALUES ('1',1,5),('2',2,3),('3',7,7),('4',7,7)");
+
+      const rows = await db.query('SELECT id FROM t WHERE t.x = t.y') as Array<{ id: string }>;
+      expect(rows.map((r) => r.id).sort()).toEqual(['3', '4']);
+      await db.close();
+    }
+  });
+
+  test('裸 x = y 仍为显式 PARSE_ERROR（语法限制，不静默）', async () => {
+    const db = await MetonaSqlark.create({ name: 'v080-a10-bare', mode: 'memory' });
+    await db.defineTable('t', { id: { type: 'string', primaryKey: true }, x: { type: 'number' }, y: { type: 'number' } });
+    await expect(db.query('SELECT id FROM t WHERE x = y')).rejects.toMatchObject({ code: 'PARSE_ERROR' });
+    await db.close();
+  });
+});
+
+describe('[v0.8.0] A10 关联 IN 子查询', () => {
+  test('关联 IN 与等价 EXISTS 结果一致（四引擎）', async () => {
+    for (const [mode, extra] of MODES) {
+      const db = await MetonaSqlark.create({ name: `v080-a10b-${mode}`, mode, ...extra } as never);
+      await db.defineTable('u', { id: { type: 'string', primaryKey: true } });
+      await db.defineTable('o', { id: { type: 'string', primaryKey: true }, user_id: { type: 'string' } });
+      await db.query("INSERT INTO u VALUES ('1'),('2'),('3')");
+      await db.query("INSERT INTO o VALUES ('o1','1'),('o2','2')");
+
+      const viaIn = await db.query(
+        'SELECT id FROM u WHERE id IN (SELECT user_id FROM o WHERE o.user_id = u.id)',
+      ) as Array<{ id: string }>;
+      const viaExists = await db.query(
+        'SELECT id FROM u WHERE EXISTS (SELECT 1 FROM o WHERE o.user_id = u.id)',
+      ) as Array<{ id: string }>;
+
+      expect(viaIn.map((r) => r.id).sort()).toEqual(['1', '2']);
+      // 关键护栏：两条语义等价的路径必须给出一致结果
+      expect(viaIn.map((r) => r.id).sort()).toEqual(viaExists.map((r) => r.id).sort());
+      await db.close();
+    }
+  });
+
+  test('非关联 IN 子查询不受影响（回归护栏）', async () => {
+    const db = await MetonaSqlark.create({ name: 'v080-a10c', mode: 'memory' });
+    await db.defineTable('u', { id: { type: 'string', primaryKey: true } });
+    await db.defineTable('o', { id: { type: 'string', primaryKey: true }, user_id: { type: 'string' } });
+    await db.query("INSERT INTO u VALUES ('1'),('2'),('3')");
+    await db.query("INSERT INTO o VALUES ('o1','1'),('o2','2')");
+
+    const rows = await db.query('SELECT id FROM u WHERE id IN (SELECT user_id FROM o)') as Array<{ id: string }>;
+    expect(rows.map((r) => r.id).sort()).toEqual(['1', '2']);
+    await db.close();
+  });
+});
diff --git a/tests/v080-subscribe.test.ts b/tests/v080-subscribe.test.ts
new file mode 100644
index 0000000..5e2829a
--- /dev/null
+++ b/tests/v080-subscribe.test.ts
@@ -0,0 +1,119 @@
+/**
+ * v0.8.0 回归 —— A9 发布订阅（本地写入必须触发订阅者）
+ *
+ * 修复前：`db.subscribe(table, fn)` 对**本地写入永不触发** —— 全库唯一调用 `emit`
+ * 的地方在 BroadcastChannel 收到其它标签页消息的分支里。于是 README:232
+ * 「订阅表变更」与 site/docs.html:667-679 的示例（`event.type: 'insert' | 'update' |
+ * 'delete'`、`event.row`）全部不成立；唯一实际触发条件是 `multiTabSync: true`
+ * 且收到 `external` 事件。
+ *
+ * 修复方式：用 `ChangeNotifierEngine` 装饰器把变更通知收敛到**引擎接口**这一个位置。
+ * 三个写入入口（SQL 语句 / Table API / QueryBuilder）与事务内写入都必须经过引擎，
+ * 因此拦一次即可全覆盖，避免在三条路径上各写一份变更描述逻辑（那正是本项目
+ * 反复出现"同一语义多份实现 → 漂移"的模式）。
+ *
+ * 覆盖：四个引擎 × 三种写入入口 × 事件内容（type/table/row/key/count）+
+ * 退订 + async 订阅者 + 跨标签页广播不形成回路。
+ */
+import { MetonaSqlark } from '../src/core';
+import type { ChangeEvent } from '../src/engine/change-notifier';
+
+const MODES: Array<[string, Record<string, unknown>]> = [
+  ['memory', {}],
+  ['disk', {}],
+  ['hybrid', {}],
+  ['aria', { diskEngine: 'memory' }],
+];
+
+describe('[v0.8.0] A9 发布订阅 —— SQL 写入路径', () => {
+  test('INSERT/UPDATE/DELETE 逐行事件，携带行内容与主键', async () => {
+    for (const [mode, extra] of MODES) {
+      const db = await MetonaSqlark.create({ name: `v080-a9-${mode}`, mode, ...extra } as never);
+      await db.defineTable('t', { id: { type: 'string', primaryKey: true }, v: { type: 'number' } });
+      const events: ChangeEvent[] = [];
+      const unsub = db.subscribe('t', (e) => { events.push(e); });
+
+      await db.query("INSERT INTO t VALUES ('a',1),('b',2)");
+      await db.query("UPDATE t SET v = 9 WHERE id = 'a'");
+      await db.query("DELETE FROM t WHERE id = 'b'");
+
+      expect(events.map((e) => e.type)).toEqual(['insert', 'insert', 'update', 'delete']);
+      // 逐行事件带主键与行内容
+      expect(events.filter((e) => e.type === 'insert').map((e) => e.key).sort()).toEqual(['a', 'b']);
+      const update = events.find((e) => e.type === 'update')!;
+      expect(update.row).toMatchObject({ id: 'a', v: 9 });
+      expect(update.table).toBe('t');
+      const del = events.find((e) => e.type === 'delete')!;
+      expect(del.row).toMatchObject({ id: 'b', v: 2 });
+
+      // 退订后不再收到
+      unsub();
+      await db.query("INSERT INTO t VALUES ('c',3)");
+      expect(events).toHaveLength(4);
+
+      await db.close();
+    }
+  });
+});
+
+describe('[v0.8.0] A9 发布订阅 —— Table API / QueryBuilder 路径', () => {
+  test('insert / insertMany / update / delete / clear 全部产生事件', async () => {
+    for (const [mode, extra] of MODES) {
+      const db = await MetonaSqlark.create({ name: `v080-a9b-${mode}`, mode, ...extra } as never);
+      await db.defineTable('t', { id: { type: 'string', primaryKey: true }, v: { type: 'number' } });
+      const events: ChangeEvent[] = [];
+      db.subscribe('t', (e) => { events.push(e); });
+
+      await db.table('t').insert({ id: 'a', v: 1 });
+      await db.table('t').insertMany([{ id: 'b', v: 2 }, { id: 'c', v: 3 }]);
+      await db.table('t').update({ v: 9 }).where({ id: 'a' }).execute();
+      await db.table('t').delete().where({ id: 'b' }).execute();
+      await db.table('t').clear();
+
+      expect(events.map((e) => e.type))
+        .toEqual(['insert', 'insert', 'insert', 'update', 'delete', 'clear']);
+      await db.close();
+    }
+  });
+});
+
+describe('[v0.8.0] A9 发布订阅 —— 语义护栏', () => {
+  test('异步订阅者会被 await（写完即通知完）', async () => {
+    const db = await MetonaSqlark.create({ name: 'v080-a9-async', mode: 'memory' });
+    await db.defineTable('t', { id: { type: 'string', primaryKey: true } });
+    const seen: string[] = [];
+    db.subscribe('t', async (e) => {
+      await new Promise((r) => setTimeout(r, 1));
+      seen.push(`${e.type}:${e.key ?? ''}`);
+    });
+    await db.query("INSERT INTO t VALUES ('x')");
+    expect(seen).toEqual(['insert:x']);
+    await db.close();
+  });
+
+  test('订阅者抛错不影响写入结果（写入已成功）', async () => {
+    const errors: Error[] = [];
+    const db = await MetonaSqlark.create({
+      name: 'v080-a9-throw',
+      mode: 'memory',
+      onError: (e: Error) => { errors.push(e); },
+    } as never);
+    await db.defineTable('t', { id: { type: 'string', primaryKey: true } });
+    db.subscribe('t', () => { throw new Error('subscriber boom'); });
+
+    await expect(db.query("INSERT INTO t VALUES ('x')")).resolves.toEqual(['x']);
+    const rows = await db.query('SELECT COUNT(*) AS c FROM t') as Array<{ c: number }>;
+    expect(rows[0].c).toBe(1);
+    expect(errors.some((e) => e.message === 'subscriber boom')).toBe(true);
+    await db.close();
+  });
+
+  test('手动 emit 仍可用，并补全 table 字段', async () => {
+    const db = await MetonaSqlark.create({ name: 'v080-a9-emit', mode: 'memory' });
+    const events: ChangeEvent[] = [];
+    db.subscribe('t', (e) => { events.push(e); });
+    await db.emit('t', { type: 'insert', row: { id: 'z' } });
+    expect(events).toEqual([{ table: 't', type: 'insert', row: { id: 'z' } }]);
+    await db.close();
+  });
+});