/**

* Session message event indexing and broadcast tests.

*/

import fs from "node:fs/promises";

import os from "node:os";

import path from "node:path";

import { afterAll, afterEach, beforeAll, describe, expect, test, vi } from "vitest";

import type { RawData } from "ws";

import {

GATEWAY_CLIENT_CAPS,

GATEWAY_CLIENT_IDS,

GATEWAY_CLIENT_MODES,

} from "../../packages/gateway-protocol/src/client-info.js";

import {

loadTranscriptEvents,

persistSessionTranscriptTurn,

} from "../config/sessions/session-accessor.js";

import { appendAssistantMessageToSessionTranscript } from "../config/sessions/transcript.js";

import type { OpenClawConfig } from "../config/types.openclaw.js";

import {

claimAgentRunContext,

clearAgentRunContext,

emitAgentEvent,

} from "../infra/agent-events.js";

import { rawDataToString } from "../infra/ws.js";

import { emitSessionLifecycleEvent } from "../sessions/session-lifecycle-events.js";

import * as transcriptEvents from "../sessions/transcript-events.js";

import { emitSessionTranscriptUpdate } from "../sessions/transcript-events.js";

import { testState } from "./test-helpers.runtime-state.js";

import {

connectOk,

connectReq,

createGatewaySuiteHarness,

installGatewayTestHooks,

onceMessage,

rpcReq,

writeSessionStore,

} from "./test-helpers.server.js";

import type { WorkerConnectionIdentity } from "./worker-environments/connection-identity.js";

import { createWorkerLiveEventReceiver } from "./worker-environments/live-events.js";

import type { WorkerTranscriptCommitStore } from "./worker-environments/transcript-commit-store.js";

import { createWorkerTranscriptCommitter } from "./worker-environments/transcript-commit.js";

installGatewayTestHooks({ scope: "suite" });

const cleanupDirs: string[] = [];

const SETUP_RPC_TIMEOUT_MS = 30_000;

let harness: Awaited<ReturnType<typeof createGatewaySuiteHarness>>;

let subscribedOperatorWs:

| Awaited<ReturnType<Awaited<ReturnType<typeof createGatewaySuiteHarness>>["openWs"]>>

| undefined;

// No explicit hook timeout: the suite harness cold-imports the full gateway

// server graph, which can legitimately exceed 60s on contended CI runners.

// Sibling gateway suites rely on the shared project hookTimeout (120s, 180s on

// Windows) for the same boot; tightening it here caused flaky hook timeouts.

beforeAll(async () => {

harness = await createGatewaySuiteHarness();

subscribedOperatorWs = await harness.openWs();

await connectOk(subscribedOperatorWs, {

scopes: ["operator.read"],

timeoutMs: SETUP_RPC_TIMEOUT_MS,

});

await rpcReq(subscribedOperatorWs, "sessions.subscribe", undefined, SETUP_RPC_TIMEOUT_MS);

});

afterAll(async () => {

subscribedOperatorWs?.close();

if (harness) {

await harness.close();

}

});

afterEach(async () => {

await Promise.all(

cleanupDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })),

);

});

async function createSessionStoreFile(): Promise<string> {

const dir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-session-message-"));

cleanupDirs.push(dir);

const storePath = path.join(dir, "sessions.json");

testState.sessionStorePath = storePath;

return storePath;

}

async function withOperatorSessionSubscriber<T>(

run: (ws: NonNullable<typeof subscribedOperatorWs>) => Promise<T>,

) {

if (!subscribedOperatorWs) {

throw new Error("subscribed operator websocket is not ready");

}

return await run(subscribedOperatorWs);

}

function waitForSessionMessageEvent(

ws: Awaited<ReturnType<Awaited<ReturnType<typeof createGatewaySuiteHarness>>["openWs"]>>,

sessionKey: string,

timeoutMs?: number,

) {

return onceMessage(

ws,

(message) =>

message.type === "event" &&

message.event === "session.message" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey === sessionKey,

timeoutMs,

);

}

function waitForSessionsChangedMessagePhase(

ws: Awaited<ReturnType<Awaited<ReturnType<typeof createGatewaySuiteHarness>>["openWs"]>>,

sessionKey: string,

) {

return onceMessage(

ws,

(message) =>

message.type === "event" &&

message.event === "sessions.changed" &&

(message.payload as { phase?: string; sessionKey?: string } | undefined)?.phase ===

"message" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey === sessionKey,

);

}

