diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 3f8568c4cc72..86b8d014f900 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -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 diff --git a/docs/docs_map.md b/docs/docs_map.md index ff1d6ee54289..6aaace1ab0a2 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -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 diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md index 3d27636de22f..d309a4bbd492 100644 --- a/docs/plugins/sdk-runtime.md +++ b/docs/plugins/sdk-runtime.md @@ -856,6 +856,36 @@ two-party event loops that do not go through the shared inbound reply runner. +## 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: diff --git a/extensions/clickclack/src/discussions/eligibility.ts b/extensions/clickclack/src/discussions/eligibility.ts index 6fbbdf6b3981..d6a7819639b5 100644 --- a/extensions/clickclack/src/discussions/eligibility.ts +++ b/extensions/clickclack/src/discussions/eligibility.ts @@ -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)}`, + }; +} diff --git a/extensions/clickclack/src/discussions/history-format.ts b/extensions/clickclack/src/discussions/history-format.ts new file mode 100644 index 000000000000..69a4928edbbd --- /dev/null +++ b/extensions/clickclack/src/discussions/history-format.ts @@ -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>, +): 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}` : ""; +} diff --git a/extensions/clickclack/src/discussions/reconcile-scheduler.ts b/extensions/clickclack/src/discussions/reconcile-scheduler.ts new file mode 100644 index 000000000000..dce3291a39e4 --- /dev/null +++ b/extensions/clickclack/src/discussions/reconcile-scheduler.ts @@ -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; + 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 }>(); + readonly #retryState = new Map(); + readonly #inFlight = new Set(); + readonly #followUps = new Set(); + #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 }); + } +} diff --git a/extensions/clickclack/src/discussions/register.ts b/extensions/clickclack/src/discussions/register.ts index 853254bcd1d4..961e78ea3d24 100644 --- a/extensions/clickclack/src/discussions/register.ts +++ b/extensions/clickclack/src/discussions/register.ts @@ -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 }), ); diff --git a/extensions/clickclack/src/discussions/service-events.test.ts b/extensions/clickclack/src/discussions/service-events.test.ts new file mode 100644 index 000000000000..fb446d500d9d --- /dev/null +++ b/extensions/clickclack/src/discussions/service-events.test.ts @@ -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((_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((resolve) => { + resolveOld = resolve; + }), + ) + .mockImplementation(() => new Promise(() => {})); + + 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(); + } + }); +}); diff --git a/extensions/clickclack/src/discussions/service-test-support.ts b/extensions/clickclack/src/discussions/service-test-support.ts index b37c8e0b7527..c07a86474ae4 100644 --- a/extensions/clickclack/src/discussions/service-test-support.ts +++ b/extensions/clickclack/src/discussions/service-test-support.ts @@ -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; + 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, diff --git a/extensions/clickclack/src/discussions/service.ts b/extensions/clickclack/src/discussions/service.ts index 7ba775f31bb6..105d8373943a 100644 --- a/extensions/clickclack/src/discussions/service.ts +++ b/extensions/clickclack/src/discussions/service.ts @@ -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; 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>(); + 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 = Promise.resolve(); #timer: ReturnType | undefined; #reconcileAllPromise: Promise | 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 | 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 { - 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 { @@ -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 { + 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" }); } diff --git a/extensions/workboard/src/change-events.test.ts b/extensions/workboard/src/change-events.test.ts index 36692abd6f7d..1a5f28eb1eb9 100644 --- a/extensions/workboard/src/change-events.test.ts +++ b/extensions/workboard/src/change-events.test.ts @@ -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[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[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[0]; diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index 72c36ac46c85..df0fb0dab7f0 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -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( diff --git a/src/gateway/server-broadcast.ts b/src/gateway/server-broadcast.ts index 9fca7c166b4c..da4d67ae62f4 100644 --- a/src/gateway/server-broadcast.ts +++ b/src/gateway/server-broadcast.ts @@ -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, 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); }; diff --git a/src/gateway/server-session-events.test.ts b/src/gateway/server-session-events.test.ts index 7ef3bc9b6f2d..3e44b0e7b5c3 100644 --- a/src/gateway/server-session-events.test.ts +++ b/src/gateway/server-session-events.test.ts @@ -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(); + } + }); }); diff --git a/src/gateway/server-session-events.ts b/src/gateway/server-session-events.ts index ccbfa48a10f3..47f4552e981e 100644 --- a/src/gateway/server-session-events.ts +++ b/src/gateway/server-session-events.ts @@ -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; type SessionMessageSubscribers = Pick; +function hasSessionsChangedReceiver(connIds: ReadonlySet): 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); diff --git a/src/plugin-sdk/core.ts b/src/plugin-sdk/core.ts index b38e065f6b74..f38caa17b14f 100644 --- a/src/plugin-sdk/core.ts +++ b/src/plugin-sdk/core.ts @@ -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, diff --git a/src/plugins/gateway-events.ts b/src/plugins/gateway-events.ts index 4f27a9cf06ce..39e4d32b6a66 100644 --- a/src/plugins/gateway-events.ts +++ b/src/plugins/gateway-events.ts @@ -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(); + 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; + 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)}`); + } + } + }); +} diff --git a/src/plugins/services.test.ts b/src/plugins/services.test.ts index 5bfec6989372..7162464fd877 100644 --- a/src/plugins/services.test.ts +++ b/src/plugins/services.test.ts @@ -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(); diff --git a/src/plugins/services.ts b/src/plugins/services.ts index 14346e449b47..fdbc65cf57a9 100644 --- a/src/plugins/services.ts +++ b/src/plugins/services.ts @@ -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(); + } }, }; }