diff --git a/docs/plugins/sdk-migration.md b/docs/plugins/sdk-migration.md index 7ac425f04d3d..890f126e4fa6 100644 --- a/docs/plugins/sdk-migration.md +++ b/docs/plugins/sdk-migration.md @@ -458,12 +458,12 @@ For local media read policy, import `getAgentScopedMediaLocalRoots(...)` or `openclaw/plugin-sdk/infra-runtime` still exists for external - compatibility, but new code should import the focused surface it actually + compatibility, but new code should use the supported surface it actually needs: - | Need | Import | + | Need | Replacement | | --- | --- | - | System event queue helpers | `openclaw/plugin-sdk/system-event-runtime` | + | New system event producers | `api.runtime.system.enqueueSystemEvent` | | Heartbeat wake, event, and visibility helpers | `openclaw/plugin-sdk/heartbeat-runtime` | | Pending delivery queue drain | `openclaw/plugin-sdk/delivery-queue-runtime` | | Channel activity telemetry | `openclaw/plugin-sdk/channel-activity-runtime` | @@ -483,6 +483,14 @@ For local media read policy, import `getAgentScopedMediaLocalRoots(...)` or | Process-local async lock | `openclaw/plugin-sdk/async-lock-runtime` | | File locks | `openclaw/plugin-sdk/file-lock` | + System event snapshot inspection and consume helpers remain available only + through the deprecated `openclaw/plugin-sdk/infra-runtime` compatibility + surface; there is no modern public replacement. Current snapshots carry an + opaque `id` for one queued occurrence. Preserve it through copies and + serialization when returning a snapshot to consume. Legacy ID-less callers + retain structural matching, which can be ambiguous after queue churn. Do + not treat the ID as persistent or valid across restarts. + File-lock nesting is owner-scoped. Pass the same `reentrantOwner` only for nested acquisitions in one logical operation; omit it for ordinary locking. Never use a process-wide constant, because unrelated work would incorrectly diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md index bc9ac7558fbf..409d7762b6f9 100644 --- a/docs/plugins/sdk-runtime.md +++ b/docs/plugins/sdk-runtime.md @@ -720,7 +720,7 @@ two-party event loops that do not go through the shared inbound reply runner. System-level utilities. ```typescript - await api.runtime.system.enqueueSystemEvent(event); + const accepted = api.runtime.system.enqueueSystemEvent(text, options); api.runtime.system.requestHeartbeat({ source: "other", intent: "event", diff --git a/docs/plugins/sdk-subpaths.md b/docs/plugins/sdk-subpaths.md index 1eab858e1004..9e5b69270147 100644 --- a/docs/plugins/sdk-subpaths.md +++ b/docs/plugins/sdk-subpaths.md @@ -302,10 +302,10 @@ Use `isLoopbackHost(host)` when a plugin must accept only the local machine. It | `plugin-sdk/expect-runtime` | Private-local after July 2026; Required-value assertion helper for provable runtime invariants | | `plugin-sdk/number-runtime` | Private-local after July 2026; Numeric coercion helper | | `plugin-sdk/secure-random-runtime` | Private-local after July 2026; Secure token/UUID helpers | - | `plugin-sdk/system-event-runtime` | Private-local after July 2026; System event queue helpers | + | `plugin-sdk/system-event-runtime` | Private-local after July 2026; Narrow system event enqueue/peek helpers | | `plugin-sdk/transport-ready-runtime` | Private-local after July 2026; Transport readiness wait helper | | `plugin-sdk/exec-approvals-runtime` | Private-local after July 2026; Exec approval policy file helpers without the broad infra-runtime barrel | - | `plugin-sdk/infra-runtime` | Deprecated compatibility shim; use the focused runtime subpaths above | + | `plugin-sdk/infra-runtime` | Deprecated compatibility shim; use injected runtime APIs or documented typed-public subpaths | | `plugin-sdk/collection-runtime` | Small bounded cache helpers | | `plugin-sdk/diagnostic-runtime` | Diagnostic flag, event, trace-context, and low-cardinality dimension normalization helpers | | `plugin-sdk/error-runtime` | Error graph, formatting, unknown-value coercion, shared error classification helpers, `PlatformMessageNotDispatchedError`, `isApprovalNotFoundError` | diff --git a/src/infra/system-events.test.ts b/src/infra/system-events.test.ts index b5ad265e255a..f48b314495f9 100644 --- a/src/infra/system-events.test.ts +++ b/src/infra/system-events.test.ts @@ -16,12 +16,14 @@ import { consumeSystemEventEntries, drainSystemEventEntries, enqueueSystemEvent, + enqueueSystemEventEntry, hasSystemEvents, isSystemEventContextChanged, peekSystemEventEntries, peekSystemEvents, resetSystemEventsForTest, resolveSystemEventDeliveryContext, + type SystemEvent, } from "./system-events.js"; type SystemEventsModule = typeof import("./system-events.js"); @@ -218,6 +220,55 @@ describe("system events (session routing)", () => { expect(peekSystemEvents(key)).toEqual(["second"]); }); + it.each([ + { + name: "prefix consume with object spread", + consume: consumeSystemEventEntries, + copy: (event: SystemEvent): SystemEvent => ({ ...event }), + }, + { + name: "selected consume with structuredClone", + consume: consumeSelectedSystemEventEntries, + copy: (event: SystemEvent): SystemEvent => structuredClone(event), + }, + { + name: "prefix consume with JSON round trip", + consume: consumeSystemEventEntries, + // oxlint-disable-next-line unicorn/prefer-structured-clone -- This case exercises JSON transport. + copy: (event: SystemEvent): SystemEvent => JSON.parse(JSON.stringify(event)) as SystemEvent, + }, + ])("does not consume an identical successor from a stale copy: $name", ({ consume, copy }) => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-08T00:00:00Z")); + + const key = "agent:main:test-stale-copied-snapshot"; + const options = { + sessionKey: key, + contextKey: "build:123", + deliveryContext: { channel: "telegram", to: "-100123", threadId: "42" }, + }; + const original = expectDefined( + enqueueSystemEventEntry("Build completed", options), + "original event", + ); + const staleCopy = copy(original); + expect(staleCopy.id).toBe(original.id); + + expect(consume(key, [original]).map((event) => event.id)).toEqual([original.id]); + const successor = expectDefined( + enqueueSystemEventEntry("Build completed", options), + "successor event", + ); + expect(successor.id).not.toBe(original.id); + expect(successor).toEqual({ ...original, id: successor.id }); + + expect(consume(key, [staleCopy])).toStrictEqual([]); + expect(peekSystemEventEntries(key).map((event) => event.id)).toEqual([successor.id]); + + expect(consume(key, [successor]).map((event) => event.id)).toEqual([successor.id]); + expect(peekSystemEventEntries(key)).toStrictEqual([]); + }); + it("matches consumed delivery contexts through normalized route identity", () => { const key = "agent:main:test-consume-route-context"; enqueueSystemEvent("first", { @@ -228,13 +279,22 @@ describe("system events (session routing)", () => { threadId: 42.9, }, }); - const inspected = peekSystemEventEntries(key); - expectDefined( - expectDefined(inspected[0], "inspected event").deliveryContext, - "inspected delivery context", - ).threadId = "42"; + const current = expectDefined(peekSystemEventEntries(key)[0], "queued event"); + const legacyCopy: SystemEvent = { + text: current.text, + ts: current.ts, + contextKey: current.contextKey, + deliveryContext: { + channel: current.deliveryContext?.channel, + to: current.deliveryContext?.to, + threadId: "42", + }, + }; + expect(legacyCopy).not.toHaveProperty("id"); - expect(consumeSystemEventEntries(key, inspected).map((entry) => entry.text)).toEqual(["first"]); + expect(consumeSystemEventEntries(key, [legacyCopy]).map((entry) => entry.text)).toEqual([ + "first", + ]); expect(peekSystemEvents(key)).toStrictEqual([]); }); diff --git a/src/infra/system-events.ts b/src/infra/system-events.ts index 82df0c27e555..284bc2c799d8 100644 --- a/src/infra/system-events.ts +++ b/src/infra/system-events.ts @@ -14,6 +14,7 @@ import { normalizeDeliveryContext, } from "../utils/delivery-context.shared.js"; import type { DeliveryContext } from "../utils/delivery-context.types.js"; +import { generateSecureUuid } from "./secure-random.js"; import { cloneSystemEventOwner, recordSystemEventOwner, @@ -22,6 +23,12 @@ import { } from "./system-event-ownership.js"; export type SystemEvent = { + /** + * OpenClaw-assigned opaque identity for one queued occurrence. Preserve it when returning a + * snapshot to consume. It changes on replacement or re-enqueue; optional only for legacy + * ID-less compatibility. + */ + id?: string; text: string; ts: number; contextKey?: string | null; @@ -148,6 +155,7 @@ function enqueueOwnedSystemEventEntry( entry.lastContextKey = normalizedContextKey; } const event: SystemEvent = { + id: generateSecureUuid(), text: cleaned, ts: Date.now(), contextKey: normalizedContextKey, @@ -220,6 +228,7 @@ function replaceSystemEventEntry(text: string, options: SystemEventOptions): Sys !areDeliveryContextsEqual(event.deliveryContext, normalizedDeliveryContext), ); const event: SystemEvent = { + id: generateSecureUuid(), text: cleaned, ts: Date.now(), contextKey: normalizedContextKey, @@ -248,7 +257,7 @@ function isDuplicateSystemEvent( ); } -function areSystemEventsEqual(left: SystemEvent, right: SystemEvent): boolean { +function areLegacySystemEventsEqual(left: SystemEvent, right: SystemEvent): boolean { return ( left.text === right.text && left.ts === right.ts && @@ -258,6 +267,14 @@ function areSystemEventsEqual(left: SystemEvent, right: SystemEvent): boolean { ); } +function matchesConsumedSystemEvent(queued: SystemEvent, consumed: SystemEvent): boolean { + if (consumed.id !== undefined) { + // Queue-owned IDs govern modern consumption; only legacy ID-less snapshots use structure. + return queued.id === consumed.id; + } + return areLegacySystemEventsEqual(queued, consumed); +} + function resetQueueState(key: string, entry: SessionQueue) { if (entry.queue.length === 0) { entry.lastContextKey = null; @@ -286,7 +303,7 @@ export function consumeSystemEventEntries( if ( consumedEntries.length > entry.queue.length || !consumedEntries.every((event, index) => - areSystemEventsEqual(expectDefined(entry.queue[index], "queue entry at index"), event), + matchesConsumedSystemEvent(expectDefined(entry.queue[index], "queue entry at index"), event), ) ) { // A keyed replacement may remove one inspected entry while a prompt is in flight. @@ -310,7 +327,7 @@ export function consumeSelectedSystemEventEntries( } const removed: SystemEvent[] = []; for (const consumed of consumedEntries) { - const index = entry.queue.findIndex((event) => areSystemEventsEqual(event, consumed)); + const index = entry.queue.findIndex((event) => matchesConsumedSystemEvent(event, consumed)); if (index === -1) { continue; } diff --git a/src/plugin-sdk/infra-runtime.ts b/src/plugin-sdk/infra-runtime.ts index be5d14266cae..cb0c1dbed479 100644 --- a/src/plugin-sdk/infra-runtime.ts +++ b/src/plugin-sdk/infra-runtime.ts @@ -1,7 +1,7 @@ /** * @deprecated Compatibility shim only. Keep old plugins working, but do not * add new imports here and do not use this subpath from repo code. - * Prefer focused openclaw/plugin-sdk/ runtime subpaths instead. + * Prefer injected runtime APIs or documented typed-public subpaths instead. */ export * from "./delivery-queue-runtime.js"; diff --git a/src/plugin-sdk/system-event-runtime.ts b/src/plugin-sdk/system-event-runtime.ts index 07fc76077542..2e7e030a3af9 100644 --- a/src/plugin-sdk/system-event-runtime.ts +++ b/src/plugin-sdk/system-event-runtime.ts @@ -1,4 +1,4 @@ -// System event queue helpers without the broad infra-runtime barrel. +// Narrow system event enqueue/peek helper surface without the broad infra-runtime barrel. export { enqueueSystemEvent,