async function emitTranscriptUpdateAndCollectMessageEvent(params: {

ws: Awaited<ReturnType<Awaited<ReturnType<typeof createGatewaySuiteHarness>>["openWs"]>>;

sessionKey: string;

sessionFile: string;

message: Record<string, unknown>;

messageId: string;

agentId?: string;

messageSeq?: number;

}) {

const messageEventPromise = waitForSessionMessageEvent(params.ws, params.sessionKey);

emitSessionTranscriptUpdate({

sessionFile: params.sessionFile,

sessionKey: params.sessionKey,

...(params.agentId ? { agentId: params.agentId } : {}),

message: params.message,

messageId: params.messageId,

...(typeof params.messageSeq === "number" ? { messageSeq: params.messageSeq } : {}),

});

const messageEvent = await messageEventPromise;

return { messageEvent };

}

async function expectNoMessageWithin(params: {

action?: () => Promise<void> | void;

watch: (timeoutMs: number) => Promise<unknown>;

timeoutMs?: number;

}): Promise<void> {

const timeoutMs = params.timeoutMs ?? 300;

const received = params.watch(timeoutMs).then(

() => true,

() => false,

);

await params.action?.();

await expect(received).resolves.toBe(false);

}

function requireRecord(value: unknown, label: string): Record<string, unknown> {

if (!value || typeof value !== "object") {

throw new Error(`expected ${label} to be an object`);

}

return value as Record<string, unknown>;

}

function expectRecordFields(value: unknown, expected: Record<string, unknown>): void {

const record = requireRecord(value, "record");

for (const [key, expectedValue] of Object.entries(expected)) {

expect(record[key]).toEqual(expectedValue);

}

}

describe("session.message websocket events", () => {

test("enforces session-scoped chat delivery on real gateway connections", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: { sessionId: "sess-main", updatedAt: Date.now() },

other: { sessionId: "sess-other", updatedAt: Date.now() },

},

storePath,

});

const copilotOrigin = "chrome-extension://abcdefghijklmnopabcdefghijklmnop";

const copilotClient = {

id: GATEWAY_CLIENT_IDS.BROWSER_COPILOT,

version: "test",

platform: "chrome",

deviceFamily: "extension",

mode: GATEWAY_CLIENT_MODES.UI,

};

const copilotIdentityPath = path.join(path.dirname(storePath), "copilot-device.json");

const unpairedWs = await harness.openWs({ origin: copilotOrigin });

const pairingWs = await harness.openWs({ origin: copilotOrigin });

const wrongOriginWs = await harness.openWs({

origin: "chrome-extension://bcdefghijklmnopabcdefghijklmnopa",

});

const mainWs = await harness.openWs({ origin: copilotOrigin });

const otherWs = await harness.openWs();

const legacyWs = await harness.openWs();

