diff --git a/extensions/clickclack/src/inbound.ts b/extensions/clickclack/src/inbound.ts index 40c90b22c4d0..1b02903449a6 100644 --- a/extensions/clickclack/src/inbound.ts +++ b/extensions/clickclack/src/inbound.ts @@ -233,6 +233,7 @@ export async function handleClickClackInbound(params: { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, }, @@ -276,7 +277,7 @@ export async function handleClickClackInbound(params: { cfg: params.config as OpenClawConfig, channel: CHANNEL_ID, accountId: params.account.accountId, - route: { agentId: route.agentId, sessionKey: route.sessionKey }, + route: { agentId: route.agentId, dmScope: route.dmScope, sessionKey: route.sessionKey }, ctxPayload, toolsAllow: params.account.toolsAllow, // Provenance stamping shares the agentActivity opt-in: with the flag off diff --git a/extensions/discord/src/monitor/message-handler.context.ts b/extensions/discord/src/monitor/message-handler.context.ts index 93b483564695..86c6448ae2ac 100644 --- a/extensions/discord/src/monitor/message-handler.context.ts +++ b/extensions/discord/src/monitor/message-handler.context.ts @@ -350,6 +350,7 @@ export async function buildDiscordMessageProcessContext(params: { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, dispatchSessionKey: effectiveSessionKey, diff --git a/extensions/feishu/src/bot.ts b/extensions/feishu/src/bot.ts index 59ca992cce36..79151d701e35 100644 --- a/extensions/feishu/src/bot.ts +++ b/extensions/feishu/src/bot.ts @@ -1382,6 +1382,7 @@ export async function handleFeishuMessage(params: { }, route: { agentId, + dmScope: route.dmScope, accountId: agentAccountId, routeSessionKey: agentSessionKey, }, diff --git a/extensions/feishu/src/comment-handler.ts b/extensions/feishu/src/comment-handler.ts index 0ebbc752a6b3..6714c1456d52 100644 --- a/extensions/feishu/src/comment-handler.ts +++ b/extensions/feishu/src/comment-handler.ts @@ -230,6 +230,7 @@ export async function handleFeishuCommentEvent( conversation: { kind: "direct", id: commentTarget, label: conversationLabel }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: commentSessionKey, dispatchSessionKey: commentSessionKey, diff --git a/extensions/googlechat/src/monitor.ts b/extensions/googlechat/src/monitor.ts index 70e6fa48efe6..00a787292056 100644 --- a/extensions/googlechat/src/monitor.ts +++ b/extensions/googlechat/src/monitor.ts @@ -303,6 +303,7 @@ async function processMessageWithPipeline(params: { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, }, diff --git a/extensions/imessage/src/monitor/inbound-processing.ts b/extensions/imessage/src/monitor/inbound-processing.ts index 60b2b8b4a69d..4d770319666a 100644 --- a/extensions/imessage/src/monitor/inbound-processing.ts +++ b/extensions/imessage/src/monitor/inbound-processing.ts @@ -1027,6 +1027,7 @@ export async function buildIMessageInboundContext(params: { }, route: { agentId: decision.route.agentId, + dmScope: decision.route.dmScope, accountId: decision.route.accountId, routeSessionKey: decision.route.sessionKey, }, diff --git a/extensions/irc/src/inbound.ts b/extensions/irc/src/inbound.ts index 755011afb938..8ed2f948d874 100644 --- a/extensions/irc/src/inbound.ts +++ b/extensions/irc/src/inbound.ts @@ -409,6 +409,7 @@ export async function handleIrcInbound(params: { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, }, diff --git a/extensions/line/src/bot-message-context.ts b/extensions/line/src/bot-message-context.ts index 59e625e65822..c459eae9a8ab 100644 --- a/extensions/line/src/bot-message-context.ts +++ b/extensions/line/src/bot-message-context.ts @@ -343,6 +343,7 @@ async function finalizeLineInboundContext(params: { From: fromAddress, To: toAddress, SessionKey: params.route.sessionKey, + DmScope: params.route.dmScope, AccountId: params.route.accountId, ChatType: params.source.isGroup ? "group" : "direct", ConversationLabel: conversationLabel, diff --git a/extensions/matrix/src/matrix/monitor/handler.ts b/extensions/matrix/src/matrix/monitor/handler.ts index 64f434296f4e..9339ecee1114 100644 --- a/extensions/matrix/src/matrix/monitor/handler.ts +++ b/extensions/matrix/src/matrix/monitor/handler.ts @@ -1627,6 +1627,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam }, route: { agentId: _route.agentId, + dmScope: _route.dmScope, accountId: _route.accountId, routeSessionKey: _route.sessionKey, }, diff --git a/extensions/mattermost/src/mattermost/monitor.ts b/extensions/mattermost/src/mattermost/monitor.ts index 3b1bcb2e64df..255d641b25a8 100644 --- a/extensions/mattermost/src/mattermost/monitor.ts +++ b/extensions/mattermost/src/mattermost/monitor.ts @@ -449,6 +449,7 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} : `mattermost:channel:${optsLocal.channelId}`, To: to, SessionKey: threadContext.sessionKey, + DmScope: route.dmScope, ParentSessionKey: threadContext.parentSessionKey, AccountId: route.accountId, ChatType: chatType, @@ -640,6 +641,7 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} : `mattermost:channel:${params.channelId}`, To: to, SessionKey: params.sessionKey, + DmScope: params.route.dmScope, ParentSessionKey: params.parentSessionKey, AccountId: params.route.accountId, ChatType: params.chatType, @@ -1334,6 +1336,7 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} : `mattermost:channel:${channelId}`, To: to, SessionKey: sessionKey, + DmScope: route.dmScope, ParentSessionKey: parentSessionKey, AccountId: route.accountId, ChatType: chatType, @@ -1698,7 +1701,11 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} cfg, channel: "mattermost", accountId: route.accountId, - route: { agentId: route.agentId, sessionKey: route.sessionKey }, + route: { + agentId: route.agentId, + dmScope: route.dmScope, + sessionKey: route.sessionKey, + }, ctxPayload, record: { updateLastRoute: diff --git a/extensions/msteams/src/monitor-handler/message-handler.ts b/extensions/msteams/src/monitor-handler/message-handler.ts index f3e2783d9d4c..d78159791bc5 100644 --- a/extensions/msteams/src/monitor-handler/message-handler.ts +++ b/extensions/msteams/src/monitor-handler/message-handler.ts @@ -879,6 +879,7 @@ export function createMSTeamsMessageHandler(deps: MSTeamsMessageHandlerDeps) { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, }, diff --git a/extensions/nextcloud-talk/src/inbound.ts b/extensions/nextcloud-talk/src/inbound.ts index 0c70b559944f..8d98079211ef 100644 --- a/extensions/nextcloud-talk/src/inbound.ts +++ b/extensions/nextcloud-talk/src/inbound.ts @@ -338,6 +338,7 @@ export async function handleNextcloudTalkInbound(params: { conversation: { kind: isGroup ? "group" : "direct", id: roomToken, label: fromLabel }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, }, diff --git a/extensions/qa-channel/src/inbound.ts b/extensions/qa-channel/src/inbound.ts index 7af6a079e413..5e772852ba1d 100644 --- a/extensions/qa-channel/src/inbound.ts +++ b/extensions/qa-channel/src/inbound.ts @@ -327,6 +327,7 @@ export async function handleQaInbound(params: { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: sessionKey, dispatchSessionKey: sessionKey, @@ -361,7 +362,7 @@ export async function handleQaInbound(params: { cfg: params.config as OpenClawConfig, channel: params.channelId, accountId: params.account.accountId, - route: { agentId: route.agentId, sessionKey: route.sessionKey }, + route: { agentId: route.agentId, dmScope: route.dmScope, sessionKey: route.sessionKey }, ctxPayload, delivery: { deliver: async (payload, info) => { diff --git a/extensions/qqbot/src/engine/gateway/inbound-context.ts b/extensions/qqbot/src/engine/gateway/inbound-context.ts index 0dc44e593002..2d6c2fbc8bec 100644 --- a/extensions/qqbot/src/engine/gateway/inbound-context.ts +++ b/extensions/qqbot/src/engine/gateway/inbound-context.ts @@ -34,7 +34,12 @@ export interface InboundGroupInfo { export interface InboundContext { event: QueuedMessage; - route: { sessionKey: string; accountId: string; agentId?: string }; + route: { + sessionKey: string; + accountId: string; + agentId?: string; + dmScope?: "main" | "per-peer" | "per-channel-peer" | "per-account-channel-peer"; + }; isGroupChat: boolean; peerId: string; qualifiedTarget: string; diff --git a/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts b/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts index b2ba8fe158d4..a18fbcddc3d1 100644 --- a/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts +++ b/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts @@ -437,7 +437,11 @@ export async function dispatchOutbound( cfg: openClawCfg, channel: "qqbot", accountId: inbound.route.accountId, - route: { agentId: routeAgentId, sessionKey: inbound.route.sessionKey }, + route: { + agentId: routeAgentId, + dmScope: inbound.route.dmScope, + sessionKey: inbound.route.sessionKey, + }, ctxPayload, record: { onRecordError: (err: unknown) => { @@ -774,6 +778,7 @@ async function buildCtxPayload( }, route: { agentId: inbound.route.agentId ?? resolveDefaultAgentId(cfg as OpenClawConfig), + dmScope: inbound.route.dmScope, routeSessionKey: inbound.route.sessionKey, accountId: inbound.route.accountId, }, diff --git a/extensions/qqbot/src/engine/gateway/stages/access-stage.ts b/extensions/qqbot/src/engine/gateway/stages/access-stage.ts index 88f1762da817..07a7b1d04fb3 100644 --- a/extensions/qqbot/src/engine/gateway/stages/access-stage.ts +++ b/extensions/qqbot/src/engine/gateway/stages/access-stage.ts @@ -11,7 +11,7 @@ type AccessStageResult = peerId: string; qualifiedTarget: string; fromAddress: string; - route: { sessionKey: string; accountId: string; agentId?: string }; + route: InboundContext["route"]; access: QQBotInboundAccess; } | { kind: "block"; context: InboundContext }; diff --git a/extensions/qqbot/src/engine/gateway/stages/stub-contexts.ts b/extensions/qqbot/src/engine/gateway/stages/stub-contexts.ts index f095b2ba9ce5..f9eee1d1cb56 100644 --- a/extensions/qqbot/src/engine/gateway/stages/stub-contexts.ts +++ b/extensions/qqbot/src/engine/gateway/stages/stub-contexts.ts @@ -6,7 +6,7 @@ import type { TypingKeepAlive } from "../typing-keepalive.js"; interface BaseStubFields { event: QueuedMessage; - route: { sessionKey: string; accountId: string; agentId?: string }; + route: InboundContext["route"]; isGroupChat: boolean; peerId: string; qualifiedTarget: string; diff --git a/extensions/qqbot/src/engine/gateway/types.ts b/extensions/qqbot/src/engine/gateway/types.ts index 147d7065783a..db3377e4a6c9 100644 --- a/extensions/qqbot/src/engine/gateway/types.ts +++ b/extensions/qqbot/src/engine/gateway/types.ts @@ -21,7 +21,12 @@ export interface GatewayPluginRuntime { channel: string; accountId: string; peer: { kind: "group" | "direct"; id: string }; - }) => { sessionKey: string; accountId: string; agentId?: string }; + }) => { + sessionKey: string; + accountId: string; + agentId?: string; + dmScope?: "main" | "per-peer" | "per-channel-peer" | "per-account-channel-peer"; + }; }; commands?: { isControlCommandMessage?: (text?: string, cfg?: unknown) => boolean; diff --git a/extensions/signal/src/monitor/event-handler.ts b/extensions/signal/src/monitor/event-handler.ts index e32e3e87b915..ad6b18b9d664 100644 --- a/extensions/signal/src/monitor/event-handler.ts +++ b/extensions/signal/src/monitor/event-handler.ts @@ -332,6 +332,7 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, }, diff --git a/extensions/slack/src/monitor/message-handler/prepare.ts b/extensions/slack/src/monitor/message-handler/prepare.ts index 0c9a2dca6354..1c4e84ebc095 100644 --- a/extensions/slack/src/monitor/message-handler/prepare.ts +++ b/extensions/slack/src/monitor/message-handler/prepare.ts @@ -1512,6 +1512,7 @@ export async function prepareSlackMessage(params: { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: sessionKey, parentSessionKey: threadKeys.parentSessionKey, diff --git a/extensions/synology-chat/src/inbound-event.ts b/extensions/synology-chat/src/inbound-event.ts index 4e12b249410b..0ab1a3461bea 100644 --- a/extensions/synology-chat/src/inbound-event.ts +++ b/extensions/synology-chat/src/inbound-event.ts @@ -108,6 +108,7 @@ export async function dispatchSynologyChatInboundEvent(params: { }, route: { agentId: resolved.route.agentId, + dmScope: resolved.route.dmScope, accountId: params.account.accountId, routeSessionKey: resolved.sessionKey, dispatchSessionKey: resolved.sessionKey, @@ -131,6 +132,7 @@ export async function dispatchSynologyChatInboundEvent(params: { accountId: params.account.accountId, route: { agentId: resolved.route.agentId, + dmScope: resolved.route.dmScope, sessionKey: resolved.route.sessionKey, }, ctxPayload: msgCtx, diff --git a/extensions/telegram/src/bot-message-context.session.ts b/extensions/telegram/src/bot-message-context.session.ts index c97e468d6026..e244ff77a3f7 100644 --- a/extensions/telegram/src/bot-message-context.session.ts +++ b/extensions/telegram/src/bot-message-context.session.ts @@ -554,6 +554,7 @@ export async function buildTelegramInboundContextPayload(params: { }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, mainSessionKey: route.mainSessionKey, diff --git a/extensions/tlon/src/monitor/index.ts b/extensions/tlon/src/monitor/index.ts index c767f1d0ea56..06907d7044dc 100644 --- a/extensions/tlon/src/monitor/index.ts +++ b/extensions/tlon/src/monitor/index.ts @@ -530,6 +530,7 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise ({ diff --git a/extensions/whatsapp/src/auto-reply/monitor/inbound-dispatch.ts b/extensions/whatsapp/src/auto-reply/monitor/inbound-dispatch.ts index 9f6e9460cc5c..5e41e8c6cabe 100644 --- a/extensions/whatsapp/src/auto-reply/monitor/inbound-dispatch.ts +++ b/extensions/whatsapp/src/auto-reply/monitor/inbound-dispatch.ts @@ -368,6 +368,7 @@ export async function buildWhatsAppInboundContext(params: { }, route: { agentId: params.route.agentId, + dmScope: params.route.dmScope, accountId: params.route.accountId, routeSessionKey: params.route.sessionKey, }, diff --git a/extensions/zalo/src/monitor.ts b/extensions/zalo/src/monitor.ts index f23d8d7a609e..de3968b15f7b 100644 --- a/extensions/zalo/src/monitor.ts +++ b/extensions/zalo/src/monitor.ts @@ -615,6 +615,7 @@ async function processMessageWithPipeline(params: ZaloMessagePipelineParams): Pr }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, }, @@ -677,7 +678,7 @@ async function processMessageWithPipeline(params: ZaloMessagePipelineParams): Pr cfg: config, channel: "zalo", accountId: account.accountId, - route: { agentId: route.agentId, sessionKey: route.sessionKey }, + route: { agentId: route.agentId, dmScope: route.dmScope, sessionKey: route.sessionKey }, ctxPayload, delivery: { preparePayload: (payload) => diff --git a/extensions/zalouser/src/monitor.ts b/extensions/zalouser/src/monitor.ts index 39be33374e91..28474f1094f1 100644 --- a/extensions/zalouser/src/monitor.ts +++ b/extensions/zalouser/src/monitor.ts @@ -577,6 +577,7 @@ async function processMessage( }, route: { agentId: route.agentId, + dmScope: route.dmScope, accountId: route.accountId, routeSessionKey: route.sessionKey, dispatchSessionKey: route.sessionKey, @@ -626,7 +627,7 @@ async function processMessage( channel: "zalouser", accountId: account.accountId, cfg: config, - route: { agentId: route.agentId, sessionKey: route.sessionKey }, + route: { agentId: route.agentId, dmScope: route.dmScope, sessionKey: route.sessionKey }, ctxPayload, delivery: { preparePayload: (payload) => { diff --git a/src/agents/openclaw-tools.session-status.test.ts b/src/agents/openclaw-tools.session-status.test.ts index dfa556550f78..7e87fe01d1dc 100644 --- a/src/agents/openclaw-tools.session-status.test.ts +++ b/src/agents/openclaw-tools.session-status.test.ts @@ -38,6 +38,9 @@ const resolveUsableCustomProviderApiKeyMock = vi.hoisted(() => const getSessionStateVersionMock = vi.hoisted(() => vi.fn((_sessionKey: string, _agentId: string) => 0), ); +const listAmbientGroupWatchTargetsMock = vi.hoisted(() => + vi.fn((_watcherSessionKey: string) => new Set()), +); const listSessionStateEventsSinceMock = vi.hoisted(() => vi.fn((_sessionKey: string, _agentId: string, _after: number, _limit: number) => ({ events: [] as Array>, @@ -357,6 +360,8 @@ vi.mock("../tasks/task-owner-access.js", () => ({ vi.mock("../sessions/session-state-events.js", () => ({ getSessionStateVersion: (sessionKey: string, agentId: string) => getSessionStateVersionMock(sessionKey, agentId), + listAmbientGroupWatchTargets: (watcherSessionKey: string) => + listAmbientGroupWatchTargetsMock(watcherSessionKey), listSessionStateEventsSince: ( sessionKey: string, agentId: string, @@ -396,6 +401,8 @@ function resetSessionStore(store: Record) { listTasksForRelatedSessionKeyForOwnerMock.mockReturnValue([]); getSessionStateVersionMock.mockReset(); getSessionStateVersionMock.mockReturnValue(0); + listAmbientGroupWatchTargetsMock.mockReset(); + listAmbientGroupWatchTargetsMock.mockReturnValue(new Set()); listSessionStateEventsSinceMock.mockReset(); listSessionStateEventsSinceMock.mockReturnValue({ events: [], @@ -568,6 +575,48 @@ describe("session_status tool", () => { expect(text).toContain('"kind": "upstream_missing"'); }); + it("returns watched group changesSince under tree visibility", async () => { + const groupSessionKey = "agent:main:telegram:group:watched"; + resetSessionStore({ + "agent:main:main": { sessionId: "s-main", updatedAt: 10 }, + [groupSessionKey]: { + sessionId: "s-group", + updatedAt: 20, + chatType: "group", + }, + }); + mockConfig = { + ...createMockConfig(), + tools: { + sessions: { visibility: "tree" }, + agentToAgent: { enabled: false }, + }, + }; + listAmbientGroupWatchTargetsMock.mockReturnValue(new Set([groupSessionKey])); + getSessionStateVersionMock.mockReturnValue(9); + listSessionStateEventsSinceMock.mockReturnValue({ + events: [ + { sequence: 9, kind: "human_direct_message", summary: "human message via telegram" }, + ], + truncated: false, + earliestAvailableSequence: 9, + historyGap: false, + }); + + const result = await getSessionStatusTool("agent:main:main").execute("call-group-state", { + sessionKey: groupSessionKey, + changesSince: 4, + }); + + expect(listAmbientGroupWatchTargetsMock).toHaveBeenCalledWith("agent:main:main"); + expect(listSessionStateEventsSinceMock).toHaveBeenCalledWith(groupSessionKey, "main", 4, 200); + expect(result.details).toMatchObject({ + ok: true, + sessionKey: groupSessionKey, + stateVersion: 9, + }); + }); + it("enables transcript usage fallback for session_status", async () => { resetSessionStore({ main: { diff --git a/src/agents/tools/sessions-access.test.ts b/src/agents/tools/sessions-access.test.ts index 93ed360ecf0c..467974bff978 100644 --- a/src/agents/tools/sessions-access.test.ts +++ b/src/agents/tools/sessions-access.test.ts @@ -1,6 +1,7 @@ // Sessions access tests cover session-tool visibility policy, sandbox clamps, // and agent-to-agent allow rules. import { describe, expect, it, vi } from "vitest"; +import { cleanupTempDirs, makeTempDir } from "../../../test/helpers/temp-dir.js"; import type { OpenClawConfig } from "../../config/config.js"; import { createAgentToAgentPolicy, @@ -10,6 +11,12 @@ import { resolveSandboxSessionToolsVisibility, resolveSessionToolsVisibility, } from "../../plugin-sdk/session-visibility.js"; +import { + listAmbientGroupWatchTargets, + registerMainSessionGroupWatch, + registerSessionStateWatch, +} from "../../sessions/session-state-events.js"; +import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db.js"; import { resolveSandboxedSessionToolContext } from "./sessions-access.js"; import { testing as sessionsResolutionTesting } from "./sessions-resolution.test-support.js"; @@ -225,6 +232,80 @@ describe("createAgentToAgentPolicy", () => { }); describe("createSessionVisibilityGuard", () => { + it("allows watched group reads under tree while denying unwatched peers", () => { + const tempDirs: string[] = []; + const stateDir = makeTempDir(tempDirs, "openclaw-session-visibility-"); + closeOpenClawStateDatabaseForTest(); + vi.stubEnv("OPENCLAW_STATE_DIR", stateDir); + try { + const requesterSessionKey = "agent:main:main"; + const watchedSessionKey = "agent:main:telegram:group:watched"; + expect( + registerMainSessionGroupWatch({ + sessionKey: watchedSessionKey, + agentId: "main", + entry: { sessionId: "watched", updatedAt: Date.now(), chatType: "group" }, + dmScope: "main", + }), + ).toBe(true); + const crossAgentSessionKey = "agent:ops:telegram:group:watched"; + registerSessionStateWatch({ + watcherSessionKey: requesterSessionKey, + targetSessionKey: crossAgentSessionKey, + }); + const explicitDirectSessionKey = "agent:main:coordinator"; + registerSessionStateWatch({ + watcherSessionKey: requesterSessionKey, + targetSessionKey: explicitDirectSessionKey, + }); + expect(listAmbientGroupWatchTargets(requesterSessionKey)).toEqual( + new Set([watchedSessionKey]), + ); + const guard = createSessionVisibilityRowChecker({ + action: "history", + requesterSessionKey, + visibility: "tree", + a2aPolicy: createAgentToAgentPolicy({} as OpenClawConfig), + }); + + expect(guard.check({ key: watchedSessionKey })).toEqual({ allowed: true }); + expect(guard.check({ key: "agent:main:telegram:group:unwatched" })).toEqual({ + allowed: false, + status: "forbidden", + error: + "Session history visibility is restricted to the current session tree (tools.sessions.visibility=tree).", + }); + expect(guard.check({ key: explicitDirectSessionKey })).toEqual({ + allowed: false, + status: "forbidden", + error: + "Session history visibility is restricted to the current session tree (tools.sessions.visibility=tree).", + }); + expect(guard.check({ key: crossAgentSessionKey })).toEqual({ + allowed: false, + status: "forbidden", + error: + "Session history visibility is restricted. Set tools.sessions.visibility=all and tools.agentToAgent.enabled=true to allow cross-agent access; use tools.agentToAgent.allow to restrict permitted agent pairs.", + }); + const sendGuard = createSessionVisibilityRowChecker({ + action: "send", + requesterSessionKey, + visibility: "tree", + a2aPolicy: createAgentToAgentPolicy({} as OpenClawConfig), + }); + expect(sendGuard.check({ key: watchedSessionKey })).toEqual({ + allowed: false, + status: "forbidden", + error: + "Session send visibility is restricted to the current session tree (tools.sessions.visibility=tree).", + }); + } finally { + closeOpenClawStateDatabaseForTest(); + vi.unstubAllEnvs(); + cleanupTempDirs(tempDirs); + } + }); + it("allows cross-agent spawned child rows in list results with tree visibility", () => { const guard = createSessionVisibilityRowChecker({ action: "list", diff --git a/src/agents/tools/sessions-list-tool.test.ts b/src/agents/tools/sessions-list-tool.test.ts index 64608e79540f..714c168c353f 100644 --- a/src/agents/tools/sessions-list-tool.test.ts +++ b/src/agents/tools/sessions-list-tool.test.ts @@ -29,6 +29,7 @@ vi.mock("../../gateway/call.js", () => ({ vi.mock("../../sessions/session-state-events.js", () => ({ getSessionStateVersions: (refs: Array<{ sessionKey: string; agentId: string }>) => mocks.getSessionStateVersions(refs), + listAmbientGroupWatchTargets: () => new Set(), })); vi.mock("./sessions-helpers.js", async (importActual) => { diff --git a/src/auto-reply/reply/session.ts b/src/auto-reply/reply/session.ts index c360da5dbc24..4ad2f842f032 100644 --- a/src/auto-reply/reply/session.ts +++ b/src/auto-reply/reply/session.ts @@ -77,6 +77,10 @@ import { interruptSessionWorkAdmissions, runExclusiveSessionLifecycleMutation, } from "../../sessions/session-lifecycle-admission.js"; +import { + classifySessionStateActor, + registerMainSessionGroupWatch, +} from "../../sessions/session-state-events.js"; import { normalizeSessionDeliveryFields } from "../../utils/delivery-context.shared.js"; import { resolveCommandTurnTargetSessionKey } from "../command-turn-context.js"; import { normalizeCommandBody } from "../commands-registry.js"; @@ -1069,6 +1073,17 @@ async function initSessionStateAttemptLocked( } sessionEntry = committed.sessionEntry; sessionId = sessionEntry.sessionId; + if ( + !isSystemEvent && + classifySessionStateActor({ inputProvenance: ctx.InputProvenance }).actorType === "human" + ) { + registerMainSessionGroupWatch({ + sessionKey, + agentId, + entry: sessionEntry, + dmScope: ctx.DmScope ?? sessionCfg?.dmScope ?? "main", + }); + } const sessionStore = committed.sessionStoreView; const sessionEntryHandle = createReplySessionEntryHandle({ sessionEntry, diff --git a/src/auto-reply/templating.ts b/src/auto-reply/templating.ts index 1fb4f392390b..f41a55808f23 100644 --- a/src/auto-reply/templating.ts +++ b/src/auto-reply/templating.ts @@ -1,6 +1,6 @@ /** Shared inbound message context types used by prompt templating and reply dispatch. */ import type { InboundEventKind } from "../channels/inbound-event/kind.js"; -import type { ReplyToMode } from "../config/types.base.js"; +import type { DmScope, ReplyToMode } from "../config/types.base.js"; import type { MediaUnderstandingDecision, MediaUnderstandingOutput, @@ -116,6 +116,8 @@ export type MsgContext = { * id, such as selected-agent global sessions. */ AgentId?: string; + /** Effective routed DM scope, including binding overrides. */ + DmScope?: DmScope; /** * Session-like key used for runtime policy (sandbox/tool policy) when the * conversation key intentionally remains broader, such as a main-session DM. diff --git a/src/channels/inbound-event/context.test.ts b/src/channels/inbound-event/context.test.ts index 92856c6e55a3..46a223c464cb 100644 --- a/src/channels/inbound-event/context.test.ts +++ b/src/channels/inbound-event/context.test.ts @@ -128,6 +128,7 @@ describe("buildChannelInboundEventContext", () => { }, route: { agentId: "main", + dmScope: "main", accountId: "acct", routeSessionKey: "agent:main:test:group:room-1", parentSessionKey: "agent:main:test:group", @@ -208,6 +209,7 @@ describe("buildChannelInboundEventContext", () => { To: "test:room:room-1", SessionKey: "agent:main:test:group:room-1", AgentId: "main", + DmScope: "main", AccountId: "acct", ParentSessionKey: "agent:main:test:group", ModelParentSessionKey: "agent:main:test:model", diff --git a/src/channels/inbound-event/context.ts b/src/channels/inbound-event/context.ts index 96ad103108be..67c4dcb13338 100644 --- a/src/channels/inbound-event/context.ts +++ b/src/channels/inbound-event/context.ts @@ -517,6 +517,7 @@ export function buildChannelInboundEventContext( To: params.reply.to, SessionKey: params.route.dispatchSessionKey ?? params.route.routeSessionKey, AgentId: params.route.agentId, + DmScope: params.route.dmScope, AccountId: params.route.accountId ?? params.accountId, ParentSessionKey: params.route.parentSessionKey, ModelParentSessionKey: params.route.modelParentSessionKey, diff --git a/src/channels/turn/kernel.test.ts b/src/channels/turn/kernel.test.ts index 6a1b145b337e..041d3745dccf 100644 --- a/src/channels/turn/kernel.test.ts +++ b/src/channels/turn/kernel.test.ts @@ -1374,7 +1374,11 @@ describe("channel turn kernel", () => { resolveTurn: () => ({ cfg, channel: "test", - route: { agentId: "observer", sessionKey: "agent:observer:test:peer" }, + route: { + agentId: "observer", + dmScope: "per-channel-peer", + sessionKey: "agent:observer:test:peer", + }, ctxPayload: createCtx({ SessionKey: "agent:observer:test:peer" }), delivery: { deliver }, record: { @@ -1391,6 +1395,11 @@ describe("channel turn kernel", () => { }); expect(result.dispatched).toBe(true); expect(events).toEqual(["record", "dispatch"]); + expect(dispatchReplyWithBufferedBlockDispatcherCore).toHaveBeenCalledWith( + expect.objectContaining({ + ctx: expect.objectContaining({ DmScope: "per-channel-peer" }), + }), + ); expect(deliver).not.toHaveBeenCalled(); if (!result.dispatched) { throw new Error("expected dispatch"); diff --git a/src/channels/turn/lifecycle.ts b/src/channels/turn/lifecycle.ts index dac060dfd61f..3f6f0b810331 100644 --- a/src/channels/turn/lifecycle.ts +++ b/src/channels/turn/lifecycle.ts @@ -29,6 +29,7 @@ export function assembleResolvedChannelTurn( const { cfg, route, ...turn } = value; return { ...turn, + ctxPayload: route.dmScope ? { ...turn.ctxPayload, DmScope: route.dmScope } : turn.ctxPayload, routeSessionKey: route.sessionKey, storePath: resolveStorePath(cfg.session?.store, { agentId: route.agentId }), recordInboundSession, @@ -37,6 +38,7 @@ export function assembleResolvedChannelTurn( const { cfg, route, ...turn } = value; return { ...turn, + ctxPayload: route.dmScope ? { ...turn.ctxPayload, DmScope: route.dmScope } : turn.ctxPayload, cfg, agentId: route.agentId, routeSessionKey: route.sessionKey, diff --git a/src/channels/turn/types.ts b/src/channels/turn/types.ts index 7c054aa09881..d8052232c4ec 100644 --- a/src/channels/turn/types.ts +++ b/src/channels/turn/types.ts @@ -17,6 +17,7 @@ import type { SupplementalContextFacts, } from "../../auto-reply/templating.js"; import type { GroupKeyResolution } from "../../config/sessions/types.js"; +import type { DmScope } from "../../config/types.base.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { DeliverOutboundPayloadsParams, @@ -85,6 +86,7 @@ export type ConversationFacts = { /** Session routing facts derived before dispatch. */ export type RouteFacts = { agentId: string; + dmScope?: DmScope; accountId?: string; routeSessionKey: string; dispatchSessionKey?: string; @@ -308,6 +310,7 @@ export type PreparedChannelTurn = { type ChannelTurnRoute = { agentId: string; + dmScope?: DmScope; sessionKey: string; }; diff --git a/src/gateway/server-methods/session-catalog.test.ts b/src/gateway/server-methods/session-catalog.test.ts index 1b5e5970bca3..f0be173be775 100644 --- a/src/gateway/server-methods/session-catalog.test.ts +++ b/src/gateway/server-methods/session-catalog.test.ts @@ -23,6 +23,7 @@ vi.mock("../../plugins/runtime.js", () => ({ })); vi.mock("../../sessions/session-state-events.js", () => ({ + listAmbientGroupWatchTargets: () => new Set(), recordSessionStateEvent: hoisted.recordSessionStateEvent, })); diff --git a/src/plugin-sdk/session-visibility.ts b/src/plugin-sdk/session-visibility.ts index 090dd23ddd4b..12a4fe22a976 100644 --- a/src/plugin-sdk/session-visibility.ts +++ b/src/plugin-sdk/session-visibility.ts @@ -7,6 +7,7 @@ import { normalizeTrimmedStringList } from "../../packages/normalization-core/sr import type { OpenClawConfig } from "../config/types.openclaw.js"; import { callGateway as defaultCallGateway } from "../gateway/call.js"; import { resolveAgentIdFromSessionKey } from "../routing/session-key.js"; +import { listAmbientGroupWatchTargets } from "../sessions/session-state-events.js"; type GatewayCaller = typeof defaultCallGateway; @@ -316,13 +317,24 @@ export function createSessionVisibilityRowChecker(params: { const requesterAgentId = normalizeLowercaseStringOrEmpty(params.requesterAgentId) || resolveAgentIdFromSessionKey(params.requesterSessionKey); + let watchedSessionKeys: Set | undefined; const check = (row: SessionVisibilityRow): SessionAccessResult => { const targetSessionKey = row.key; const targetAgentId = row.agentId ?? resolveAgentIdFromSessionKey(targetSessionKey); const isRequesterSession = targetSessionKey === params.requesterSessionKey || targetSessionKey === "current"; - const isRequesterOwned = rowOwnedByRequester(row, params.requesterSessionKey); + // Only a durable ambient-group marker makes the paired target + // ownership-equivalent for same-agent reads. Explicit A2A watches, absent + // markers, send access, and cross-agent targets remain fail-closed. + const isWatchedRead = + params.action !== "send" && + params.visibility === "tree" && + targetAgentId === requesterAgentId && + (watchedSessionKeys ??= listAmbientGroupWatchTargets(params.requesterSessionKey)).has( + targetSessionKey, + ); + const isRequesterOwned = rowOwnedByRequester(row, params.requesterSessionKey) || isWatchedRead; // Row ownership is stronger than agent ids: ACP children may use a backend // agent id while still belonging to the requester that spawned them. if ( diff --git a/src/sessions/session-state-events.test.ts b/src/sessions/session-state-events.test.ts index 89fb49bf5abc..9babf4ea8007 100644 --- a/src/sessions/session-state-events.test.ts +++ b/src/sessions/session-state-events.test.ts @@ -19,6 +19,7 @@ import { getSessionStateVersion, handleSessionStateSessionDeleted, handleSessionStateSessionReset, + listAmbientGroupWatchTargets, listSessionStateEventsSince, recordSessionCompacted, recordSessionGoalChanged, @@ -26,6 +27,7 @@ import { recordSessionStateEvent, recordSubagentSpawned, recordSubagentTerminalState, + registerMainSessionGroupWatch, registerSessionStateWatch, sweepSessionStateWatchNotices, } from "./session-state-events.js"; @@ -36,6 +38,7 @@ const tempDirs: string[] = []; const watcher = "agent:main:main"; const nestedWatcher = "agent:main:subagent:parent"; const child = "agent:main:subagent:child"; +const group = "agent:main:telegram:group:room-1"; const cfg = {} as OpenClawConfig; let disposeHeartbeatWakeHandler: (() => void) | undefined; @@ -63,6 +66,7 @@ function eventInput( function readCursor( database: ReturnType, watcherSessionKey = watcher, + targetSessionKey = child, ) { return openOpenClawStateDatabase(database) .db.prepare( @@ -70,7 +74,7 @@ function readCursor( FROM session_watch_cursors WHERE watcher_session_key = ? AND target_session_key = ?`, ) - .get(watcherSessionKey, child) as + .get(watcherSessionKey, targetSessionKey) as | { last_seen_sequence: number; notified_sequence: number; @@ -527,6 +531,213 @@ describe("session state events", () => { }); }); + it("registers the main watcher once only under dmScope main", () => { + const database = createDatabaseOptions(); + expect( + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "main" }, + { ...database, now: 100 }, + ), + ).toBe(true); + expect( + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "main" }, + { ...database, now: 200 }, + ), + ).toBe(true); + expect( + registerMainSessionGroupWatch( + { + sessionKey: "agent:main:slack:channel:room-2", + agentId: "main", + dmScope: "per-channel-peer", + }, + database, + ), + ).toBe(false); + + const rows = openOpenClawStateDatabase(database) + .db.prepare( + `SELECT watcher_session_key, target_session_key, updated_at + FROM session_watch_cursors + WHERE watcher_session_key = ?`, + ) + .all(watcher); + expect(rows).toEqual([ + { + watcher_session_key: watcher, + target_session_key: group, + updated_at: 100, + }, + ]); + expect(listAmbientGroupWatchTargets(watcher, database)).toEqual(new Set([group])); + openOpenClawStateDatabase(database) + .db.prepare( + `DELETE FROM session_watch_cursors + WHERE watcher_session_key = ? AND target_session_key = ?`, + ) + .run(watcher, group); + expect(listAmbientGroupWatchTargets(watcher, database)).toEqual(new Set()); + }); + + it("revokes ambient watches outside main scope but preserves explicit watches", () => { + const database = createDatabaseOptions(); + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "main" }, + database, + ); + expect( + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "per-channel-peer" }, + database, + ), + ).toBe(false); + expect(listAmbientGroupWatchTargets(watcher, database)).toEqual(new Set()); + expect(readCursor(database, watcher, group)).toBeUndefined(); + + registerSessionStateWatch({ watcherSessionKey: watcher, targetSessionKey: group }, database); + expect( + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "per-channel-peer" }, + database, + ), + ).toBe(false); + expect(readCursor(database, watcher, group)).toBeDefined(); + }); + + it("records and coalesces group activity without an immediate wake", async () => { + vi.useFakeTimers(); + const wakes = vi.fn(async () => ({ status: "ran" as const, durationMs: 1 })); + disposeHeartbeatWakeHandler = setHeartbeatWakeHandler(wakes); + await vi.advanceTimersByTimeAsync(300); + wakes.mockClear(); + const database = createDatabaseOptions(); + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "main" }, + database, + ); + + for (const actorId of ["human-1", "human-2"]) { + recordSessionHumanDirectMessage( + { + sessionKey: group, + entry: { sessionId: "session-group", updatedAt: Date.now(), chatType: "group" }, + agentId: "main", + actor: { actorType: "human", actorId }, + channel: "telegram", + }, + database, + ); + } + await vi.advanceTimersByTimeAsync(300); + + expect(listSessionStateEventsSince(group, "main", 0, 200, database).events).toHaveLength(2); + expect(peekSystemEventEntries(watcher)).toHaveLength(1); + const cursor = readCursor(database, watcher, group); + expect(cursor).toBeDefined(); + expect(cursor!.material_sequence).toBeGreaterThan(cursor!.notified_sequence); + expect(wakes).not.toHaveBeenCalled(); + }); + + it("prunes orphaned ambient markers while retaining markers for active cursors", () => { + const database = createDatabaseOptions(); + const dormantGroup = "agent:main:slack:channel:dormant"; + const registeredAt = 100; + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "main" }, + { ...database, now: registeredAt }, + ); + registerMainSessionGroupWatch( + { sessionKey: dormantGroup, agentId: "main", dmScope: "main" }, + { ...database, now: registeredAt }, + ); + + const activeAt = registeredAt + SESSION_STATE_RETENTION_MS + 1; + recordSessionHumanDirectMessage( + { + sessionKey: group, + entry: { sessionId: "session-group", updatedAt: activeAt, chatType: "group" }, + agentId: "main", + actor: { actorType: "human", actorId: "human-1" }, + channel: "telegram", + }, + { ...database, now: activeAt }, + ); + sweepSessionStateWatchNotices({ ...database, now: activeAt }); + + expect(listAmbientGroupWatchTargets(watcher, database)).toEqual(new Set([group])); + const cursors = openOpenClawStateDatabase(database) + .db.prepare("SELECT COUNT(*) AS count FROM session_watch_cursors") + .get() as { count: number }; + expect(cursors.count).toBe(2); + }); + + it("keeps explicit A2A group watches on the immediate wake path", async () => { + vi.useFakeTimers(); + const wakes = vi.fn(async () => ({ status: "ran" as const, durationMs: 1 })); + disposeHeartbeatWakeHandler = setHeartbeatWakeHandler(wakes); + await vi.advanceTimersByTimeAsync(300); + wakes.mockClear(); + const database = createDatabaseOptions(); + const coordinator = "agent:main:coordinator"; + registerSessionStateWatch( + { watcherSessionKey: coordinator, targetSessionKey: group }, + database, + ); + + recordSessionHumanDirectMessage( + { + sessionKey: group, + entry: { sessionId: "session-group", updatedAt: Date.now(), chatType: "group" }, + agentId: "main", + actor: { actorType: "human", actorId: "human-1" }, + channel: "telegram", + }, + database, + ); + await vi.advanceTimersByTimeAsync(300); + + expect(peekSystemEventEntries(coordinator)).toHaveLength(1); + expect(wakes).toHaveBeenCalledTimes(1); + }); + + it("promotes an ambient main-to-group watch to explicit immediate delivery", async () => { + vi.useFakeTimers(); + const wakes = vi.fn(async () => ({ status: "ran" as const, durationMs: 1 })); + disposeHeartbeatWakeHandler = setHeartbeatWakeHandler(wakes); + await vi.advanceTimersByTimeAsync(300); + wakes.mockClear(); + const database = createDatabaseOptions(); + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "main" }, + database, + ); + expect(listAmbientGroupWatchTargets(watcher, database)).toEqual(new Set([group])); + + registerSessionStateWatch({ watcherSessionKey: watcher, targetSessionKey: group }, database); + expect(listAmbientGroupWatchTargets(watcher, database)).toEqual(new Set()); + // Later inbound group registration must not downgrade the explicit watch. + registerMainSessionGroupWatch( + { sessionKey: group, agentId: "main", dmScope: "main" }, + database, + ); + + recordSessionHumanDirectMessage( + { + sessionKey: group, + entry: { sessionId: "session-group", updatedAt: Date.now(), chatType: "group" }, + agentId: "main", + actor: { actorType: "human", actorId: "human-1" }, + channel: "telegram", + }, + database, + ); + await vi.advanceTimersByTimeAsync(300); + + expect(peekSystemEventEntries(watcher)).toHaveLength(1); + expect(wakes).toHaveBeenCalledTimes(1); + }); + it("gates unparented human turns on registered watchers", () => { const database = createDatabaseOptions(); const entry = { sessionId: "session-child", updatedAt: Date.now() }; diff --git a/src/sessions/session-state-events.ts b/src/sessions/session-state-events.ts index 64a990bb15d7..20193cb9c6bf 100644 --- a/src/sessions/session-state-events.ts +++ b/src/sessions/session-state-events.ts @@ -3,6 +3,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { Insertable, Selectable } from "kysely"; import { loadSessionEntry } from "../config/sessions/session-accessor.js"; import type { SessionEntry } from "../config/sessions/types.js"; +import type { DmScope } from "../config/types.base.js"; import { executeSqliteQuerySync, executeSqliteQueryTakeFirstSync, @@ -10,13 +11,14 @@ import { } from "../infra/kysely-sync.js"; import { normalizeSqliteNumber } from "../infra/sqlite-number.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; -import { resolveAgentIdFromSessionKey } from "../routing/session-key.js"; +import { buildAgentMainSessionKey, resolveAgentIdFromSessionKey } from "../routing/session-key.js"; import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; import { openOpenClawStateDatabase, runOpenClawStateWriteTransaction, type OpenClawStateDatabaseOptions, } from "../state/openclaw-state-db.js"; +import { classifySessionKind } from "./classify-session-kind.js"; import type { InputProvenance } from "./input-provenance.js"; import { NOTIFY_BY_SESSION_STATE_EVENT_KIND as NOTIFY_BY_KIND, @@ -68,6 +70,7 @@ type SessionWatchCursorRow = Selectable = []; try { const event = runOpenClawStateWriteTransaction(({ db }) => { @@ -340,7 +364,12 @@ export function recordSessionStateEvent( sequence: insertedSequence, now, }); - notices.push({ watcherSessionKey, targetSessionKey: input.sessionKey, lastSeenSequence }); + notices.push({ + watcherSessionKey, + targetSessionKey: input.sessionKey, + lastSeenSequence, + queueOnly: hasAmbientGroupWatchMarker(db, watcherSessionKey, input.sessionKey), + }); } const row = executeSqliteQueryTakeFirstSync( @@ -499,6 +528,7 @@ export function acknowledgeSessionStateNotices( watcherSessionKey: string; targetSessionKey: string; lastSeenSequence: number; + queueOnly: boolean; }> = []; try { runOpenClawStateWriteTransaction(({ db }) => { @@ -527,6 +557,7 @@ export function acknowledgeSessionStateNotices( watcherSessionKey, targetSessionKey, lastSeenSequence: notified, + queueOnly: hasAmbientGroupWatchMarker(db, watcherSessionKey, targetSessionKey), }); } } @@ -546,13 +577,18 @@ export function handleSessionStateSessionReset( ): void { try { runOpenClawStateWriteTransaction(({ db }) => { - // Cursor rows only exist for agent-qualified watcher keys (see - // isNotifiableWatcherKey), so a bare-key reset cannot cross agents here. + // Notifiable cursor rows use agent-qualified keys. The encoded ambient + // marker is deleted alongside its owner without interpreting bare keys. executeSqliteQuerySync( db, getSessionStateKysely(db) .deleteFrom("session_watch_cursors") - .where("watcher_session_key", "=", sessionKey), + .where((eb) => + eb.or([ + eb("watcher_session_key", "=", sessionKey), + eb("watcher_session_key", "=", ambientGroupWatchMarkerKey(sessionKey)), + ]), + ), ); }, options); } catch (error) { @@ -591,6 +627,7 @@ export function handleSessionStateSessionDeleted( .where((eb) => eb.or([ eb("watcher_session_key", "=", sessionKey), + eb("watcher_session_key", "=", ambientGroupWatchMarkerKey(sessionKey)), eb("target_session_key", "=", sessionKey), ]), ), @@ -640,6 +677,7 @@ export function sweepSessionStateWatchNotices( watcherSessionKey: row.watcher_session_key, targetSessionKey: row.target_session_key, lastSeenSequence: normalizeSqliteNumber(row.last_seen_sequence) ?? 0, + queueOnly: hasAmbientGroupWatchMarker(db, row.watcher_session_key, row.target_session_key), }); } pruneSessionStateEvents({ ...options, now }); @@ -710,12 +748,38 @@ function pruneSessionStateEvents( kysely.deleteFrom("session_state_events").where("sequence", "<=", sequenceCutoff), ); } + const cursorCutoff = now - SESSION_STATE_RETENTION_MS; executeSqliteQuerySync( db, kysely .deleteFrom("session_watch_cursors") - .where("updated_at", "<", now - SESSION_STATE_RETENTION_MS), + .where("updated_at", "<", cursorCutoff) + .where("watcher_session_key", "not like", `${AMBIENT_GROUP_WATCH_MARKER_PREFIX}%`), ); + const staleMarkers = executeSqliteQuerySync( + db, + kysely + .selectFrom("session_watch_cursors") + .select(["watcher_session_key", "target_session_key"]) + .where("updated_at", "<", cursorCutoff) + .where("watcher_session_key", "like", `${AMBIENT_GROUP_WATCH_MARKER_PREFIX}%`), + ).rows; + for (const marker of staleMarkers) { + const watcherSessionKey = decodeAmbientGroupWatchMarkerKey(marker.watcher_session_key); + if ( + watcherSessionKey && + readCursor(db, watcherSessionKey, marker.target_session_key) !== undefined + ) { + continue; + } + executeSqliteQuerySync( + db, + kysely + .deleteFrom("session_watch_cursors") + .where("watcher_session_key", "=", marker.watcher_session_key) + .where("target_session_key", "=", marker.target_session_key), + ); + } }, options); lastPruneAt = now; } catch (error) { @@ -783,6 +847,7 @@ function hasSessionStateWatchers( .selectFrom("session_watch_cursors") .select("watcher_session_key") .where("target_session_key", "=", targetSessionKey) + .where("watcher_session_key", "not like", `${AMBIENT_GROUP_WATCH_MARKER_PREFIX}%`) .limit(1), ); return row !== undefined; @@ -793,6 +858,31 @@ function hasSessionStateWatchers( } } +/** List durable ambient-group targets owned by one watcher; failures grant nothing. */ +export function listAmbientGroupWatchTargets( + watcherSessionKey: string, + options: OpenClawStateDatabaseOptions = {}, +): Set { + try { + const { db } = openOpenClawStateDatabase(options); + const rows = executeSqliteQuerySync( + db, + getSessionStateKysely(db) + .selectFrom("session_watch_cursors") + .select("target_session_key") + .where("watcher_session_key", "=", ambientGroupWatchMarkerKey(watcherSessionKey)), + ).rows; + return new Set( + rows + .map((row) => row.target_session_key) + .filter((targetSessionKey) => readCursor(db, watcherSessionKey, targetSessionKey)), + ); + } catch (error) { + log.warn(`failed to list ambient group watch targets: ${String(error)}`); + return new Set(); + } +} + /** Register an explicit watcher (e.g. a sessions_send coordinator) for a target session. */ export function registerSessionStateWatch( params: { watcherSessionKey: string; targetSessionKey: string; targetAgentId?: string }, @@ -808,6 +898,15 @@ export function registerSessionStateWatch( try { let registered = false; runOpenClawStateWriteTransaction(({ db }) => { + // An explicit watch promotes an ambient group watch back to the normal + // immediate-wake path and removes its tree-read authorization marker. + executeSqliteQuerySync( + db, + getSessionStateKysely(db) + .deleteFrom("session_watch_cursors") + .where("watcher_session_key", "=", ambientGroupWatchMarkerKey(params.watcherSessionKey)) + .where("target_session_key", "=", params.targetSessionKey), + ); // Re-watching must not clobber pending-notice cursor state. if (readCursor(db, params.watcherSessionKey, params.targetSessionKey)) { registered = true; @@ -839,6 +938,98 @@ export function registerSessionStateWatch( } } +/** Register the personal agent's main session to observe one routed group session. */ +export function registerMainSessionGroupWatch( + params: { + sessionKey: string; + agentId: string; + entry?: SessionEntry; + dmScope: DmScope; + }, + options: OpenClawStateDatabaseOptions & { now?: number } = {}, +): boolean { + if (classifySessionKind(params.sessionKey, params.entry) !== "group") { + return false; + } + const watcherSessionKey = buildAgentMainSessionKey({ + agentId: params.agentId, + }); + const now = options.now ?? Date.now(); + try { + const { db: readDb } = openOpenClawStateDatabase(options); + if (params.dmScope !== "main") { + const markerKey = ambientGroupWatchMarkerKey(watcherSessionKey); + if (!readCursor(readDb, markerKey, params.sessionKey)) { + return false; + } + runOpenClawStateWriteTransaction(({ db }) => { + // Recheck provenance in the write transaction: an explicit registration + // may have promoted this pair after the read-only preflight. + if (!readCursor(db, markerKey, params.sessionKey)) { + return; + } + executeSqliteQuerySync( + db, + getSessionStateKysely(db) + .deleteFrom("session_watch_cursors") + .where("target_session_key", "=", params.sessionKey) + .where("watcher_session_key", "in", [watcherSessionKey, markerKey]), + ); + }, options); + return false; + } + // This runs on every human group turn. Keep the steady-state path read-only; + // the transaction below is only for first registration and its race recheck. + if (readCursor(readDb, watcherSessionKey, params.sessionKey)) { + return true; + } + let registered = false; + runOpenClawStateWriteTransaction(({ db }) => { + const existing = readCursor(db, watcherSessionKey, params.sessionKey); + const markerKey = ambientGroupWatchMarkerKey(watcherSessionKey); + const marker = readCursor(db, markerKey, params.sessionKey); + if (existing) { + // Missing marker means an explicit watch already owns this pair. Do not + // downgrade it when later human group turns revisit registration. + registered = true; + return; + } + const head = executeSqliteQueryTakeFirstSync( + db, + getSessionStateKysely(db) + .selectFrom("session_state_heads") + .select("last_sequence") + .where("session_key", "=", params.sessionKey) + .where("agent_id", "=", params.agentId), + ); + const sequence = normalizeOptionalSqliteNumber(head?.last_sequence) ?? 0; + upsertSeedCursor({ + db, + watcherSessionKey, + targetSessionKey: params.sessionKey, + sequence, + now, + }); + if (!marker) { + // The paired marker is durable provenance, not a notifiable watcher. It + // authorizes tree reads and queue-only delivery only for this auto-watch. + upsertSeedCursor({ + db, + watcherSessionKey: markerKey, + targetSessionKey: params.sessionKey, + sequence, + now, + }); + } + registered = true; + }, options); + return registered; + } catch (error) { + log.warn(`failed to register ambient group watch: ${String(error)}`); + return false; + } +} + export function recordSessionHumanDirectMessage( params: { sessionKey: string; diff --git a/src/sessions/session-state-notices.ts b/src/sessions/session-state-notices.ts index c6c43c877051..0d09411375dd 100644 --- a/src/sessions/session-state-notices.ts +++ b/src/sessions/session-state-notices.ts @@ -36,7 +36,7 @@ function shouldWakeWatcher(watcherSessionKey: string): boolean { // for one agent's child could be drained and acknowledged by another agent's global // turn — a cross-A2A metadata leak plus a lost notification. Until watcher identity // is agent-scoped end-to-end, such watchers get durable events and changesSince but -// no notices or cursors. +// no notices. Non-notifiable ambient marker rows are also deliberately ignored. export function isNotifiableWatcherKey(watcherSessionKey: string): boolean { return parseAgentSessionKey(watcherSessionKey) != null; } @@ -45,11 +45,18 @@ export function enqueueSessionStateNotice(params: { watcherSessionKey: string; targetSessionKey: string; lastSeenSequence: number; + queueOnly?: boolean; }): void { enqueueSystemEvent(sessionStateNoticeText(params.targetSessionKey, params.lastSeenSequence), { sessionKey: params.watcherSessionKey, contextKey: `${SESSION_STATE_CONTEXT_PREFIX}${encodeNoticeTarget(params.targetSessionKey)}`, + ...(params.queueOnly ? { replace: true } : {}), }); + // Group activity is ambient context. Coalesce it for the next main turn instead + // of waking the personal agent once per inbound group message. + if (params.queueOnly) { + return; + } if (!shouldWakeWatcher(params.watcherSessionKey)) { return; } diff --git a/src/state/openclaw-state-schema.generated.ts b/src/state/openclaw-state-schema.generated.ts index 4cd61d4fca4c..6f8bbeb49033 100644 --- a/src/state/openclaw-state-schema.generated.ts +++ b/src/state/openclaw-state-schema.generated.ts @@ -167,10 +167,10 @@ CREATE TABLE IF NOT EXISTS session_state_heads ( PRIMARY KEY (session_key, agent_id) ) STRICT; --- Watcher identity is the bare session key, matching the process-local system-event --- queue it feeds. Producers only create rows for agent-qualified watcher keys; --- bare keys (session.scope="global") are ambiguous across agents and are excluded --- from the notice protocol until watcher identity is agent-scoped end-to-end. +-- Notifiable watcher identity is the bare session key, matching the process-local +-- system-event queue it feeds. Ambient group watches also own non-notifiable marker +-- rows. Other bare keys (session.scope="global") are ambiguous across agents and +-- excluded until watcher identity is agent-scoped end-to-end. CREATE TABLE IF NOT EXISTS session_watch_cursors ( watcher_session_key TEXT NOT NULL, target_session_key TEXT NOT NULL, diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index 5920f5f492bb..2d3da7f7e6a3 100644 --- a/src/state/openclaw-state-schema.sql +++ b/src/state/openclaw-state-schema.sql @@ -162,10 +162,10 @@ CREATE TABLE IF NOT EXISTS session_state_heads ( PRIMARY KEY (session_key, agent_id) ) STRICT; --- Watcher identity is the bare session key, matching the process-local system-event --- queue it feeds. Producers only create rows for agent-qualified watcher keys; --- bare keys (session.scope="global") are ambiguous across agents and are excluded --- from the notice protocol until watcher identity is agent-scoped end-to-end. +-- Notifiable watcher identity is the bare session key, matching the process-local +-- system-event queue it feeds. Ambient group watches also own non-notifiable marker +-- rows. Other bare keys (session.scope="global") are ambiguous across agents and +-- excluded until watcher identity is agent-scoped end-to-end. CREATE TABLE IF NOT EXISTS session_watch_cursors ( watcher_session_key TEXT NOT NULL, target_session_key TEXT NOT NULL,