feat(plugins): fire PERSISTED_DATA_CHANGED hook on persisted-data changes (#7805)

* feat(plugins): fire PERSISTED_DATA_CHANGED hook on persisted-data changes

Wires the dead PluginHooks.PERSISTED_DATA_UPDATE enum into a fired hook,
renamed PERSISTED_DATA_CHANGED. Plugins are notified when their persisted
data changes for any reason after the host's initial boot load — local
writes, remote incremental sync via bulkApplyOperations, and post-boot
wholesale loadAllData paths (SYNC_IMPORT / BACKUP_IMPORT / validation
repair / recovery).

Selector-based effect on selectPluginUserDataFeatureState, gated on
SyncTriggerService.afterInitialSyncDoneAndDataLoadedInitially$ so the
boot-time state seeds the pairwise baseline. Differ compares prev/next
by === on the encoded data blob (never decoded). Stage A composite
entityIds (pluginId:key) are normalized to owner pluginId and deduped
so a plugin with N keyed entries changing in one emission fires exactly
once. Per-pluginId dispatch via new PluginHooksService.dispatchHookToPlugin.

Effect is { dispatch: false } and creates no ops, so no sync-window
guard is needed — and adding one (skipDuringSyncWindow) would silently
suppress the very remote-sync deliveries the hook is designed to catch.

Closes #7754. Follow-up: doc-mode adoption tracked in #7752.

* test(plugins): tighten PERSISTED_DATA_CHANGED spec + docs

Multi-review pass 2 surfaced that the 5s timeout spec asserted only that
the dispatcher resolved — it would have silently passed if the timeout
race were removed entirely. Spy on PluginLog.err and assert the timeout
message actually reached the catch branch.

Also:
- Add `:` guard to PluginHooksService.registerHookHandler so the
  persistence-key grammar is enforced at both the persistence and
  hooks-registry endpoints (defense-in-depth; composeId already throws
  at the bridge).
- Add async-rejected-promise spec to cover the Promise.race branch that
  sync-throw didn't exercise.
- Carry the PERSISTED_DATA_CHANGED contract paragraph into
  docs/plugin-development.md and docs/wiki/3.01-API.md (previously only
  in packages/plugin-api/README.md).
- Clarify loadSyncedData(key?) in the README for keyed plugins.

docs/plugin-development.md

@@ -463,9 +463,21 @@ const hooks = {
   CURRENT_TASK_CHANGE: 'currentTaskChange',
   FINISH_DAY: 'finishDay',
   LANGUAGE_CHANGE: 'languageChange',
-  PERSISTED_DATA_UPDATE: 'persistedDataUpdate',
+  PERSISTED_DATA_CHANGED: 'persistedDataChanged',
   ACTION: 'action',
 };
+```
+
+`PERSISTED_DATA_CHANGED` fires whenever this plugin's persisted data
+changes — local writes, remote sync deliveries, and bulk imports —
+*after* the host has finished its initial boot load. The handler
+receives no payload; re-call `loadSyncedData(key?)` for any key your
+plugin tracks to get fresh data. There is no replay-on-register and no
+guaranteed ordering across rapid changes, so handlers must be
+idempotent. The typical pattern is: call `loadSyncedData()` once on
+plugin init, then subscribe to this hook for subsequent updates.
+
+```javascript
 
 // Register hook listener
 PluginAPI.registerHook(PluginAPI.Hooks.TASK_COMPLETE, (taskId) => {

docs/wiki/3.01-API.md

@@ -168,7 +168,9 @@ The Plugin API is exposed to plugins via a global `PluginAPI` object. Plugins ru
 
 ### Hooks (Events)
 
-Plugins can register handlers for: `taskCreated`, `taskComplete`, `taskUpdate`, `taskDelete`, `currentTaskChange`, `finishDay`, `languageChange`, `persistedDataUpdate`, `action`, `anyTaskUpdate`, `projectListUpdate`. Payload types are defined in `plugin-api` types (`HookPayloadMap`).
+Plugins can register handlers for: `taskCreated`, `taskComplete`, `taskUpdate`, `taskDelete`, `currentTaskChange`, `finishDay`, `languageChange`, `persistedDataChanged`, `action`, `anyTaskUpdate`, `projectListUpdate`. Payload types are defined in `plugin-api` types (`HookPayloadMap`).
+
+`persistedDataChanged` fires after the initial boot load on any change to this plugin's persisted data (local write, remote sync, bulk import). The payload is `void`; re-call `loadSyncedData(key?)` for fresh data. No replay-on-register, no ordering guarantee across rapid changes — handlers must be idempotent.
 
 ### Plugin Data Types
 

packages/plugin-api/README.md

@@ -103,11 +103,21 @@ enum PluginHooks {
   TASK_DELETE = 'taskDelete',
   FINISH_DAY = 'finishDay',
   LANGUAGE_CHANGE = 'languageChange',
-  PERSISTED_DATA_UPDATE = 'persistedDataUpdate',
+  PERSISTED_DATA_CHANGED = 'persistedDataChanged',
   ACTION = 'action',
 }
 ```
 
+**`PERSISTED_DATA_CHANGED`** fires on any persistent-data change to this
+plugin after the host has finished its initial boot load — including
+remote sync deliveries and bulk imports. Handler receives no payload;
+re-call `loadSyncedData(key?)` for any key your plugin tracks to get
+fresh data (scoped to the calling plugin). Contract: call
+`loadSyncedData()` on plugin init for the initial state; then use this
+hook for subsequent changes. There is no replay-on-register, no
+per-key discrimination in the event, and no guaranteed ordering across
+rapid changes. Handlers must be idempotent.
+
 ### 2. Required Permissions
 
 Add these to your manifest.json based on what your plugin needs:

packages/plugin-api/src/types.ts

@@ -21,7 +21,7 @@ export enum PluginHooks {
   CURRENT_TASK_CHANGE = 'currentTaskChange',
   FINISH_DAY = 'finishDay',
   LANGUAGE_CHANGE = 'languageChange',
-  PERSISTED_DATA_UPDATE = 'persistedDataUpdate',
+  PERSISTED_DATA_CHANGED = 'persistedDataChanged',
   ACTION = 'action',
   ANY_TASK_UPDATE = 'anyTaskUpdate',
   PROJECT_LIST_UPDATE = 'projectListUpdate',
@@ -166,10 +166,6 @@ export interface LanguageChangePayload {
   [key: string]: unknown;
 }
 
-export interface PersistedDataUpdatePayload {
-  data: string;
-}
-
 export interface ActionPayload {
   action: string;
   payload?: unknown;
@@ -233,7 +229,7 @@ export interface HookPayloadMap {
   [PluginHooks.CURRENT_TASK_CHANGE]: CurrentTaskChangePayload;
   [PluginHooks.FINISH_DAY]: FinishDayPayload;
   [PluginHooks.LANGUAGE_CHANGE]: LanguageChangePayload;
-  [PluginHooks.PERSISTED_DATA_UPDATE]: PersistedDataUpdatePayload;
+  [PluginHooks.PERSISTED_DATA_CHANGED]: void;
   [PluginHooks.ACTION]: ActionPayload;
   [PluginHooks.ANY_TASK_UPDATE]: AnyTaskUpdatePayload;
   [PluginHooks.PROJECT_LIST_UPDATE]: ProjectListUpdatePayload;

src/app/plugins/plugin-hooks.effects.spec.ts

@@ -1,6 +1,6 @@
 import { TestBed } from '@angular/core/testing';
 import { provideMockActions } from '@ngrx/effects/testing';
-import { EMPTY, Observable, of } from 'rxjs';
+import { EMPTY, Observable, of, ReplaySubject } from 'rxjs';
 import { PluginHooksEffects } from './plugin-hooks.effects';
 import { WorkContextService } from '../features/work-context/work-context.service';
 import { provideMockStore, MockStore } from '@ngrx/store/testing';
@@ -13,12 +13,17 @@ import {
   selectCurrentTask,
   selectTaskById,
 } from '../features/tasks/store/task.selectors';
+import { selectPluginUserDataFeatureState } from './store/plugin-user-data.reducer';
+import { SyncTriggerService } from '../imex/sync/sync-trigger.service';
+import { HydrationStateService } from '../op-log/apply/hydration-state.service';
+import { PluginUserData } from './plugin-persistence.model';
 
 describe('PluginHooksEffects', () => {
   let effects: PluginHooksEffects;
   let actions$: Observable<any>;
   let pluginServiceMock: jasmine.SpyObj<PluginService>;
   let store: MockStore;
+  let gateSubject: ReplaySubject<boolean>;
 
   const createMockTask = (overrides: Partial<TaskWithSubTasks> = {}): TaskWithSubTasks =>
     ({
@@ -55,7 +60,16 @@ describe('PluginHooksEffects', () => {
 
   beforeEach(() => {
     mockTask = createMockTask();
-    pluginServiceMock = jasmine.createSpyObj('PluginService', ['dispatchHook']);
+    pluginServiceMock = jasmine.createSpyObj('PluginService', [
+      'dispatchHook',
+      'dispatchHookToPlugin',
+    ]);
+
+    // MUST be a ReplaySubject(1) / Subject — NOT BehaviorSubject(true) or
+    // of(true). The boot-suppression spec depends on the gate being un-emitted
+    // at boot `loadAllData` time; a BehaviorSubject(true) would emit on
+    // subscribe and make that spec pass for the wrong reason.
+    gateSubject = new ReplaySubject<boolean>(1);
 
     TestBed.configureTestingModule({
       providers: [
@@ -76,6 +90,26 @@ describe('PluginHooksEffects', () => {
         // workContextChange$ reads activeWorkContext$ at construction; an
         // empty stream keeps that effect inert for the other effects' tests.
         { provide: WorkContextService, useValue: { activeWorkContext$: EMPTY } },
+        {
+          provide: SyncTriggerService,
+          useValue: {
+            afterInitialSyncDoneAndDataLoadedInitially$: gateSubject.asObservable(),
+          },
+        },
+        // Provided with isInSyncWindow=true so a future regression that adds
+        // `skipDuringSyncWindow`/`waitForSyncWindow` to the effect would
+        // silently drop/defer emissions and fail the firePersistedDataChanged$
+        // specs. See plan §"Boot gate" — the hook must fire during the sync
+        // window because remote-sync deliveries are exactly the motivating
+        // case.
+        {
+          provide: HydrationStateService,
+          useValue: {
+            isInSyncWindow: () => true,
+            isInSyncWindow$: of(true),
+            isApplyingRemoteOps: () => true,
+          },
+        },
       ],
     });
 
@@ -345,4 +379,169 @@ describe('PluginHooksEffects', () => {
       }, 0);
     });
   });
+
+  describe('firePersistedDataChanged$', () => {
+    const entry = (id: string, data: string): PluginUserData => ({ id, data });
+
+    const setPluginData = (data: PluginUserData[]): void => {
+      store.overrideSelector(selectPluginUserDataFeatureState, data);
+      store.refreshState();
+    };
+
+    it('does not fire while the gate has not emitted (boot suppression)', (done) => {
+      actions$ = of();
+      store.overrideSelector(selectPluginUserDataFeatureState, [entry('a', 'gz:1')]);
+
+      const sub = effects.firePersistedDataChanged$.subscribe();
+      // Drive a "loadAllData" boot dispatch by swapping the selector before the
+      // gate fires. Effect must NOT emit because the gate is still un-emitted.
+      setPluginData([entry('a', 'gz:2')]);
+
+      setTimeout(() => {
+        expect(pluginServiceMock.dispatchHookToPlugin).not.toHaveBeenCalled();
+
+        // Now open the gate. The current state becomes the pairwise baseline;
+        // still no fire because pairwise needs a second emission.
+        gateSubject.next(true);
+
+        setTimeout(() => {
+          expect(pluginServiceMock.dispatchHookToPlugin).not.toHaveBeenCalled();
+          sub.unsubscribe();
+          done();
+        }, 0);
+      }, 0);
+    });
+
+    it('fires once per changed pluginId on local writes after the gate opens', (done) => {
+      actions$ = of();
+      store.overrideSelector(selectPluginUserDataFeatureState, [entry('a', 'gz:1')]);
+
+      const sub = effects.firePersistedDataChanged$.subscribe();
+      gateSubject.next(true);
+
+      // Baseline established; a second emission with changed data fires.
+      setPluginData([entry('a', 'gz:2')]);
+
+      setTimeout(() => {
+        expect(pluginServiceMock.dispatchHookToPlugin).toHaveBeenCalledTimes(1);
+        expect(pluginServiceMock.dispatchHookToPlugin).toHaveBeenCalledWith(
+          'a',
+          PluginHooks.PERSISTED_DATA_CHANGED,
+        );
+        sub.unsubscribe();
+        done();
+      }, 0);
+    });
+
+    it('fires the diff after a post-boot wholesale load (SYNC_IMPORT / BACKUP_IMPORT / recovery)', (done) => {
+      // Same code path as #post-boot loadAllData — selector emits a new array,
+      // differ reports only the entries whose data actually changed.
+      actions$ = of();
+      store.overrideSelector(selectPluginUserDataFeatureState, [
+        entry('keep', 'gz:s'),
+        entry('change', 'gz:old'),
+        entry('removed', 'gz:gone'),
+      ]);
+
+      const sub = effects.firePersistedDataChanged$.subscribe();
+      gateSubject.next(true);
+
+      setPluginData([
+        entry('keep', 'gz:s'), // unchanged → no fire
+        entry('change', 'gz:new'), // updated → fire
+        entry('added', 'gz:fresh'), // added → fire
+        // 'removed' missing → fire
+      ]);
+
+      setTimeout(() => {
+        const dispatched = pluginServiceMock.dispatchHookToPlugin.calls
+          .allArgs()
+          .map(([pluginId]) => pluginId)
+          .sort();
+        expect(dispatched).toEqual(['added', 'change', 'removed']);
+        sub.unsubscribe();
+        done();
+      }, 0);
+    });
+
+    it('isolates handlers per plugin — a change in A does not fire B', (done) => {
+      actions$ = of();
+      store.overrideSelector(selectPluginUserDataFeatureState, [
+        entry('a', 'gz:1'),
+        entry('b', 'gz:1'),
+      ]);
+
+      const sub = effects.firePersistedDataChanged$.subscribe();
+      gateSubject.next(true);
+
+      setPluginData([entry('a', 'gz:2'), entry('b', 'gz:1')]);
+
+      setTimeout(() => {
+        expect(pluginServiceMock.dispatchHookToPlugin).toHaveBeenCalledTimes(1);
+        expect(pluginServiceMock.dispatchHookToPlugin).toHaveBeenCalledWith(
+          'a',
+          PluginHooks.PERSISTED_DATA_CHANGED,
+        );
+        expect(pluginServiceMock.dispatchHookToPlugin).not.toHaveBeenCalledWith(
+          'b',
+          jasmine.anything(),
+        );
+        sub.unsubscribe();
+        done();
+      }, 0);
+    });
+
+    it('fires on delete', (done) => {
+      actions$ = of();
+      store.overrideSelector(selectPluginUserDataFeatureState, [
+        entry('a', 'gz:1'),
+        entry('b', 'gz:1'),
+      ]);
+
+      const sub = effects.firePersistedDataChanged$.subscribe();
+      gateSubject.next(true);
+
+      setPluginData([entry('a', 'gz:1')]);
+
+      setTimeout(() => {
+        expect(pluginServiceMock.dispatchHookToPlugin).toHaveBeenCalledTimes(1);
+        expect(pluginServiceMock.dispatchHookToPlugin).toHaveBeenCalledWith(
+          'b',
+          PluginHooks.PERSISTED_DATA_CHANGED,
+        );
+        sub.unsubscribe();
+        done();
+      }, 0);
+    });
+
+    it('collapses keyed entityIds to the owner pluginId and fires once', (done) => {
+      // Stage A keyed storage: a plugin with multiple `pluginId:key` entries
+      // changing in one emission must fire its handler (registered under the
+      // bare pluginId) exactly once.
+      actions$ = of();
+      store.overrideSelector(selectPluginUserDataFeatureState, [
+        entry('foo:doc-1', 'gz:1'),
+        entry('foo:doc-2', 'gz:1'),
+      ]);
+
+      const sub = effects.firePersistedDataChanged$.subscribe();
+      gateSubject.next(true);
+
+      setPluginData([
+        entry('foo:doc-1', 'gz:2'), // updated
+        entry('foo:doc-2', 'gz:2'), // updated
+        entry('foo:doc-3', 'gz:new'), // added
+      ]);
+
+      setTimeout(() => {
+        expect(pluginServiceMock.dispatchHookToPlugin).toHaveBeenCalledTimes(1);
+        expect(pluginServiceMock.dispatchHookToPlugin).toHaveBeenCalledWith(
+          'foo',
+          PluginHooks.PERSISTED_DATA_CHANGED,
+        );
+        sub.unsubscribe();
+        done();
+      }, 0);
+    });
+  });
 });

src/app/plugins/plugin-hooks.effects.ts

@@ -49,6 +49,9 @@ import { PlannerActions } from '../features/planner/store/planner.actions';
 import { LanguageCode } from '../core/locale.constants';
 import { WorkContextService } from '../features/work-context/work-context.service';
 import { toActiveWorkContext } from './util/active-work-context.util';
+import { SyncTriggerService } from '../imex/sync/sync-trigger.service';
+import { selectPluginUserDataFeatureState } from './store/plugin-user-data.reducer';
+import { diffChangedPluginIds } from './util/plugin-data-diff.util';
 
 @Injectable()
 export class PluginHooksEffects {
@@ -57,6 +60,7 @@ export class PluginHooksEffects {
   private readonly pluginService = inject(PluginService);
   private readonly pluginI18nService = inject(PluginI18nService);
   private readonly workContextService = inject(WorkContextService);
+  private readonly syncTrigger = inject(SyncTriggerService);
 
   taskComplete$ = createEffect(
     () =>
@@ -363,6 +367,45 @@ export class PluginHooksEffects {
     { dispatch: false },
   );
 
+  // Selector-based (not action-based) because remote `PLUGIN_USER_DATA`
+  // upserts arrive through `bulkApplyOperations` — an `ofType` filter on the
+  // local action wouldn't see them. The feature-state subscription catches
+  // local writes, remote incremental sync, and post-boot `loadAllData` paths
+  // (SYNC_IMPORT / BACKUP_IMPORT / validation repair / recovery) alike.
+  //
+  // Gated on `afterInitialSyncDoneAndDataLoadedInitially$` so the boot-time
+  // selector emission seeds `pairwise` as the baseline rather than producing
+  // a per-plugin flood at startup. House pattern, cf. `task-due.effects.ts`.
+  //
+  // No inner `waitForSyncWindow` / `skipDuringSyncWindow`: the effect is
+  // `{ dispatch: false }` and creates no ops, so sync rule 2 does not apply.
+  // Critically, `skipDuringSyncWindow` would suppress emissions during
+  // `_isApplyingRemoteOps` — exactly the remote-sync delivery this hook
+  // exists to fire on.
+  firePersistedDataChanged$ = createEffect(
+    () =>
+      this.syncTrigger.afterInitialSyncDoneAndDataLoadedInitially$.pipe(
+        filter((done) => done),
+        switchMap(() =>
+          this.store.pipe(
+            select(selectPluginUserDataFeatureState),
+            pairwise(),
+            map(([prev, next]) => diffChangedPluginIds(prev, next)),
+            filter((ids) => ids.length > 0),
+            tap((ids) => {
+              for (const pluginId of ids) {
+                this.pluginService.dispatchHookToPlugin(
+                  pluginId,
+                  PluginHooks.PERSISTED_DATA_CHANGED,
+                );
+              }
+            }),
+          ),
+        ),
+      ),
+    { dispatch: false },
+  );
+
   // private static hiddenActions = [];
   anyAction$ = createEffect(
     () =>

src/app/plugins/plugin-hooks.spec.ts

@@ -0,0 +1,111 @@
+import { PluginHooks } from '@super-productivity/plugin-api';
+import { PluginHooksService } from './plugin-hooks';
+import { PluginLog } from '../core/log';
+
+describe('PluginHooksService.dispatchHookToPlugin', () => {
+  let service: PluginHooksService;
+  let logErrSpy: jasmine.Spy;
+
+  beforeEach(() => {
+    service = new PluginHooksService();
+    logErrSpy = spyOn(PluginLog, 'err');
+  });
+
+  it('invokes only the targeted plugin handler and is a no-op for unregistered ids', async () => {
+    const handlerA = jasmine.createSpy('handlerA');
+    const handlerB = jasmine.createSpy('handlerB');
+    service.registerHookHandler('plugin-a', PluginHooks.PERSISTED_DATA_CHANGED, handlerA);
+    service.registerHookHandler('plugin-b', PluginHooks.PERSISTED_DATA_CHANGED, handlerB);
+
+    await service.dispatchHookToPlugin('plugin-a', PluginHooks.PERSISTED_DATA_CHANGED);
+
+    expect(handlerA).toHaveBeenCalledTimes(1);
+    expect(handlerB).not.toHaveBeenCalled();
+
+    // Unregistered pluginId — must not throw or affect existing handlers.
+    await expectAsync(
+      service.dispatchHookToPlugin('plugin-nope', PluginHooks.PERSISTED_DATA_CHANGED),
+    ).toBeResolved();
+    expect(handlerA).toHaveBeenCalledTimes(1);
+    expect(handlerB).not.toHaveBeenCalled();
+  });
+
+  it('swallows synchronously-thrown handler errors and logs them', async () => {
+    const throwing = jasmine
+      .createSpy('throwing')
+      .and.throwError(new Error('boom from plugin'));
+    service.registerHookHandler('bad', PluginHooks.PERSISTED_DATA_CHANGED, throwing);
+
+    await expectAsync(
+      service.dispatchHookToPlugin('bad', PluginHooks.PERSISTED_DATA_CHANGED),
+    ).toBeResolved();
+    expect(throwing).toHaveBeenCalledTimes(1);
+    // Pin the error-swallow branch so a future refactor that drops the
+    // try/catch surfaces as a spec failure rather than a silent regression.
+    expect(logErrSpy).toHaveBeenCalledTimes(1);
+    expect(logErrSpy.calls.mostRecent().args[0]).toContain(
+      'Plugin bad persistedDataChanged handler error',
+    );
+  });
+
+  it('swallows async-rejected handler promises and logs them', async () => {
+    const rejecting = jasmine
+      .createSpy('rejecting')
+      .and.returnValue(Promise.reject(new Error('async boom')));
+    service.registerHookHandler(
+      'async-bad',
+      PluginHooks.PERSISTED_DATA_CHANGED,
+      rejecting,
+    );
+
+    await expectAsync(
+      service.dispatchHookToPlugin('async-bad', PluginHooks.PERSISTED_DATA_CHANGED),
+    ).toBeResolved();
+    expect(rejecting).toHaveBeenCalledTimes(1);
+    expect(logErrSpy).toHaveBeenCalledTimes(1);
+  });
+
+  it('times out a stuck handler at HOOK_TIMEOUT_MS and logs the timeout', async () => {
+    jasmine.clock().install();
+    try {
+      const stuck = jasmine
+        .createSpy('stuck')
+        .and.returnValue(new Promise<void>(() => {}));
+      service.registerHookHandler('hang', PluginHooks.PERSISTED_DATA_CHANGED, stuck);
+
+      const dispatch = service.dispatchHookToPlugin(
+        'hang',
+        PluginHooks.PERSISTED_DATA_CHANGED,
+      );
+      // Trip the 5s timeout race.
+      jasmine.clock().tick(6000);
+      await expectAsync(dispatch).toBeResolved();
+      expect(stuck).toHaveBeenCalledTimes(1);
+      // Without this assertion the spec would silently pass even if the
+      // timeout race were removed entirely (dispatcher always resolves).
+      expect(logErrSpy).toHaveBeenCalledTimes(1);
+      const [msg, err] = logErrSpy.calls.mostRecent().args;
+      expect(msg).toContain('Plugin hang persistedDataChanged handler error');
+      expect((err as Error).message).toBe('Hook handler timed out');
+    } finally {
+      jasmine.clock().uninstall();
+    }
+  });
+
+  it('refuses to register handlers under pluginIds containing ":"', () => {
+    // Defense-in-depth: composeId already throws on colon-bearing pluginIds at
+    // the persistence boundary, but the hooks Map is independently callable —
+    // and the persistence boundary cleared the only programmatic path that
+    // would have validated. A handler registered as 'victim:doc' would never
+    // fire today (the differ collapses to bare owner), but the invariant is
+    // worth pinning here so a future shape change can't reopen the gap.
+    const handler = jasmine.createSpy('handler');
+    expect(() =>
+      service.registerHookHandler(
+        'victim:doc',
+        PluginHooks.PERSISTED_DATA_CHANGED,
+        handler,
+      ),
+    ).toThrowError(/must not contain ':'/);
+  });
+});

src/app/plugins/plugin-hooks.ts

@@ -14,13 +14,22 @@ export class PluginHooksService {
   private _handlers = new Map<Hooks, Map<string, PluginHookHandler<any>>>();
 
   /**
-   * Register a hook handler
+   * Register a hook handler.
+   * Rejects pluginIds containing `:` so the persistence-key grammar
+   * (`pluginId[:key]`) cannot be subverted via the hooks registry — the
+   * differ collapses keyed entityIds to owner pluginIds for dispatch, and a
+   * spoofed `victim:doc` registration would silently never fire.
    */
   registerHookHandler<T extends Hooks>(
     pluginId: string,
     hook: T,
     handler: PluginHookHandler<T>,
   ): void {
+    if (pluginId.includes(':')) {
+      throw new Error(
+        `Plugin id "${pluginId}" must not contain ':' — the colon is reserved as the persistence-key delimiter.`,
+      );
+    }
     if (!this._handlers.has(hook)) {
       this._handlers.set(hook, new Map());
     }
@@ -29,32 +38,45 @@ export class PluginHooksService {
   }
 
   /**
-   * Dispatch a hook to all registered handlers
+   * Dispatch a hook to all registered handlers (fan-out).
+   * Use for events that potentially affect every plugin — task/project/language
+   * changes. For per-plugin events (e.g. `PERSISTED_DATA_CHANGED`) use
+   * {@link dispatchHookToPlugin} so only the owner's handler runs.
    */
   async dispatchHook(hook: Hooks, payload?: unknown): Promise<void> {
     const handlers = this._handlers.get(hook);
     if (!handlers || handlers.size === 0) {
       return;
     }
-
     for (const [pluginId, handler] of handlers) {
-      let timeoutId: ReturnType<typeof setTimeout>;
-      try {
-        await Promise.race([
-          handler(payload),
-          new Promise<void>((_, reject) => {
-            timeoutId = setTimeout(
-              () => reject(new Error('Hook handler timed out')),
-              PluginHooksService.HOOK_TIMEOUT_MS,
-            );
-          }),
-        ]).finally(() => clearTimeout(timeoutId));
-      } catch (error) {
-        PluginLog.err(`Plugin ${pluginId} ${hook} handler error:`, error);
-      }
+      await this._invokeWithTimeout(pluginId, hook, handler, payload);
     }
   }
 
+  /**
+   * Dispatch a hook to a single plugin's handler (scoped).
+   * Counterpart to {@link dispatchHook} — use for per-owner events such as
+   * `PERSISTED_DATA_CHANGED` where only the affected plugin should be notified.
+   * No-op if `pluginId` has no handler registered for `hook`.
+   *
+   * Callers are expected to invoke without awaiting (fire-and-forget). Each
+   * call allocates one `Promise.race([handler, 5s timeout])` (see
+   * `HOOK_TIMEOUT_MS` above) — the closure is released within the timeout
+   * window, so concurrent dispatches stay bounded by the number of in-flight
+   * plugins.
+   */
+  async dispatchHookToPlugin<T extends Hooks>(
+    pluginId: string,
+    hook: T,
+    payload?: unknown,
+  ): Promise<void> {
+    const handler = this._handlers.get(hook)?.get(pluginId);
+    if (!handler) {
+      return;
+    }
+    await this._invokeWithTimeout(pluginId, hook, handler, payload);
+  }
+
   /**
    * Unregister all hooks for a plugin
    */
@@ -70,4 +92,26 @@ export class PluginHooksService {
   clearAllHooks(): void {
     this._handlers.clear();
   }
+
+  private async _invokeWithTimeout(
+    pluginId: string,
+    hook: Hooks,
+    handler: PluginHookHandler<any>,
+    payload?: unknown,
+  ): Promise<void> {
+    let timeoutId: ReturnType<typeof setTimeout> | undefined;
+    try {
+      await Promise.race([
+        handler(payload),
+        new Promise<void>((_, reject) => {
+          timeoutId = setTimeout(
+            () => reject(new Error('Hook handler timed out')),
+            PluginHooksService.HOOK_TIMEOUT_MS,
+          );
+        }),
+      ]).finally(() => clearTimeout(timeoutId));
+    } catch (error) {
+      PluginLog.err(`Plugin ${pluginId} ${hook} handler error:`, error);
+    }
+  }
 }

src/app/plugins/plugin.service.ts

@@ -1048,6 +1048,17 @@ export class PluginService implements OnDestroy {
     await this._pluginHooks.dispatchHook(hookName, payload);
   }
 
+  async dispatchHookToPlugin(
+    pluginId: string,
+    hookName: Hooks,
+    payload?: unknown,
+  ): Promise<void> {
+    if (!this._isInitialized) {
+      return;
+    }
+    await this._pluginHooks.dispatchHookToPlugin(pluginId, hookName, payload);
+  }
+
   async loadPluginFromPath(pluginPath: string): Promise<PluginInstance> {
     const pluginInstance = await this._loadPlugin(pluginPath);
 

src/app/plugins/util/plugin-data-diff.util.spec.ts

@@ -0,0 +1,49 @@
+import { PluginUserData } from '../plugin-persistence.model';
+import { diffChangedPluginIds } from './plugin-data-diff.util';
+
+const entry = (id: string, data: string): PluginUserData => ({ id, data });
+
+describe('diffChangedPluginIds', () => {
+  it('returns [] when prev and next are entry-equal but a fresh array (no-op write)', () => {
+    // Covers the reducer's `state.map(...)` path producing a new array for an
+    // identical write — the encoded blob's `===` check on the entry must skip
+    // the no-op.
+    const prev = [entry('a', 'gz:abc'), entry('b', 'gz:def')];
+    const next = prev.map((e) => ({ ...e }));
+    expect(next).not.toBe(prev);
+    expect(diffChangedPluginIds(prev, next)).toEqual([]);
+  });
+
+  it('reports added, updated, and deleted in one pass', () => {
+    const prev = [
+      entry('keep', 'gz:same'),
+      entry('change', 'gz:old'),
+      entry('removed', 'gz:gone'),
+    ];
+    const next = [
+      entry('keep', 'gz:same'),
+      entry('change', 'gz:new'),
+      entry('added', 'gz:fresh'),
+    ];
+    const changed = diffChangedPluginIds(prev, next);
+    expect(changed.sort()).toEqual(['added', 'change', 'removed']);
+  });
+
+  it('normalizes keyed entityIds to the owner pluginId and dedupes', () => {
+    // Stage A: one plugin can own many `pluginId:key` entries. A handler is
+    // registered under the bare pluginId, so the differ must collapse the
+    // composites and fire the hook once per owner.
+    const prev = [
+      entry('foo:doc-1', 'gz:1'),
+      entry('foo:doc-2', 'gz:1'),
+      entry('bar', 'gz:1'),
+    ];
+    const next = [
+      entry('foo:doc-1', 'gz:2'), // updated
+      entry('foo:doc-3', 'gz:new'), // added — same owner
+      entry('bar', 'gz:1'), // unchanged
+      // 'foo:doc-2' missing → deleted, same owner
+    ];
+    expect(diffChangedPluginIds(prev, next)).toEqual(['foo']);
+  });
+});

src/app/plugins/util/plugin-data-diff.util.ts

@@ -0,0 +1,40 @@
+import { PluginUserData, PluginUserDataState } from '../plugin-persistence.model';
+import { extractOwnerPluginId } from './plugin-persistence-key.util';
+
+/**
+ * Returns the deduped set of *owner* pluginIds whose persisted data changed
+ * between two snapshots — added, deleted, or `data` differs.
+ *
+ * `PluginUserData.id` is the storage entityId, which is `pluginId` for the
+ * legacy single-blob form and `pluginId:key` for Stage A keyed entries. Hook
+ * handlers are registered under the bare pluginId, so this function maps
+ * back to the owner before deduping. A plugin with five keyed entries that
+ * all change in one emission fires the hook exactly once.
+ *
+ * Correctness contract: equality is `===` on the encoded `data` blob.
+ * **Never decode the gzip+base64 payload to "compare smartly."** Decoding
+ * allocates and the encoded form is what sync compares — the differ has to
+ * agree. Re-encoding identical plaintext yields byte-identical base64 per
+ * WHATWG `CompressionStream('gzip')` semantics, so identity check on the
+ * blob is sufficient to skip no-op writes.
+ */
+export const diffChangedPluginIds = (
+  prev: PluginUserDataState,
+  next: PluginUserDataState,
+): string[] => {
+  const prevMap = new Map<string, PluginUserData>(prev.map((e) => [e.id, e]));
+  const owners = new Set<string>();
+
+  for (const entry of next) {
+    const prior = prevMap.get(entry.id);
+    if (!prior || prior.data !== entry.data) {
+      owners.add(extractOwnerPluginId(entry.id));
+    }
+    prevMap.delete(entry.id);
+  }
+  for (const id of prevMap.keys()) {
+    owners.add(extractOwnerPluginId(id));
+  }
+
+  return Array.from(owners);
+};

src/app/plugins/util/plugin-persistence-key.util.ts

@@ -32,6 +32,17 @@ export const composeId = (pluginId: string, key?: string): string => {
 export const isPluginIdMatch = (entityId: string, pluginId: string): boolean =>
   entityId === pluginId || entityId.startsWith(pluginId + ':');
 
+/**
+ * Inverse of {@link composeId} — strip an optional `:key` suffix off a
+ * persistence entityId to recover the bare owner pluginId. Plugin hook
+ * handlers are registered under the bare pluginId, so any host-side
+ * dispatch keyed by entityId must normalize through here first.
+ */
+export const extractOwnerPluginId = (entityId: string): string => {
+  const idx = entityId.indexOf(':');
+  return idx === -1 ? entityId : entityId.slice(0, idx);
+};
+
 /**
  * Bound on a single plugin's persistence key length. Generous for any
  * realistic per-plugin keyspace (e.g. document-mode uses `doc:<uuid>`,