try {

const scopedCaps = [GATEWAY_CLIENT_CAPS.SESSION_SCOPED_EVENTS];

const unpaired = await connectReq(unpairedWs, {

scopes: ["operator.read", "operator.write"],

caps: [...scopedCaps, GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS],

client: copilotClient,

deviceIdentityPath: path.join(path.dirname(storePath), "unpaired-copilot-device.json"),

prePairDevice: false,

});

expect(unpaired.ok).toBe(false);

expect(unpaired.error?.code).toBe("NOT_PAIRED");

unpairedWs.close();

const pairedHello = await connectOk(pairingWs, {

scopes: ["operator.read", "operator.write"],

caps: [...scopedCaps, GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS],

client: copilotClient,

deviceIdentityPath: copilotIdentityPath,

prePairDevice: true,

browserOrigin: copilotOrigin,

});

const deviceToken = (pairedHello as { auth?: { deviceToken?: string } }).auth?.deviceToken;

expect(deviceToken).toBeTruthy();

pairingWs.close();

const wrongOrigin = await connectReq(wrongOriginWs, {

scopes: ["operator.read", "operator.write"],

caps: [...scopedCaps, GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS],

client: copilotClient,

deviceIdentityPath: copilotIdentityPath,

deviceToken,

skipDefaultAuth: true,

});

expect(wrongOrigin.ok).toBe(false);

expect(wrongOrigin.error?.code).toBe("NOT_PAIRED");

expect(wrongOrigin.error?.message).toContain("dedicated paired device identity");

wrongOriginWs.close();

await connectOk(mainWs, {

scopes: ["operator.read", "operator.write"],

caps: [...scopedCaps, GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS],

client: copilotClient,

deviceIdentityPath: copilotIdentityPath,

deviceToken,

skipDefaultAuth: true,

});

await connectOk(otherWs, { scopes: ["operator.read"], caps: scopedCaps });

await connectOk(legacyWs, { scopes: ["operator.read"] });

await rpcReq(mainWs, "sessions.messages.subscribe", { key: "main" });

await rpcReq(otherWs, "sessions.messages.subscribe", { key: "other" });

const mainEvent = onceMessage(

mainWs,

(message) =>

message.type === "event" &&

message.event === "chat" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey ===

"agent:main:main",

);

const legacyEvent = onceMessage(

legacyWs,

(message) =>

message.type === "event" &&

message.event === "chat" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey ===

"agent:main:main",

);

const otherReceived = onceMessage(

otherWs,

(message) => message.type === "event" && message.event === "chat",

300,

).then(

() => true,

() => false,

);

const send = await rpcReq(mainWs, "chat.send", {

sessionKey: "main",

message: "/status",

toolBindings: {

browser: {

kind: "tab",

tabId: 7,

target: "host",

profile: "chrome",

targetId: "target-7",

},

},

idempotencyKey: "scoped-delivery-proof",

});

expect(send.ok, JSON.stringify(send)).toBe(true);

await expect(Promise.all([mainEvent, legacyEvent])).resolves.toHaveLength(2);

await expect(otherReceived).resolves.toBe(false);

} finally {

unpairedWs.close();

pairingWs.close();

wrongOriginWs.close();

mainWs.close();

otherWs.close();

legacyWs.close();

}

});

test("denies browser-copilot the broad sessions.subscribe and blocks foreign session transcript delivery", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: { sessionId: "sess-main", updatedAt: Date.now() },

other: { sessionId: "sess-other", updatedAt: Date.now() },

},

storePath,

});

const copilotOrigin = "chrome-extension://abcdefghijklmnopabcdefghijklmnop";

const copilotClient = {

id: GATEWAY_CLIENT_IDS.BROWSER_COPILOT,

version: "test",

platform: "chrome",

deviceFamily: "extension",

mode: GATEWAY_CLIENT_MODES.UI,

};

const copilotIdentityPath = path.join(path.dirname(storePath), "copilot-repro-device.json");

const scopedCaps = [GATEWAY_CLIENT_CAPS.SESSION_SCOPED_EVENTS];

const pairingWs = await harness.openWs({ origin: copilotOrigin });

const copilotWs = await harness.openWs({ origin: copilotOrigin });

try {

const pairedHello = await connectOk(pairingWs, {

scopes: ["operator.read", "operator.write"],

caps: [...scopedCaps, GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS],

client: copilotClient,

deviceIdentityPath: copilotIdentityPath,

prePairDevice: true,

browserOrigin: copilotOrigin,

});

const deviceToken = (pairedHello as { auth?: { deviceToken?: string } }).auth?.deviceToken;

expect(deviceToken).toBeTruthy();

pairingWs.close();

// The NEW "session-scoped / per-tab" principal connects (dedicated paired identity+origin).

await connectOk(copilotWs, {

scopes: ["operator.read", "operator.write"],

caps: [...scopedCaps, GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS],

client: copilotClient,

deviceIdentityPath: copilotIdentityPath,

deviceToken,

skipDefaultAuth: true,

});

// The legacy broad sessions.subscribe must be rejected for a session-scoped principal so it

// cannot join the unscoped fanout that carries session.message/sessions.changed.

const broad = await rpcReq(copilotWs, "sessions.subscribe");

expect(broad.ok).toBe(false);

expect(broad.error?.message ?? "").toContain("sessions.messages.subscribe");

// A DIFFERENT session's live transcript must NOT reach this connection.

const foreignDelivery = onceMessage(

copilotWs,

(message) =>

message.type === "event" &&

message.event === "session.message" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey === "agent:main:other",

1500,

).then(

() => true,

() => false,

);

const appended = await appendAssistantMessageToSessionTranscript({

sessionKey: "agent:main:other",

text: "SECRET transcript that belongs to a different tab/session",

storePath,

});

expect(appended.ok).toBe(true);

await expect(foreignDelivery).resolves.toBe(false);

} finally {

pairingWs.close();

copilotWs.close();

}

});

