feat(plugins): deliver sessions.changed to plugin services (#114813)

* feat(plugins): deliver sessions.changed to plugin services * docs: regenerate docs map * refactor(clickclack): split reconcile scheduler and history formatting out of the discussion service * style: format discussion service
2026-08-12 21:53:00 -06:00 · 2026-07-27 20:56:44 -04:00
parent 1de8b74e70
commit 0bfe7dd357
19 changed files with 932 additions and 97 deletions
@@ -57,7 +57,7 @@ eb4c757fe0086c1dbfa4c3f3caf3dcff0d3cab3924c608237f08f740a6ee5f59  module/command
 20f3f8042de53e4eee61b64de9102c8c202b9299e6a29235647a4729f70145f2  module/config-mutation
 189fa5a240cad0404cd281ad0a14a105a8f3231278d87b71cfbc4f96fb8e48ef  module/config-runtime
 c1ea9510dfda047609a99d5d2cd1f1560f5d469a36e6b695766213d695c25b0f  module/conversation-runtime
-4f44038ff66f5945b58a6816e9e1a2e1103b50e14215da8c3e5c95cc064d2fbf  module/core
+9782f7c9fdaec5887dea5f1ef1e2ee4e2dbcb98d7a3dd76726925962d3c2e527  module/core
 4af19d59c2f18674e7d7f7dc1b358b644dc707e6bd601dc47168bd9e4a669940  module/dedupe-runtime
 f70c93d28053ca2e8353e45e6515ce7acef188097c6117d1545965d0699c8004  module/device-bootstrap
 6215d3af5923bf5a616d73062534968b69f448e3e30adc64ae9caebdd1a46d71  module/diagnostic-runtime
@@ -7686,6 +7686,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`.
  - H2: Config loading and writes
  - H2: Reusable runtime utilities
  - H2: Runtime namespaces
+  - H2: Gateway service events
  - H2: Storing runtime references
  - H2: Other top-level api fields
  - H2: Related
@@ -856,6 +856,36 @@ two-party event loops that do not go through the shared inbound reply runner.
  </Accordion>
 </AccordionGroup>

+## Gateway service events
+
+Long-lived services registered with `api.registerService(...)` receive a process-local
+`ctx.gatewayEvents` facade when the process runs a Gateway broadcaster; in runtimes without one the
+field is absent, so feature-detect it and keep a fallback (for example a coarse poll). Use
+`onSessionsChanged(...)` to react after the Gateway broadcasts a `sessions.changed` notice:
+
+```typescript
+let unsubscribeSessionsChanged: (() => void) | undefined;
+
+api.registerService({
+  id: "session-index",
+  start(ctx) {
+    unsubscribeSessionsChanged = ctx.gatewayEvents?.onSessionsChanged((event) => {
+      // event: { sessionKey, agentId?, label?, displayName?, reason?, phase? }
+      refreshSession(event.sessionKey);
+    });
+  },
+  stop() {
+    unsubscribeSessionsChanged?.();
+    unsubscribeSessionsChanged = undefined;
+  },
+});
+```
+
+The handler runs in the Gateway process and does not add a Gateway protocol subscription. Keep the
+returned unsubscribe function and call it during service cleanup. The payload is a lightweight
+change notice; use `api.runtime.agent.session.getSessionEntry(...)` when the plugin needs the full
+current session entry.
+
 ## Storing runtime references

 Use `createPluginRuntimeStore` to store the runtime reference for use outside the `register` callback:
@@ -1,3 +1,4 @@
+import type { SessionDiscussionInfo } from "openclaw/plugin-sdk/session-discussion";
 import { listEnabledClickClackAccounts } from "../accounts.js";
 import type { CoreConfig, ResolvedClickClackAccount } from "../types.js";
 import type { ClickClackDiscussionBinding } from "./binding-store.js";
@@ -42,3 +43,16 @@ export function resolveDiscussionBindingAccount(
  }
  return { state: "active", account };
 }
+
+/** Public embed/open URLs for one bound discussion channel. */
+export function discussionInfoForBinding(
+  binding: ClickClackDiscussionBinding,
+  account: ResolvedClickClackAccount,
+): SessionDiscussionInfo {
+  const baseUrl = normalizedServerBaseUrl(account);
+  return {
+    state: "open",
+    embedUrl: `${baseUrl}/embed/channel/${encodeURIComponent(binding.workspaceRouteId)}/${encodeURIComponent(binding.channelRouteId)}`,
+    openUrl: `${baseUrl}/app/${encodeURIComponent(binding.workspaceRouteId)}/${encodeURIComponent(binding.channelRouteId)}`,
+  };
+}
@@ -0,0 +1,26 @@
+import type { ClickClackClient } from "../http-client.js";
+
+// JSON-escape plus Unicode line separators so one history record stays one line.
+function discussionRecordJson(value: string): string {
+  return JSON.stringify(value).replace(
+    /[\u0085\u2028\u2029]/gu,
+    (separator) => `\\u${separator.charCodeAt(0).toString(16).padStart(4, "0")}`,
+  );
+}
+
+/** Renders channel history as line-per-message records for the discussion tool. */
+export function formatDiscussionHistory(
+  history: Awaited<ReturnType<ClickClackClient["latestChannelMessages"]>>,
+): string {
+  const text = history.messages
+    .map((message) => {
+      const author =
+        message.author?.display_name || message.author?.handle || message.author_id || "Unknown";
+      return `timestamp=${discussionRecordJson(message.created_at)} [Author ${discussionRecordJson(author)} id=${discussionRecordJson(message.author_id)}] text=${discussionRecordJson(message.body)}`;
+    })
+    .join("\n");
+  const truncationNote = history.truncated
+    ? "\n[History scan reached its safety bound; older active threads may be omitted.]"
+    : "";
+  return text ? `${text}${truncationNote}` : "";
+}
@@ -0,0 +1,116 @@
+const RECONCILE_DEBOUNCE_MS = 250;
+const RECONCILE_RETRY_MS = 1_000;
+const RECONCILE_RETRY_MAX_MS = 60_000;
+
+type SchedulerHooks = {
+  /** Gate evaluated at scheduling time; false drops the request. */
+  shouldSchedule: (sessionKey: string) => boolean;
+  run: (sessionKey: string) => Promise<void>;
+  warn: (message: string) => void;
+};
+
+/**
+ * Per-session debounced reconcile scheduling with failure backoff.
+ * Contracts: bursts coalesce onto the earliest scheduled run (steady event
+ * traffic can never postpone one); a failed run sets a not-before floor with
+ * growing backoff that later events cannot undercut; events during an
+ * in-flight run collapse into one follow-up scheduled after settlement; and
+ * `supersede()` invalidates callbacks from earlier activations so stop/start
+ * cycles cannot corrupt the bookkeeping of the next one.
+ */
+export class DiscussionReconcileScheduler {
+  readonly #hooks: SchedulerHooks;
+  readonly #timers = new Map<string, { fireAt: number; timer: ReturnType<typeof setTimeout> }>();
+  readonly #retryState = new Map<string, { delayMs: number; notBefore: number }>();
+  readonly #inFlight = new Set<string>();
+  readonly #followUps = new Set<string>();
+  #generation = 0;
+
+  constructor(hooks: SchedulerHooks) {
+    this.#hooks = hooks;
+  }
+
+  /** Invalidates callbacks scheduled by prior activations. */
+  supersede(): void {
+    this.#generation += 1;
+  }
+
+  clear(): void {
+    for (const scheduled of this.#timers.values()) {
+      clearTimeout(scheduled.timer);
+    }
+    this.#timers.clear();
+    this.#retryState.clear();
+    this.#inFlight.clear();
+    this.#followUps.clear();
+  }
+
+  schedule(sessionKey: string, delayMs = RECONCILE_DEBOUNCE_MS): void {
+    if (!this.#hooks.shouldSchedule(sessionKey)) {
+      return;
+    }
+    if (this.#inFlight.has(sessionKey)) {
+      this.#followUps.add(sessionKey);
+      return;
+    }
+    const retry = this.#retryState.get(sessionKey);
+    const fireAt = Math.max(Date.now() + delayMs, retry?.notBefore ?? 0);
+    const existing = this.#timers.get(sessionKey);
+    if (existing) {
+      if (existing.fireAt <= fireAt) {
+        return;
+      }
+      clearTimeout(existing.timer);
+    }
+    const generation = this.#generation;
+    const timer = setTimeout(
+      () => {
+        if (generation !== this.#generation) {
+          return;
+        }
+        this.#timers.delete(sessionKey);
+        this.#inFlight.add(sessionKey);
+        void this.#hooks
+          .run(sessionKey)
+          .then(
+            () => {
+              if (generation !== this.#generation) {
+                return;
+              }
+              this.#retryState.delete(sessionKey);
+            },
+            (error: unknown) => {
+              if (generation !== this.#generation || !this.#hooks.shouldSchedule(sessionKey)) {
+                return;
+              }
+              const previous = this.#retryState.get(sessionKey);
+              const nextRetryMs = Math.min(
+                previous === undefined ? RECONCILE_RETRY_MS : previous.delayMs * 2,
+                RECONCILE_RETRY_MAX_MS,
+              );
+              this.#retryState.set(sessionKey, {
+                delayMs: nextRetryMs,
+                notBefore: Date.now() + nextRetryMs,
+              });
+              this.#hooks.warn(
+                `discussion event reconcile failed for ${sessionKey}; retrying in ${nextRetryMs}ms: ${String(error)}`,
+              );
+              this.#followUps.add(sessionKey);
+            },
+          )
+          .finally(() => {
+            if (generation !== this.#generation) {
+              return;
+            }
+            this.#inFlight.delete(sessionKey);
+            if (this.#followUps.delete(sessionKey)) {
+              this.schedule(sessionKey);
+            }
+          });
+      },
+      Math.max(fireAt - Date.now(), 0),
+    );
+    timer.unref?.();
+    this.#timers.set(sessionKey, { fireAt, timer });
+  }
+}
@@ -15,6 +15,15 @@ export function registerClickClackDiscussions(api: OpenClawPluginApi): void {
  }

  const service = new ClickClackDiscussionService(api.runtime);
+  api.registerService({
+    id: "clickclack-discussion-session-events",
+    start: ({ gatewayEvents }) => {
+      service.bindGatewayEvents(gatewayEvents);
+    },
+    stop: () => {
+      service.cleanup();
+    },
+  });
  api.registerTool((context) =>
    createClickClackDiscussionTool({ service, sessionKey: context.sessionKey }),
  );
@@ -0,0 +1,311 @@
+import type {
+  OpenClawPluginGatewayEvents,
+  OpenClawPluginSessionsChangedEvent,
+} from "openclaw/plugin-sdk/core";
+import { describe, expect, it, vi } from "vitest";
+import { createHarness } from "./service-test-support.js";
+
+function createGatewayEventsHarness() {
+  const handlers = new Set<(event: OpenClawPluginSessionsChangedEvent) => void>();
+  const unsubscribe = vi.fn();
+  const gatewayEvents = {
+    emit: vi.fn(),
+    onSessionsChanged: vi.fn((handler) => {
+      handlers.add(handler);
+      let active = true;
+      return () => {
+        if (!active) {
+          return;
+        }
+        active = false;
+        handlers.delete(handler);
+        unsubscribe();
+      };
+    }),
+  } satisfies OpenClawPluginGatewayEvents;
+  return {
+    gatewayEvents,
+    unsubscribe,
+    emit(event: OpenClawPluginSessionsChangedEvent) {
+      for (const handler of handlers) {
+        handler(event);
+      }
+    },
+  };
+}
+
+describe("ClickClack discussion session events", () => {
+  it("runs one catch-up reconciliation when gateway events bind", async () => {
+    vi.useFakeTimers();
+    const gateway = createGatewayEventsHarness();
+    const harness = createHarness({ label: "Catch up" });
+    const sessionKey = "agent:main:event-catch-up";
+    try {
+      await harness.service.open(sessionKey);
+      const reconcile = vi.spyOn(harness.service, "reconcile").mockResolvedValue(undefined);
+
+      harness.service.bindGatewayEvents(gateway.gatewayEvents);
+      await vi.advanceTimersByTimeAsync(0);
+
+      expect(reconcile).toHaveBeenCalledOnce();
+      expect(reconcile).toHaveBeenCalledWith(sessionKey);
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("reconciles a renamed bound session from sessions.changed", async () => {
+    vi.useFakeTimers();
+    const gateway = createGatewayEventsHarness();
+    const harness = createHarness(
+      { label: "Original", category: "Projects" },
+      { gatewayEvents: gateway.gatewayEvents },
+    );
+    const sessionKey = "agent:main:event-rename";
+    try {
+      await harness.service.open(sessionKey);
+      harness.updateChannel.mockClear();
+      harness.setSessionEntry({ label: "Renamed", category: "Projects" });
+
+      gateway.emit({ sessionKey, label: "Renamed", reason: "rename" });
+      await vi.advanceTimersByTimeAsync(249);
+      expect(harness.updateChannel).not.toHaveBeenCalled();
+      await vi.advanceTimersByTimeAsync(1);
+
+      expect(harness.updateChannel).toHaveBeenCalledOnce();
+      expect(harness.updateChannel).toHaveBeenCalledWith(
+        "chn_discussion",
+        expect.objectContaining({ display_title: "Renamed", name: "renamed" }),
+      );
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("debounces sessions.changed bursts per bound session", async () => {
+    vi.useFakeTimers();
+    const gateway = createGatewayEventsHarness();
+    const harness = createHarness({ label: "Burst" }, { gatewayEvents: gateway.gatewayEvents });
+    const sessionKey = "agent:main:event-burst";
+    try {
+      await harness.service.open(sessionKey);
+      const reconcile = vi.spyOn(harness.service, "reconcile").mockResolvedValue(undefined);
+
+      // Coalescing keeps the earliest scheduled run: bursts collapse into one
+      // reconcile per window, and steady event traffic cannot postpone it.
+      gateway.emit({ sessionKey, phase: "message" });
+      await vi.advanceTimersByTimeAsync(100);
+      gateway.emit({ sessionKey, phase: "message" });
+      await vi.advanceTimersByTimeAsync(100);
+      gateway.emit({ sessionKey, label: "Burst renamed", reason: "rename" });
+      await vi.advanceTimersByTimeAsync(49);
+      expect(reconcile).not.toHaveBeenCalled();
+      await vi.advanceTimersByTimeAsync(1);
+
+      expect(reconcile).toHaveBeenCalledOnce();
+      expect(reconcile).toHaveBeenCalledWith(sessionKey);
+      await vi.advanceTimersByTimeAsync(1_000);
+      expect(reconcile).toHaveBeenCalledOnce();
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("retries a failed event reconciliation once without polling bindings", async () => {
+    vi.useFakeTimers();
+    const gateway = createGatewayEventsHarness();
+    const harness = createHarness({ label: "Retry" }, { gatewayEvents: gateway.gatewayEvents });
+    const sessionKey = "agent:main:event-retry";
+    try {
+      await harness.service.open(sessionKey);
+      harness.updateChannel.mockClear();
+      harness.updateChannel.mockRejectedValueOnce(new Error("temporary outage"));
+      harness.setSessionEntry({ label: "Retry renamed" });
+
+      gateway.emit({ sessionKey, label: "Retry renamed", reason: "rename" });
+      await vi.advanceTimersByTimeAsync(250);
+      expect(harness.updateChannel).toHaveBeenCalledOnce();
+      await vi.advanceTimersByTimeAsync(999);
+      expect(harness.updateChannel).toHaveBeenCalledOnce();
+      await vi.advanceTimersByTimeAsync(1);
+
+      expect(harness.updateChannel).toHaveBeenCalledTimes(2);
+      expect(harness.updateChannel).toHaveBeenLastCalledWith(
+        "chn_discussion",
+        expect.objectContaining({ display_title: "Retry renamed", name: "retry-renamed" }),
+      );
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("unsubscribes and cancels event reconciliation during cleanup", async () => {
+    vi.useFakeTimers();
+    const gateway = createGatewayEventsHarness();
+    const harness = createHarness({ label: "Cleanup" }, { gatewayEvents: gateway.gatewayEvents });
+    const sessionKey = "agent:main:event-cleanup";
+    try {
+      await harness.service.open(sessionKey);
+      const reconcile = vi.spyOn(harness.service, "reconcile").mockResolvedValue(undefined);
+      gateway.emit({ sessionKey, phase: "message" });
+      harness.service.cleanup();
+      await vi.advanceTimersByTimeAsync(250);
+      gateway.emit({ sessionKey, reason: "rename" });
+      await vi.advanceTimersByTimeAsync(250);
+
+      expect(gateway.unsubscribe).toHaveBeenCalledOnce();
+      expect(reconcile).not.toHaveBeenCalled();
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("defers events arriving during a failing reconcile to the backoff floor", async () => {
+    vi.useFakeTimers();
+    const gateway = createGatewayEventsHarness();
+    const harness = createHarness({ label: "Slow" }, { gatewayEvents: gateway.gatewayEvents });
+    const sessionKey = "agent:main:event-slow-failure";
+    try {
+      await harness.service.open(sessionKey);
+      let rejectInFlight: ((error: Error) => void) | undefined;
+      const reconcile = vi.spyOn(harness.service, "reconcile").mockImplementationOnce(
+        () =>
+          new Promise<void>((_resolve, reject) => {
+            rejectInFlight = reject;
+          }),
+      );
+
+      gateway.emit({ sessionKey, label: "Slow renamed", reason: "rename" });
+      await vi.advanceTimersByTimeAsync(250);
+      expect(reconcile).toHaveBeenCalledOnce();
+
+      // Event during the in-flight run: must not arm a fresh 250ms timer that
+      // would outlive (and undercut) the failure's backoff floor.
+      gateway.emit({ sessionKey, label: "Slow renamed again", reason: "rename" });
+      reconcile.mockResolvedValue(undefined);
+      rejectInFlight?.(new Error("outage"));
+      await vi.advanceTimersByTimeAsync(999);
+      expect(reconcile).toHaveBeenCalledOnce();
+      await vi.advanceTimersByTimeAsync(1);
+      expect(reconcile).toHaveBeenCalledTimes(2);
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("does not poll bindings after a successful open while events are bound", async () => {
+    vi.useFakeTimers();
+    const gateway = createGatewayEventsHarness();
+    const harness = createHarness(
+      { label: "No poll" },
+      { gatewayEvents: gateway.gatewayEvents, startTimer: true },
+    );
+    try {
+      await harness.service.open("agent:main:no-poll");
+      const reconcileAll = vi.spyOn(harness.service, "reconcileAll");
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(reconcileAll).not.toHaveBeenCalled();
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("keeps polling bindings when no gateway event subscription exists", async () => {
+    vi.useFakeTimers();
+    const harness = createHarness({ label: "Poll fallback" }, { startTimer: true });
+    try {
+      await harness.service.open("agent:main:poll-fallback");
+      const reconcileAll = vi.spyOn(harness.service, "reconcileAll").mockResolvedValue(undefined);
+      await vi.advanceTimersByTimeAsync(60_000);
+      // Broadcaster-less runtimes never receive sessions.changed; without the
+      // interval poll their bindings would never reconcile renames or archives.
+      expect(reconcileAll).toHaveBeenCalled();
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("ignores settle callbacks from a superseded activation after restart", async () => {
+    vi.useFakeTimers();
+    const gateway = createGatewayEventsHarness();
+    const harness = createHarness({ label: "Stale" }, { gatewayEvents: gateway.gatewayEvents });
+    const sessionKey = "agent:main:event-stale-settle";
+    try {
+      await harness.service.open(sessionKey);
+      let resolveOld: (() => void) | undefined;
+      const reconcile = vi
+        .spyOn(harness.service, "reconcile")
+        .mockImplementationOnce(
+          () =>
+            new Promise<void>((resolve) => {
+              resolveOld = resolve;
+            }),
+        )
+        .mockImplementation(() => new Promise<void>(() => {}));
+
+      gateway.emit({ sessionKey, label: "Stale renamed", reason: "rename" });
+      await vi.advanceTimersByTimeAsync(250);
+      expect(reconcile).toHaveBeenCalledOnce();
+
+      // Restart while the old reconcile is mid-flight; its settle callbacks
+      // must not clear the new activation's in-flight marker.
+      harness.service.cleanup();
+      harness.service.bindGatewayEvents(gateway.gatewayEvents);
+      await vi.advanceTimersByTimeAsync(0);
+      expect(reconcile).toHaveBeenCalledTimes(2);
+
+      resolveOld?.();
+      await vi.advanceTimersByTimeAsync(0);
+      gateway.emit({ sessionKey, label: "Stale renamed again", reason: "rename" });
+      await vi.advanceTimersByTimeAsync(1_000);
+      // The catch-up run is still in flight, so the event coalesces instead of
+      // arming a third reconcile through corrupted bookkeeping.
+      expect(reconcile).toHaveBeenCalledTimes(2);
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("recovers the fallback poll after a broadcaster-less stop/start cycle", async () => {
+    vi.useFakeTimers();
+    const harness = createHarness({ label: "Restart" }, { startTimer: true });
+    try {
+      await harness.service.open("agent:main:poll-restart");
+      harness.service.cleanup();
+      harness.service.bindGatewayEvents(undefined);
+      const reconcileAll = vi.spyOn(harness.service, "reconcileAll").mockResolvedValue(undefined);
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(reconcileAll).toHaveBeenCalled();
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+
+  it("keeps the interval while an ambiguous discussion open is pending", async () => {
+    vi.useFakeTimers();
+    const harness = createHarness({ label: "Pending retry" }, { startTimer: true });
+    try {
+      harness.createChannel.mockRejectedValueOnce(new Error("ambiguous create"));
+      await expect(harness.service.open("agent:main:pending-retry")).rejects.toThrow(
+        "ambiguous create",
+      );
+      const reconcileAll = vi.spyOn(harness.service, "reconcileAll").mockResolvedValue(undefined);
+
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(reconcileAll).toHaveBeenCalledOnce();
+    } finally {
+      harness.service.cleanup();
+      vi.useRealTimers();
+    }
+  });
+});
@@ -1,5 +1,5 @@
 import { createPluginRuntimeMock } from "openclaw/plugin-sdk/channel-test-helpers";
-import type { PluginRuntime } from "openclaw/plugin-sdk/core";
+import type { OpenClawPluginGatewayEvents, PluginRuntime } from "openclaw/plugin-sdk/core";
 import type { PluginStateSyncKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime";
 import { vi } from "vitest";
 import type { ClickClackClient } from "../http-client.js";
@@ -77,7 +77,11 @@ export function createHarness(
        archivedAt?: number;
      }
    | undefined,
-  options: { bindingGenerationFactory?: () => string } = {},
+  options: {
+    bindingGenerationFactory?: () => string;
+    gatewayEvents?: Pick<OpenClawPluginGatewayEvents, "onSessionsChanged">;
+    startTimer?: boolean;
+  } = {},
 ) {
  let sessionEntry = entry;
  const config = discussionConfig();
@@ -167,7 +171,8 @@ export function createHarness(
    clientFactory: () => client,
    installationId: TEST_INSTALLATION_ID,
    bindingGenerationFactory: options.bindingGenerationFactory ?? (() => TEST_BINDING_GENERATION),
-    startTimer: false,
+    startTimer: options.startTimer ?? false,
+    ...(options.gatewayEvents ? { gatewayEvents: options.gatewayEvents } : {}),
  });
  return {
    runtime,
@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import type { PluginRuntime } from "openclaw/plugin-sdk/core";
+import type { OpenClawPluginGatewayEvents, PluginRuntime } from "openclaw/plugin-sdk/core";
 import type {
  SessionDiscussionInfo,
  SessionDiscussionProvider,
@@ -25,10 +25,12 @@ import {
 import { controlSessionUrl } from "./control-session-url.js";
 import {
  discussionAccounts,
+  discussionInfoForBinding,
  normalizedServerBaseUrl,
  resolveDiscussionBindingAccount,
  type DiscussionBindingAccountResolution,
 } from "./eligibility.js";
+import { formatDiscussionHistory } from "./history-format.js";
 import { getClickClackDiscussionInstallationId } from "./installation.js";
 import {
  discussionCredentialFingerprint,
@@ -36,6 +38,7 @@ import {
  resolveDiscussionLabel,
  truncateDiscussionDisplayTitle,
 } from "./naming.js";
+import { DiscussionReconcileScheduler } from "./reconcile-scheduler.js";
 import {
  clearClickClackDiscussionChannelRevoked,
  isClickClackDiscussionChannelRevoked,
@@ -56,30 +59,12 @@ type DiscussionServiceOptions = {
  clientFactory?: (account: ResolvedClickClackAccount) => ClickClackClient;
  installationId?: string;
  bindingGenerationFactory?: () => string;
+  gatewayEvents?: Pick<OpenClawPluginGatewayEvents, "onSessionsChanged">;
  startTimer?: boolean;
 };

 type DiscussionBindingUseResolution = DiscussionBindingAccountResolution | { state: "retargeted" };

-function discussionInfoForBinding(
-  binding: ClickClackDiscussionBinding,
-  account: ResolvedClickClackAccount,
-): SessionDiscussionInfo {
-  const baseUrl = normalizedServerBaseUrl(account);
-  return {
-    state: "open",
-    embedUrl: `${baseUrl}/embed/channel/${encodeURIComponent(binding.workspaceRouteId)}/${encodeURIComponent(binding.channelRouteId)}`,
-    openUrl: `${baseUrl}/app/${encodeURIComponent(binding.workspaceRouteId)}/${encodeURIComponent(binding.channelRouteId)}`,
-  };
-}
-
-function discussionRecordJson(value: string): string {
-  return JSON.stringify(value).replace(
-    /[\u0085\u2028\u2029]/gu,
-    (separator) => `\\u${separator.charCodeAt(0).toString(16).padStart(4, "0")}`,
-  );
-}
-
 export class ClickClackDiscussionService {
  readonly provider: SessionDiscussionProvider;
  readonly #runtime: PluginRuntime;
@@ -89,9 +74,21 @@ export class ClickClackDiscussionService {
  readonly #bindingGenerationFactory: () => string;
  readonly #timersEnabled: boolean;
  readonly #sessionLocks = new Map<string, Promise<unknown>>();
+  readonly #reconcileScheduler = new DiscussionReconcileScheduler({
+    shouldSchedule: (sessionKey) =>
+      !this.#closed &&
+      (this.#store.get(sessionKey) !== undefined ||
+        listPendingDiscussionOpens(this.#runtime).some(
+          (pending) => pending.sessionKey === sessionKey,
+        )),
+    run: async (sessionKey) => await this.reconcile(sessionKey),
+    warn: (message) => this.#logger().warn(message),
+  });
  #channelMutationLock: Promise<unknown> = Promise.resolve();
  #timer: ReturnType<typeof setInterval> | undefined;
  #reconcileAllPromise: Promise<void> | undefined;
+  #unsubscribeSessionsChanged: (() => void) | undefined;
+  #closed = false;

  constructor(runtime: PluginRuntime, options: DiscussionServiceOptions = {}) {
    this.#runtime = runtime;
@@ -107,6 +104,32 @@ export class ClickClackDiscussionService {
      info: async ({ sessionKey }) => await this.info(sessionKey),
      open: async ({ sessionKey }) => await this.open(sessionKey),
    };
+    // Activation (event subscription + catch-up reconciles) belongs to the
+    // registered service lifecycle via bindGatewayEvents; construction alone
+    // must not touch remote channels. Tests may inject events for immediacy.
+    if (options.gatewayEvents) {
+      this.bindGatewayEvents(options.gatewayEvents);
+    }
+    if (this.#timersEnabled) {
+      this.#ensureTimer();
+    }
+  }
+
+  bindGatewayEvents(
+    gatewayEvents: Pick<OpenClawPluginGatewayEvents, "onSessionsChanged"> | undefined,
+  ): void {
+    // Service (re)start reactivates regardless of capability: a broadcaster-less
+    // runtime must still recover its fallback poll after a stop/start cycle.
+    // Only cleanup() closes the service.
+    this.#unsubscribeSessionsChanged?.();
+    this.#closed = false;
+    this.#reconcileScheduler.supersede();
+    this.#unsubscribeSessionsChanged = gatewayEvents?.onSessionsChanged((event) => {
+      this.#reconcileScheduler.schedule(event.sessionKey);
+    });
+    for (const sessionKey of this.#listReconcileSessionKeys()) {
+      this.#reconcileScheduler.schedule(sessionKey, 0);
+    }
    if (this.#timersEnabled) {
      this.#ensureTimer();
    }
@@ -174,37 +197,51 @@ export class ClickClackDiscussionService {
          }
        }
      }
-      const binding = await openClickClackDiscussionBinding({
-        runtime: this.#runtime,
-        store: this.#store,
-        account,
-        clientFactory: this.#clientFactory,
-        installationId: this.#installationId,
-        bindingGenerationFactory: this.#bindingGenerationFactory,
-        sessionKey,
-        ensureTimer: () => this.#ensureTimer(),
-        reconcilePendingOpen: async (pending) =>
-          await this.#reconcilePendingOpen(pending, { allowRetry: false }),
-        withChannelMutationLock: async (run) => await this.#withChannelMutationLock(run),
-        finalizePendingBinding: (key, nextBinding) =>
-          this.#finalizePendingBinding(key, nextBinding),
-        warn: (message) => this.#logger().warn(message),
-      });
+      let binding: ClickClackDiscussionBinding | undefined;
+      try {
+        binding = await openClickClackDiscussionBinding({
+          runtime: this.#runtime,
+          store: this.#store,
+          account,
+          clientFactory: this.#clientFactory,
+          installationId: this.#installationId,
+          bindingGenerationFactory: this.#bindingGenerationFactory,
+          sessionKey,
+          ensureTimer: () => this.#ensureTimer(),
+          reconcilePendingOpen: async (pending) =>
+            await this.#reconcilePendingOpen(pending, { allowRetry: false }),
+          withChannelMutationLock: async (run) => await this.#withChannelMutationLock(run),
+          finalizePendingBinding: (key, nextBinding) =>
+            this.#finalizePendingBinding(key, nextBinding),
+          warn: (message) => this.#logger().warn(message),
+        });
+      } finally {
+        this.#ensureTimer();
+      }
      if (!binding) {
        return { state: "available" };
      }
-      this.#ensureTimer();
      return discussionInfoForBinding(binding, account);
    });
  }

  async reconcile(sessionKey: string): Promise<void> {
-    await this.#withSessionLock(sessionKey, async () => {
-      const binding = this.#store.get(sessionKey);
-      if (binding) {
-        await this.#reconcileBinding(sessionKey, binding);
+    try {
+      await this.#withSessionLock(sessionKey, async () => {
+        const binding = this.#store.get(sessionKey);
+        if (binding) {
+          await this.#reconcileBinding(sessionKey, binding);
+        }
+      });
+      const pending = listPendingDiscussionOpens(this.#runtime).find(
+        (candidate) => candidate.sessionKey === sessionKey,
+      );
+      if (pending) {
+        await this.#reconcilePendingOpen(pending);
      }
-    });
+    } finally {
+      this.#ensureTimer();
+    }
  }

  async reconcileAll(): Promise<void> {
@@ -212,22 +249,13 @@ export class ClickClackDiscussionService {
      return await this.#reconcileAllPromise;
    }
    this.#reconcileAllPromise = (async () => {
-      for (const { sessionKey } of this.#store.entries()) {
+      for (const sessionKey of this.#listReconcileSessionKeys()) {
        try {
          await this.reconcile(sessionKey);
        } catch (error) {
          this.#logger().warn(`discussion reconcile failed for ${sessionKey}: ${String(error)}`);
        }
      }
-      for (const pending of listPendingDiscussionOpens(this.#runtime)) {
-        try {
-          await this.#reconcilePendingOpen(pending);
-        } catch (error) {
-          this.#logger().warn(
-            `discussion pending-open reconcile failed for ${pending.sessionKey}: ${String(error)}`,
-          );
-        }
-      }
    })().finally(() => {
      this.#reconcileAllPromise = undefined;
    });
@@ -268,23 +296,19 @@ export class ClickClackDiscussionService {
      binding.channelId,
      limit,
    );
-    const text = history.messages
-      .map((message) => {
-        const author =
-          message.author?.display_name || message.author?.handle || message.author_id || "Unknown";
-        return `timestamp=${discussionRecordJson(message.created_at)} [Author ${discussionRecordJson(author)} id=${discussionRecordJson(message.author_id)}] text=${discussionRecordJson(message.body)}`;
-      })
-      .join("\n");
-    const truncationNote = history.truncated
-      ? "\n[History scan reached its safety bound; older active threads may be omitted.]"
-      : "";
+    const text = formatDiscussionHistory(history);
    return {
      binding,
-      text: text ? `${text}${truncationNote}` : "The bound discussion has no messages yet.",
+      text: text || "The bound discussion has no messages yet.",
    };
  }

  cleanup(): void {
+    this.#closed = true;
+    this.#unsubscribeSessionsChanged?.();
+    this.#unsubscribeSessionsChanged = undefined;
+    this.#reconcileScheduler.supersede();
+    this.#reconcileScheduler.clear();
    if (this.#timer) {
      clearInterval(this.#timer);
      this.#timer = undefined;
@@ -607,35 +631,43 @@ export class ClickClackDiscussionService {
  }

  #ensureTimer(): void {
-    if (
-      !this.#timersEnabled ||
-      this.#timer ||
-      (this.#store.entries().length === 0 && listPendingDiscussionOpens(this.#runtime).length === 0)
-    ) {
+    const hasPendingOpens = listPendingDiscussionOpens(this.#runtime).length > 0;
+    // Without a gateway-event subscription (no broadcaster in this process),
+    // bindings fall back to the interval poll or renames would never reconcile.
+    const needsBindingPoll =
+      this.#unsubscribeSessionsChanged === undefined && this.#store.entries().length > 0;
+    if (this.#closed || (!hasPendingOpens && !needsBindingPoll)) {
+      if (this.#timer) {
+        clearInterval(this.#timer);
+        this.#timer = undefined;
+      }
      return;
    }
-    // The plugin event facade does not expose sessions.changed, and gateway.request
-    // has no subscriber connection to receive it. Reconcile only while bindings
-    // or ambiguous creates exist, at a coarse cadence, so this is not a hot poll.
+    if (!this.#timersEnabled || this.#timer) {
+      return;
+    }
+    // Session changes drive normal binding reconciliation through gateway events.
+    // Only ambiguous creates still need time-based retries while their durable
+    // pending-open record exists.
    this.#timer = setInterval(() => {
      void this.reconcileAll()
        .catch((error: unknown) => {
          this.#logger().warn(`discussion reconcile pass failed: ${String(error)}`);
        })
        .finally(() => {
-          if (
-            this.#store.entries().length === 0 &&
-            listPendingDiscussionOpens(this.#runtime).length === 0 &&
-            this.#timer
-          ) {
-            clearInterval(this.#timer);
-            this.#timer = undefined;
-          }
+          this.#ensureTimer();
        });
    }, RECONCILE_INTERVAL_MS);
    this.#timer.unref?.();
  }

+  #listReconcileSessionKeys(): Set<string> {
+    return new Set([
+      ...this.#store.entries().map(({ sessionKey }) => sessionKey),
+      ...listPendingDiscussionOpens(this.#runtime).map(({ sessionKey }) => sessionKey),
+    ]);
+  }
+
  #logger() {
    return this.#runtime.logging.getChildLogger({ plugin: "clickclack", feature: "discussions" });
  }
@@ -28,7 +28,7 @@ describe("createWorkboardChangeEventService", () => {
    const context = {
      config: {},
      stateDir: "/tmp/workboard-change-events-test",
-      gatewayEvents: { emit },
+      gatewayEvents: { emit, onSessionsChanged: () => () => undefined },
      logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() },
    } satisfies Parameters<typeof service.start>[0];

@@ -75,7 +75,7 @@ describe("createWorkboardChangeEventService", () => {
    const context = {
      config: {},
      stateDir: "/tmp/workboard-change-events-test",
-      gatewayEvents: { emit },
+      gatewayEvents: { emit, onSessionsChanged: () => () => undefined },
      logger: { debug: vi.fn(), info: vi.fn(), warn, error: vi.fn() },
    } satisfies Parameters<typeof service.start>[0];

@@ -110,7 +110,7 @@ describe("createWorkboardChangeEventService", () => {
    const context = {
      config: {},
      stateDir: "/tmp/workboard-change-events-test",
-      gatewayEvents: { emit: vi.fn() },
+      gatewayEvents: { emit: vi.fn(), onSessionsChanged: () => () => undefined },
      logger: { debug: vi.fn(), info: vi.fn(), warn, error: vi.fn() },
    } satisfies Parameters<typeof service.start>[0];

@@ -186,7 +186,8 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
      // +4: focused CLI root-option constants and parsers.
      // +6: model-picker action/capability and authoritative session-apply contracts.
      // +1: logger file-transport flush for graceful shutdown drains.
-      4738,
+      // +1: process-local sessions.changed plugin notification payload.
+      4739,
      env,
    ),
    publicFunctionExports: readPluginSdkSurfaceBudgetEnv(
@@ -5,6 +5,7 @@ import {
 // Gateway WebSocket broadcaster.
 // Applies event scope guards and slow-consumer handling before sending frames.
 import { logRejectedLargePayload } from "../logging/diagnostic-payload.js";
+import { queuePluginSessionsChanged } from "../plugins/gateway-events.js";
 import { isBrowserCopilotClient } from "../utils/message-channel.js";
 import {
  ADMIN_SCOPE,
@@ -201,6 +202,10 @@ export function createGatewayBroadcaster(params: {
    targetConnIds?: ReadonlySet<string>,
    explicitPluginScope?: GatewayPluginEventScope,
  ) => {
+    if (event === "sessions.changed") {
+      // Delivery is queued here so process-local handlers run after websocket fanout returns.
+      queuePluginSessionsChanged(payload);
+    }
    if (params.clients.size === 0) {
      return;
    }
@@ -324,9 +329,6 @@ export function createGatewayBroadcaster(params: {
    broadcastInternal(event, payload, opts);

  const broadcastToConnIds: GatewayBroadcastToConnIdsFn = (event, payload, connIds, opts) => {
-    if (connIds.size === 0) {
-      return;
-    }
    broadcastInternal(event, payload, opts, connIds);
  };

@@ -14,10 +14,11 @@ const sessionRow = vi.hoisted(() => ({
  agentRuntime: { id: "openclaw", source: "model" },
 }));
 const isEmbeddedAgentRunInProgressMock = vi.hoisted(() => vi.fn());
+const projectChatDisplayMessageMock = vi.hoisted(() => vi.fn((message: unknown) => message));

 vi.mock("../config/io.js", () => ({ getRuntimeConfig: () => ({}) }));
 vi.mock("./chat-display-projection.js", () => ({
-  projectChatDisplayMessage: (message: unknown) => message,
+  projectChatDisplayMessage: projectChatDisplayMessageMock,
 }));
 vi.mock("./session-utils.js", () => ({
  attachOpenClawTranscriptMeta: (message: unknown) => message,
@@ -37,6 +38,8 @@ vi.mock("../agents/embedded-agent-runner/runs.js", async () => {

 const { createLifecycleEventBroadcastHandler, createTranscriptUpdateBroadcastHandler } =
  await import("./server-session-events.js");
+const { createGatewayBroadcaster } = await import("./server-broadcast.js");
+const { subscribePluginSessionsChanged } = await import("../plugins/gateway-events.js");

 function createActiveRun(projectSessionActive: boolean): ChatAbortControllerEntry {
  return {
@@ -167,6 +170,36 @@ describe("createTranscriptUpdateBroadcastHandler", () => {
      senderIsOwner: true,
    });
  });
+
+  it("publishes message-phase changes to plugins without websocket subscribers", async () => {
+    const received = vi.fn();
+    const unsubscribe = subscribePluginSessionsChanged(received);
+    const { broadcastToConnIds } = createGatewayBroadcaster({ clients: new Set() });
+    const handler = createTranscriptUpdateBroadcastHandler({
+      broadcastToConnIds,
+      sessionEventSubscribers: { getAll: () => new Set() },
+      sessionMessageSubscribers: { get: () => new Set() },
+      chatAbortControllers: new Map(),
+    });
+    projectChatDisplayMessageMock.mockReturnValueOnce(undefined).mockReturnValueOnce(undefined);
+
+    try {
+      handler({
+        sessionFile: "/tmp/sess-main.jsonl",
+        sessionKey: "agent:main:main",
+        message: { role: "toolResult", content: [] },
+        messageId: "message-1",
+        messageSeq: 1,
+      });
+      await vi.waitFor(() => expect(received).toHaveBeenCalledOnce());
+      expect(received).toHaveBeenCalledWith({
+        sessionKey: "agent:main:main",
+        phase: "message",
+      });
+    } finally {
+      unsubscribe();
+    }
+  });
 });

 describe("createLifecycleEventBroadcastHandler", () => {
@@ -197,4 +230,31 @@ describe("createLifecycleEventBroadcastHandler", () => {
      { dropIfSlow: true },
    );
  });
+
+  it("publishes lifecycle changes to plugins without websocket subscribers", async () => {
+    const received = vi.fn();
+    const unsubscribe = subscribePluginSessionsChanged(received);
+    const { broadcastToConnIds } = createGatewayBroadcaster({ clients: new Set() });
+    const handler = createLifecycleEventBroadcastHandler({
+      broadcastToConnIds,
+      sessionEventSubscribers: { getAll: () => new Set() },
+      chatAbortControllers: new Map(),
+    });
+
+    try {
+      handler({
+        sessionKey: "agent:main:main",
+        reason: "rename",
+        label: "Renamed session",
+      });
+      await Promise.resolve();
+      expect(received).toHaveBeenCalledWith({
+        sessionKey: "agent:main:main",
+        label: "Renamed session",
+        reason: "rename",
+      });
+    } finally {
+      unsubscribe();
+    }
+  });
 });
@@ -9,6 +9,7 @@ import {
  resolveTranscriptSessionKeyBySessionId,
 } from "../config/sessions/session-accessor.js";
 import { parseSqliteSessionFileMarker } from "../config/sessions/sqlite-marker.js";
+import { hasPluginSessionsChangedSubscribers } from "../plugins/gateway-events.js";
 import { normalizeAgentId } from "../routing/session-key.js";
 import type { SessionLifecycleEvent } from "../sessions/session-lifecycle-events.js";
 import type { InternalSessionTranscriptUpdate } from "../sessions/transcript-events.js";
@@ -38,6 +39,10 @@ import {
 type SessionEventSubscribers = Pick<SessionEventSubscriberRegistry, "getAll">;
 type SessionMessageSubscribers = Pick<SessionMessageSubscriberRegistry, "get">;

+function hasSessionsChangedReceiver(connIds: ReadonlySet<string>): boolean {
+  return connIds.size > 0 || hasPluginSessionsChangedSubscribers();
+}
+
 function readMessageIdempotencyKey(message: unknown): string | undefined {
  if (!message || typeof message !== "object" || Array.isArray(message)) {
    return undefined;
@@ -184,7 +189,9 @@ async function handleTranscriptUpdateBroadcast(
    }
  }
  if (connIds.size === 0) {
-    return;
+    if (!hasPluginSessionsChangedSubscribers() || projectChatDisplayMessage(update.message)) {
+      return;
+    }
  }
  let messageSeq = asPositiveSafeInteger(update.messageSeq);
  if (messageSeq === undefined) {
@@ -264,7 +271,7 @@ async function handleTranscriptUpdateBroadcast(
  // Messages suppressed from display can still change transcript state, so
  // notify broad session listeners even when no session.message is emitted.
  const sessionEventConnIds = params.sessionEventSubscribers.getAll();
-  if (sessionEventConnIds.size === 0) {
+  if (!hasSessionsChangedReceiver(sessionEventConnIds)) {
    return;
  }
  params.broadcastToConnIds(
@@ -296,7 +303,7 @@ export function createLifecycleEventBroadcastHandler(params: {
      text?: string;
    };
    const connIds = params.sessionEventSubscribers.getAll();
-    if (connIds.size === 0) {
+    if (!hasSessionsChangedReceiver(connIds)) {
      return;
    }
    const sessionRow = loadGatewaySessionRow(event.sessionKey);
@@ -142,8 +142,11 @@ export type {
  OpenClawPluginToolContext,
  OpenClawPluginToolFactory,
 } from "../plugins/types.js";
-export type { OpenClawPluginGatewayEventScope } from "../plugins/gateway-events.js";
-export type { OpenClawPluginGatewayEvents } from "../plugins/gateway-events.js";
+export type {
+  OpenClawPluginGatewayEventScope,
+  OpenClawPluginGatewayEvents,
+  OpenClawPluginSessionsChangedEvent,
+} from "../plugins/gateway-events.js";
 export type {
  MemoryPluginCapability,
  MemoryPluginPublicArtifact,
@@ -1,11 +1,83 @@
+import { createSubsystemLogger } from "../logging/subsystem.js";
 import type { PluginJsonValue } from "./host-hook-json.js";

+const log = createSubsystemLogger("plugins");
+
 export type OpenClawPluginGatewayEventScope = "operator.read" | "operator.write" | "operator.admin";

+export type OpenClawPluginSessionsChangedEvent = {
+  sessionKey: string;
+  agentId?: string;
+  label?: string;
+  displayName?: string;
+  reason?: string;
+  phase?: string;
+};
+
+type SessionsChangedHandler = (event: OpenClawPluginSessionsChangedEvent) => unknown;
+
+const sessionsChangedHandlers = new Set<SessionsChangedHandler>();
+
 export type OpenClawPluginGatewayEvents = {
  emit: (
    event: string,
    payload: PluginJsonValue,
    opts: { scope: OpenClawPluginGatewayEventScope },
  ) => void;
+  /**
+   * Native plugins can already read full session entries through the injected runtime;
+   * this notice only avoids polling and does not widen session access.
+   */
+  onSessionsChanged: (handler: (event: OpenClawPluginSessionsChangedEvent) => void) => () => void;
 };
+
+export function subscribePluginSessionsChanged(
+  handler: (event: OpenClawPluginSessionsChangedEvent) => void,
+): () => void {
+  const subscription: SessionsChangedHandler = (event) => handler(event);
+  sessionsChangedHandlers.add(subscription);
+  return () => {
+    sessionsChangedHandlers.delete(subscription);
+  };
+}
+
+export function hasPluginSessionsChangedSubscribers(): boolean {
+  return sessionsChangedHandlers.size > 0;
+}
+
+export function queuePluginSessionsChanged(payload: unknown): void {
+  if (!payload || typeof payload !== "object" || Array.isArray(payload)) {
+    return;
+  }
+  const source = payload as Record<string, unknown>;
+  if (typeof source.sessionKey !== "string" || source.sessionKey.length === 0) {
+    return;
+  }
+  if (sessionsChangedHandlers.size === 0) {
+    return;
+  }
+  const subscriptions = [...sessionsChangedHandlers];
+  const event: OpenClawPluginSessionsChangedEvent = {
+    sessionKey: source.sessionKey,
+    ...(typeof source.agentId === "string" ? { agentId: source.agentId } : {}),
+    ...(typeof source.label === "string" ? { label: source.label } : {}),
+    ...(typeof source.displayName === "string" ? { displayName: source.displayName } : {}),
+    ...(typeof source.reason === "string" ? { reason: source.reason } : {}),
+    ...(typeof source.phase === "string" ? { phase: source.phase } : {}),
+  };
+  queueMicrotask(() => {
+    for (const handler of subscriptions) {
+      if (!sessionsChangedHandlers.has(handler)) {
+        continue;
+      }
+      try {
+        const result = handler(event);
+        void Promise.resolve(result).catch((error: unknown) => {
+          log.warn(`plugin sessions.changed handler failed: ${String(error)}`);
+        });
+      } catch (error) {
+        log.warn(`plugin sessions.changed handler failed: ${String(error)}`);
+      }
+    }
+  });
+}
@@ -17,6 +17,7 @@ vi.mock("../logging/subsystem.js", () => ({
 }));

 import { STATE_DIR } from "../config/paths.js";
+import { queuePluginSessionsChanged, subscribePluginSessionsChanged } from "./gateway-events.js";
 import { registerPluginHttpRoute } from "./http-registry.js";
 import {
  pinActivePluginHttpRouteRegistry,
@@ -193,6 +194,124 @@ describe("startPluginServices", () => {
    );
  });

+  it("omits gateway events entirely when no broadcaster exists", async () => {
+    let context: OpenClawPluginServiceContext | undefined;
+    const handle = await startPluginServices({
+      registry: createRegistry([
+        {
+          id: "events",
+          start: (ctx) => {
+            context = ctx;
+          },
+        },
+      ]),
+      config: createServiceConfig(),
+    });
+
+    // Presence of ctx.gatewayEvents is the capability signal plugins
+    // feature-detect; a facade with a dropping emit would defeat fallbacks.
+    expect(context?.gatewayEvents).toBeUndefined();
+    await handle.stop();
+  });
+
+  it("subscribes services to sessions.changed and revokes them on stop", async () => {
+    const received = vi.fn();
+    let context: OpenClawPluginServiceContext | undefined;
+    const handle = await startPluginServices({
+      registry: createRegistry([
+        {
+          id: "events",
+          start: (ctx) => {
+            context = ctx;
+            ctx.gatewayEvents?.onSessionsChanged(received);
+          },
+        },
+      ]),
+      config: createServiceConfig(),
+      broadcastPluginEvent: vi.fn(),
+    });
+
+    queuePluginSessionsChanged({ sessionKey: "agent:main:main", reason: "rename", ignored: 1 });
+    await Promise.resolve();
+    expect(received).toHaveBeenCalledWith({
+      sessionKey: "agent:main:main",
+      reason: "rename",
+    });
+    expect(() =>
+      context?.gatewayEvents?.emit("changed", {}, { scope: "operator.read" }),
+    ).not.toThrow();
+
+    await handle.stop();
+    queuePluginSessionsChanged({ sessionKey: "agent:main:main", reason: "archive" });
+    await Promise.resolve();
+    expect(received).toHaveBeenCalledOnce();
+  });
+
+  it("keeps duplicate handler subscriptions independent", async () => {
+    const received = vi.fn();
+    const unsubscribeFirst = subscribePluginSessionsChanged(received);
+    const unsubscribeSecond = subscribePluginSessionsChanged(received);
+
+    unsubscribeFirst();
+    queuePluginSessionsChanged({ sessionKey: "agent:main:main" });
+    await Promise.resolve();
+
+    expect(received).toHaveBeenCalledOnce();
+    unsubscribeSecond();
+  });
+
+  it("uses a stable sessions.changed subscription snapshot", async () => {
+    const received = vi.fn();
+    let unsubscribe: () => void = () => undefined;
+    const handler = () => {
+      received();
+      unsubscribe();
+      unsubscribe = subscribePluginSessionsChanged(handler);
+    };
+    unsubscribe = subscribePluginSessionsChanged(handler);
+
+    queuePluginSessionsChanged({ sessionKey: "agent:main:main" });
+    await Promise.resolve();
+
+    expect(received).toHaveBeenCalledOnce();
+    unsubscribe();
+  });
+
+  it("logs a throwing sessions.changed handler without blocking siblings", async () => {
+    const received = vi.fn();
+    const rejectingHandler = (() =>
+      Promise.reject(new Error("async handler failed"))) as () => void;
+    const handle = await startPluginServices({
+      registry: createRegistry([
+        {
+          id: "events",
+          start: (ctx) => {
+            ctx.gatewayEvents?.onSessionsChanged(() => {
+              throw new Error("handler failed");
+            });
+            ctx.gatewayEvents?.onSessionsChanged(rejectingHandler);
+            ctx.gatewayEvents?.onSessionsChanged(received);
+          },
+        },
+      ]),
+      config: createServiceConfig(),
+      broadcastPluginEvent: vi.fn(),
+    });
+
+    queuePluginSessionsChanged({ sessionKey: "agent:main:main", phase: "message" });
+    await Promise.resolve();
+    await Promise.resolve();
+
+    expect(received).toHaveBeenCalledOnce();
+    expect(mockedLogger.warn).toHaveBeenCalledWith(
+      "plugin sessions.changed handler failed: Error: handler failed",
+    );
+    expect(mockedLogger.warn).toHaveBeenCalledWith(
+      "plugin sessions.changed handler failed: Error: async handler failed",
+    );
+    await handle.stop();
+  });
+
  it("rejects unsafe event names, scopes, and payloads", async () => {
    let context: OpenClawPluginServiceContext | undefined;
    const broadcastPluginEvent = vi.fn();
@@ -7,6 +7,7 @@ import {
  onTrustedInternalDiagnosticEvent,
 } from "../infra/diagnostic-events.js";
 import { createSubsystemLogger } from "../logging/subsystem.js";
+import { subscribePluginSessionsChanged } from "./gateway-events.js";
 import { isPluginJsonValue, type PluginJsonValue } from "./host-hook-json.js";
 import { withPluginHttpRouteRegistry } from "./http-registry.js";
 import type { PluginServiceRegistration } from "./registry-types.js";
@@ -71,10 +72,16 @@ function createScopedGatewayEvents(params: {
  gatewayEvents?: OpenClawPluginServiceContext["gatewayEvents"];
  revoke: () => void;
 } {
+  // No broadcaster means no gateway events at all: emits have nowhere to go and
+  // sessions.changed is queued by the broadcaster itself. Omitting the facade
+  // keeps `ctx.gatewayEvents` presence as the capability signal plugins
+  // feature-detect; a silently dropping emit would defeat their fallbacks.
  if (!params.broadcast) {
    return { revoke: () => undefined };
  }
+  const broadcast = params.broadcast;
  let active = true;
+  const subscriptions = new Set<() => void>();
  return {
    gatewayEvents: {
      emit: (event, payload: PluginJsonValue, opts) => {
@@ -94,11 +101,31 @@ function createScopedGatewayEvents(params: {
        ) {
          throw new Error("plugin gateway event scope must be an operator scope");
        }
-        params.broadcast?.(`plugin.${params.pluginId}.${event}`, payload, opts.scope);
+        broadcast(`plugin.${params.pluginId}.${event}`, payload, opts.scope);
+      },
+      onSessionsChanged: (handler) => {
+        if (!active) {
+          throw new Error("plugin service gateway event subscriber is no longer active");
+        }
+        const unsubscribe = subscribePluginSessionsChanged(handler);
+        let subscribed = true;
+        const release = () => {
+          if (!subscribed) {
+            return;
+          }
+          subscribed = false;
+          subscriptions.delete(release);
+          unsubscribe();
+        };
+        subscriptions.add(release);
+        return release;
      },
    },
    revoke: () => {
      active = false;
+      for (const unsubscribe of subscriptions) {
+        unsubscribe();
+      }
    },
  };
 }