feat: keep main agents aware of group activity (#110332)

* feat(sessions): auto-watch group activity from main * feat(sessions): allow watchers to read watched sessions * fix(sessions): skip group watches for system turns * fix(sessions): preserve watched session semantics * fix(channels): propagate effective dm scope * fix(sessions): canonicalize main watch identity * fix(sessions): distinguish ambient group watches * fix(sessions): revoke ineligible ambient watches * chore: drop CHANGELOG edit (release generation owns it)
2026-07-25 12:51:14 +00:00 · 2026-07-18 04:06:58 +01:00
parent 9dec7acc91
commit 31ed59bdf6
44 changed files with 662 additions and 31 deletions
--- 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
--- 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,
--- 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,
        },
--- 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,
--- 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,
    },
--- 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,
    },
--- 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,
    },
--- 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,
--- 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,
        },
--- 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:
--- 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,
      },
--- 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,
    },
--- 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) => {
--- 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;
--- 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,
    },
--- 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 };
--- 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;
--- 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;
--- 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,
      },
--- 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,
--- 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,
--- 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,
--- 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<v
      },
      route: {
        agentId: route.agentId,
+        dmScope: route.dmScope,
        accountId: route.accountId,
        routeSessionKey: route.sessionKey,
      },
@@ -598,7 +599,7 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
      channel: "tlon",
      accountId: route.accountId,
      cfg,
-      route: { agentId: route.agentId, sessionKey: route.sessionKey },
+      route: { agentId: route.agentId, dmScope: route.dmScope, sessionKey: route.sessionKey },
      ctxPayload,
      delivery: {
        preparePayload: prepareReplyPayload,
--- a/extensions/twitch/src/monitor.ts
+++ b/extensions/twitch/src/monitor.ts
@@ -100,6 +100,7 @@ async function processTwitchMessage(params: {
          },
          route: {
            agentId: route.agentId,
+            dmScope: route.dmScope,
            accountId: route.accountId,
            routeSessionKey: route.sessionKey,
          },
@@ -122,7 +123,7 @@ async function processTwitchMessage(params: {
          cfg,
          channel: "twitch",
          accountId,
-          route: { agentId: route.agentId, sessionKey: route.sessionKey },
+          route: { agentId: route.agentId, dmScope: route.dmScope, sessionKey: route.sessionKey },
          ctxPayload,
          delivery: {
            durable: () => ({
--- 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,
    },
--- 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) =>
--- 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) => {
--- 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<string>()),
+);
 const listSessionStateEventsSinceMock = vi.hoisted(() =>
  vi.fn((_sessionKey: string, _agentId: string, _after: number, _limit: number) => ({
    events: [] as Array<Record<string, unknown>>,
@@ -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<string, SessionEntry>) {
  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: {
--- 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",
--- 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<string>(),
 }));

 vi.mock("./sessions-helpers.js", async (importActual) => {
--- 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,
--- 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.
--- 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",
--- 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,
--- 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");
--- a/src/channels/turn/lifecycle.ts
+++ b/src/channels/turn/lifecycle.ts
@@ -29,6 +29,7 @@ export function assembleResolvedChannelTurn<TDispatchResult>(
    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<TDispatchResult>(
  const { cfg, route, ...turn } = value;
  return {
    ...turn,
+    ctxPayload: route.dmScope ? { ...turn.ctxPayload, DmScope: route.dmScope } : turn.ctxPayload,
    cfg,
    agentId: route.agentId,
    routeSessionKey: route.sessionKey,
--- 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<TDispatchResult = DispatchFromConfigResult> = {

 type ChannelTurnRoute = {
  agentId: string;
+  dmScope?: DmScope;
  sessionKey: string;
 };

--- 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<string>(),
  recordSessionStateEvent: hoisted.recordSessionStateEvent,
 }));

--- 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<string> | 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 (
--- 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<typeof createDatabaseOptions>,
  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() };
--- 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<OpenClawStateKyselyDatabase["session_wat
 const SESSION_STATE_RETENTION_MS = 30 * 24 * 60 * 60_000;
 const SESSION_STATE_MAX_ROWS = 50_000;
 const SESSION_STATE_PRUNE_INTERVAL_MS = 60 * 60_000;
+const AMBIENT_GROUP_WATCH_MARKER_PREFIX = "ambient-group-watch:";
 const log = createSubsystemLogger("sessions/state-events");
 let lastPruneAt = 0;

@@ -146,6 +149,26 @@ function readCursor(
  );
 }

+function ambientGroupWatchMarkerKey(watcherSessionKey: string): string {
+  return `${AMBIENT_GROUP_WATCH_MARKER_PREFIX}${Buffer.from(watcherSessionKey, "utf8").toString("hex")}`;
+}
+
+function decodeAmbientGroupWatchMarkerKey(markerKey: string): string | undefined {
+  const encoded = markerKey.slice(AMBIENT_GROUP_WATCH_MARKER_PREFIX.length);
+  if (!encoded || encoded.length % 2 !== 0 || !/^[0-9a-f]+$/.test(encoded)) {
+    return undefined;
+  }
+  return Buffer.from(encoded, "hex").toString("utf8");
+}
+
+function hasAmbientGroupWatchMarker(
+  db: DatabaseSync,
+  watcherSessionKey: string,
+  targetSessionKey: string,
+): boolean {
+  return Boolean(readCursor(db, ambientGroupWatchMarkerKey(watcherSessionKey), targetSessionKey));
+}
+
 function upsertSeedCursor(params: {
  db: DatabaseSync;
  watcherSessionKey: string;
@@ -258,6 +281,7 @@ export function recordSessionStateEvent(
    watcherSessionKey: string;
    targetSessionKey: string;
    lastSeenSequence: number;
+    queueOnly: boolean;
  }> = [];
  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<string> {
+  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;
--- 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;
  }
--- 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,
--- 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,