test("rejects client identity changes across a dedicated copilot pairing", async () => {

const storePath = await createSessionStoreFile();

const copilotOrigin = "chrome-extension://abcdefghijklmnopabcdefghijklmnop";

const identityPath = path.join(path.dirname(storePath), "other-client-device.json");

const copilotIdentityPath = path.join(path.dirname(storePath), "copilot-paired-device.json");

const controlWs = await harness.openWs({ origin: `http://127.0.0.1:${harness.port}` });

const copilotWs = await harness.openWs({

origin: copilotOrigin,

});

const pairingWs = await harness.openWs({

origin: copilotOrigin,

});

const downgradeWs = await harness.openWs();

const wrongModeWs = await harness.openWs({

origin: "chrome-extension://abcdefghijklmnopabcdefghijklmnop",

});

const webOriginWs = await harness.openWs({ origin: `http://127.0.0.1:${harness.port}` });

const missingCapsWs = await harness.openWs({

origin: "chrome-extension://abcdefghijklmnopabcdefghijklmnop",

});

try {

const clientBase = {

version: "test",

platform: "chrome",

deviceFamily: "extension",

mode: GATEWAY_CLIENT_MODES.UI,

};

const wrongMode = await connectReq(wrongModeWs, {

caps: [GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS, GATEWAY_CLIENT_CAPS.SESSION_SCOPED_EVENTS],

client: {

...clientBase,

id: GATEWAY_CLIENT_IDS.BROWSER_COPILOT,

mode: GATEWAY_CLIENT_MODES.WEBCHAT,

},

deviceIdentityPath: path.join(path.dirname(storePath), "wrong-mode-device.json"),

prePairDevice: false,

scopes: ["operator.read", "operator.write"],

});

expect(wrongMode.ok).toBe(false);

expect(wrongMode.error?.message).toContain("requires ui mode");

wrongModeWs.close();

const missingCaps = await connectReq(missingCapsWs, {

caps: [GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS],

client: { ...clientBase, id: GATEWAY_CLIENT_IDS.BROWSER_COPILOT },

deviceIdentityPath: path.join(path.dirname(storePath), "missing-caps-device.json"),

prePairDevice: false,

scopes: ["operator.read", "operator.write"],

});

expect(missingCaps.ok).toBe(false);

expect(missingCaps.error?.message).toContain("session-scoped-events");

missingCapsWs.close();

const webOrigin = await connectReq(webOriginWs, {

caps: [GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS, GATEWAY_CLIENT_CAPS.SESSION_SCOPED_EVENTS],

client: { ...clientBase, id: GATEWAY_CLIENT_IDS.BROWSER_COPILOT },

deviceIdentityPath: path.join(path.dirname(storePath), "web-origin-device.json"),

prePairDevice: false,

scopes: ["operator.read", "operator.write"],

});

expect(webOrigin.ok).toBe(false);

expect(webOrigin.error?.message).toContain("canonical Chrome extension origin");

webOriginWs.close();

await connectOk(controlWs, {

client: { ...clientBase, id: GATEWAY_CLIENT_IDS.CONTROL_UI },

deviceIdentityPath: identityPath,

prePairDevice: true,

scopes: ["operator.read", "operator.write"],

});

controlWs.close();

const response = await connectReq(copilotWs, {

caps: [GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS, GATEWAY_CLIENT_CAPS.SESSION_SCOPED_EVENTS],

client: { ...clientBase, id: GATEWAY_CLIENT_IDS.BROWSER_COPILOT },

deviceIdentityPath: identityPath,

prePairDevice: false,

scopes: ["operator.read", "operator.write"],

});

expect(response.ok).toBe(false);

expect(response.error?.code).toBe("NOT_PAIRED");

expect(response.error?.message).toContain("dedicated paired device identity");

await connectOk(pairingWs, {

caps: [GATEWAY_CLIENT_CAPS.RUN_TOOL_BINDINGS, GATEWAY_CLIENT_CAPS.SESSION_SCOPED_EVENTS],

client: { ...clientBase, id: GATEWAY_CLIENT_IDS.BROWSER_COPILOT },

deviceIdentityPath: copilotIdentityPath,

prePairDevice: true,

browserOrigin: copilotOrigin,

scopes: ["operator.read", "operator.write"],

});

pairingWs.close();

const downgrade = await connectReq(downgradeWs, {

client: {

...clientBase,

id: GATEWAY_CLIENT_IDS.TEST,

mode: GATEWAY_CLIENT_MODES.TEST,

},

deviceIdentityPath: copilotIdentityPath,

prePairDevice: false,

scopes: ["operator.read", "operator.write"],

});

expect(downgrade.ok).toBe(false);

expect(downgrade.error?.code).toBe("NOT_PAIRED");

expect(downgrade.error?.message).toContain("dedicated paired device identity");

} finally {

controlWs.close();

copilotWs.close();

pairingWs.close();

downgradeWs.close();

wrongModeWs.close();

webOriginWs.close();

missingCapsWs.close();

}

});

test("includes spawned session ownership metadata on lifecycle sessions.changed events", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

child: {

sessionId: "sess-child",

updatedAt: Date.now(),

goal: {

schemaVersion: 1,

id: "goal-child",

objective: "Finish child work",

status: "active",

createdAt: 1,

updatedAt: 2,

tokenStart: 0,

tokensUsed: 42,

continuationTurns: 0,

},

spawnedBy: "agent:main:parent",

spawnedWorkspaceDir: "/tmp/subagent-workspace",

spawnedCwd: "/tmp/task-repo",

forkedFromParent: true,

spawnDepth: 2,

subagentRole: "orchestrator",

subagentControlScope: "children",

displayName: "Ops Child",

},

},

storePath,

});

await withOperatorSessionSubscriber(async (ws) => {

const changedEvent = onceMessage(

ws,

(message) =>

message.type === "event" &&

message.event === "sessions.changed" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey ===

"agent:main:child",

);

emitSessionLifecycleEvent({

sessionKey: "agent:main:child",

reason: "reactivated",

});

const event = await changedEvent;

expectRecordFields(event.payload, {

sessionKey: "agent:main:child",

reason: "reactivated",

spawnedBy: "agent:main:parent",

spawnedWorkspaceDir: "/tmp/subagent-workspace",

spawnedCwd: "/tmp/task-repo",

forkedFromParent: true,

spawnDepth: 2,

subagentRole: "orchestrator",

subagentControlScope: "children",

displayName: "Ops Child",

goal: {

schemaVersion: 1,

id: "goal-child",

objective: "Finish child work",

status: "active",

createdAt: 1,

updatedAt: 2,

tokenStart: 0,

tokenStartFresh: true,

tokensUsed: 42,

continuationTurns: 0,

},

});

});

});

test("only sends transcript events to subscribed operator clients", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: {

sessionId: "sess-main",

updatedAt: Date.now(),

},

},

storePath,

});

const subscribedWs = await harness.openWs();

const unsubscribedWs = await harness.openWs();

const nodeWs = await harness.openWs();

try {

await connectOk(subscribedWs, { scopes: ["operator.read"] });

await rpcReq(subscribedWs, "sessions.subscribe");

await connectOk(unsubscribedWs, { scopes: ["operator.read"] });

await connectOk(nodeWs, { role: "node", scopes: [] });

const subscribedEvent = onceMessage(

subscribedWs,

(message) =>

message.type === "event" &&

message.event === "session.message" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey ===

"agent:main:main",

);

const appended = await appendAssistantMessageToSessionTranscript({

sessionKey: "agent:main:main",

text: "subscribed only",

storePath,

});

expect(appended.ok).toBe(true);

const event = await subscribedEvent;

expectRecordFields(event, {

type: "event",

event: "session.message",

});

await expectNoMessageWithin({

watch: (timeoutMs) =>

onceMessage(

unsubscribedWs,

(message) => message.type === "event" && message.event === "session.message",

timeoutMs,

),

});

await expectNoMessageWithin({

watch: (timeoutMs) =>

onceMessage(

nodeWs,

(message) => message.type === "event" && message.event === "session.message",

timeoutMs,

),

});

} finally {

subscribedWs.close();

unsubscribedWs.close();

nodeWs.close();

}

});

test("broadcasts appended transcript messages with the session key", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: {

sessionId: "sess-main",

updatedAt: Date.now(),

},

},

storePath,

});

const emitSpy = vi.spyOn(transcriptEvents, "emitSessionTranscriptUpdate");

try {

const appended = await appendAssistantMessageToSessionTranscript({

sessionKey: "agent:main:main",

text: "live websocket message",

storePath,

});

expect(appended.ok).toBe(true);

if (!appended.ok) {

throw new Error(`append failed: ${appended.reason}`);

}

const emitParams = requireRecord(emitSpy.mock.calls.at(0)?.[0], "transcript update params");

expect(emitParams.sessionFile).toBe(appended.sessionFile);

expect(emitParams.sessionKey).toBe("agent:main:main");

expect(emitParams.messageId).toBe(appended.messageId);

expectRecordFields(emitParams.message, {

role: "assistant",

content: [{ type: "text", text: "live websocket message" }],

});

await expect(

loadTranscriptEvents({

agentId: "main",

sessionId: "sess-main",

sessionKey: "agent:main:main",

storePath,

}),

).resolves.toContainEqual(

expect.objectContaining({

message: expect.objectContaining({

content: [{ type: "text", text: "live websocket message" }],

}),

type: "message",

}),

);

} finally {

emitSpy.mockRestore();

}

});

test("strips blocked original content from live session.message events", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: {

sessionId: "sess-main",

updatedAt: Date.now(),

},

},

storePath,

});

const transcriptPath = path.join(path.dirname(storePath), "sess-main.jsonl");

await fs.writeFile(

transcriptPath,

JSON.stringify({ type: "session", version: 1, id: "sess-main" }) + "\n",

"utf-8",

);

await withOperatorSessionSubscriber(async (ws) => {

const { messageEvent } = await emitTranscriptUpdateAndCollectMessageEvent({

ws,

sessionKey: "agent:main:main",

sessionFile: transcriptPath,

messageId: "blocked-1",

message: {

role: "user",

content: [{ type: "text", text: "The agent cannot read this message." }],

__openclaw: {

beforeAgentRunBlocked: { blockedBy: "policy-plugin", blockedAt: 1 },

},

},

});

const payload = messageEvent.payload as {

message?: { content?: unknown; __openclaw?: { beforeAgentRunBlocked?: unknown } };

};

expect(payload.message?.content).toEqual([

{ type: "text", text: "The agent cannot read this message." },

]);

expect(JSON.stringify(payload.message)).not.toContain("secret blocked prompt");

expect(JSON.stringify(payload.message)).not.toContain("contains protected content");

});

});

test("broadcasts redacted blocked user appends to live session listeners", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: {

sessionId: "sess-main",

updatedAt: Date.now(),

},

},

storePath,

});

await withOperatorSessionSubscriber(async (ws) => {

const messageEventPromise = waitForSessionMessageEvent(ws, "agent:main:main");

emitSessionTranscriptUpdate({

sessionFile: path.join(path.dirname(storePath), "sess-main.jsonl"),

sessionKey: "agent:main:main",

messageId: "blocked-message",

message: {

role: "user",

content: [{ type: "text", text: "The agent cannot read this message." }],

__openclaw: {

beforeAgentRunBlocked: {

blockedBy: "policy-plugin",

blockedAt: Date.now(),

},

},

},

});

const messageEvent = await messageEventPromise;

const payload = messageEvent.payload as {

message?: {

role?: unknown;

content?: unknown;

__openclaw?: { beforeAgentRunBlocked?: unknown };

};

};

expect(payload.message?.role).toBe("user");

expect(payload.message?.content).toEqual([

{ type: "text", text: "The agent cannot read this message." },

]);

expect(JSON.stringify(payload.message)).not.toContain("secret blocked prompt");

expect(JSON.stringify(payload.message)).not.toContain("contains protected content");

});

});

test("does not broadcast hidden runtime-context custom messages as live chat messages", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

"hidden-runtime": {

sessionId: "sess-hidden-runtime",

updatedAt: Date.now(),

},

},

storePath,

});

await withOperatorSessionSubscriber(async (ws) => {

const changedEventPromise = waitForSessionsChangedMessagePhase(

ws,

"agent:main:hidden-runtime",

);

await expectNoMessageWithin({

watch: (timeoutMs) =>

onceMessage(

ws,

(message) =>

message.type === "event" &&

message.event === "session.message" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey ===

"agent:main:hidden-runtime",

timeoutMs,

),

action: () => {

emitSessionTranscriptUpdate({

sessionFile: path.join(path.dirname(storePath), "sess-hidden-runtime.jsonl"),

sessionKey: "agent:main:hidden-runtime",

messageId: "runtime-context-1",

messageSeq: 1,

message: {

role: "custom",

customType: "openclaw.runtime-context",

content: "secret runtime context",

display: false,

},

});

},

});

const changedEvent = await changedEventPromise;

expectRecordFields(changedEvent.payload, {

sessionKey: "agent:main:hidden-runtime",

phase: "message",

goal: null,

});

});

});

test("does not duplicate displayable transcript updates with sessions.changed", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: {

sessionId: "sess-main",

updatedAt: Date.now(),

},

},

storePath,

});

await withOperatorSessionSubscriber(async (ws) => {

const messageEventPromise = waitForSessionMessageEvent(ws, "agent:main:main");

await expectNoMessageWithin({

action: () => {

emitSessionTranscriptUpdate({

sessionFile: path.join(path.dirname(storePath), "sess-main.jsonl"),

sessionKey: "agent:main:main",

message: {

role: "assistant",

content: [{ type: "text", text: "single frame" }],

timestamp: Date.now(),

},

messageId: "msg-single-frame",

messageSeq: 1,

});

},

watch: (timeoutMs) =>

onceMessage(

ws,

(message) =>

message.type === "event" &&

message.event === "sessions.changed" &&

(message.payload as { phase?: string; sessionKey?: string } | undefined)?.phase ===

"message" &&

(message.payload as { sessionKey?: string } | undefined)?.sessionKey ===

"agent:main:main",

timeoutMs,

),

});

const messageEvent = await messageEventPromise;

expectRecordFields(messageEvent.payload, {

sessionKey: "agent:main:main",

messageId: "msg-single-frame",

messageSeq: 1,

});

});

});

test("broadcasts identity-only transcript updates to live session listeners", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: {

sessionId: "sess-main",

updatedAt: Date.now(),

},

},

storePath,

});

await withOperatorSessionSubscriber(async (ws) => {

const messageEventPromise = waitForSessionMessageEvent(ws, "agent:main:main");

emitSessionTranscriptUpdate({

target: {

agentId: "main",

sessionId: "sess-main",

sessionKey: "agent:main:main",

},

message: {

role: "assistant",

content: [{ type: "text", text: "identity frame" }],

timestamp: Date.now(),

},

messageId: "msg-identity-frame",

messageSeq: 1,

});

const messageEvent = await messageEventPromise;

expectRecordFields(messageEvent.payload, {

sessionKey: "agent:main:main",

messageId: "msg-identity-frame",

messageSeq: 1,

});

});

});

test("includes live usage metadata on session.message transcript events", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: {

sessionId: "sess-main",

updatedAt: Date.now(),

modelProvider: "openai",

model: "gpt-5.4",

contextTokens: 123_456,

totalTokens: 0,

totalTokensFresh: false,

},

},

storePath,

});

const transcriptMessage = {

role: "assistant",

content: [{ type: "text", text: "usage snapshot" }],

provider: "openai",

model: "gpt-5.4",

usage: {

input: 2_000,

output: 400,

cacheRead: 300,

cacheWrite: 100,

cost: { total: 0.0042 },

},

timestamp: Date.now(),

};

const turn = await persistSessionTranscriptTurn(

{

agentId: "main",

sessionId: "sess-main",

sessionKey: "agent:main:main",

storePath,

},

{

messages: [{ message: transcriptMessage }],

updateMode: "none",

},

);

await withOperatorSessionSubscriber(async (ws) => {

const { messageEvent } = await emitTranscriptUpdateAndCollectMessageEvent({

ws,

sessionKey: "agent:main:main",

sessionFile: turn.sessionFile,

message: transcriptMessage,

messageId: "msg-usage",

});

expectRecordFields(messageEvent.payload, {

sessionKey: "agent:main:main",

messageId: "msg-usage",

messageSeq: 1,

totalTokens: 2_400,

totalTokensFresh: true,

contextTokens: 123_456,

estimatedCostUsd: 0.0042,

modelProvider: "openai",

model: "gpt-5.4",

});

});

});

test("prefers carried transcript sequence for live session events", async () => {

const storePath = await createSessionStoreFile();

await writeSessionStore({

entries: {

main: {

sessionId: "sess-main",

updatedAt: Date.now(),

},

},

storePath,

});

await withOperatorSessionSubscriber(async (ws) => {

const { messageEvent } = await emitTranscriptUpdateAndCollectMessageEvent({

ws,

sessionKey: "agent:main:main",