fix: allow guests to notify owned child sessions (#162986)

Allow session-scoped operators to notify their owned sessions while preserving current authority and target checks.

Fixes #162924.

docs/concepts/session-tool.md

@@ -293,6 +293,8 @@ Isolated scheduled jobs can wait for an inline reply, but receive no detached re
 
 These reply deliveries apply to new or follow-up turns. Default sends with no reply wait to your own running child skip separate reply delivery and leave completion with the active run's owner. `mode: "steer"` returns admission only for guidance added to an active run and leaves completion with that run's existing owner. It uses the existing `sessions_send` access checks. For the built-in runtime, a busy tool or model response can delay transcript persistence until the next steering boundary; the send's reply-wait deadline does not withdraw admitted guidance. Acceptance is not proof of transcript persistence or model consumption, and does not make the in-memory steering queue restart-durable. The receiving run retains source authority until the input settles or that exact run ends or aborts; a missing backend settlement callback cannot retain it past the run. Existing explicit cancellation, run-lifecycle, and authorization rules still apply. `mode: "notify"` queues context without starting a turn. Registered task completion and paused-task resume keep their existing completion owner and do not add a second reply delivery.
 
+An operator with `operator.sessions.write` can use `mode: "notify"` for an authorized session they own, including an owned child. Notifications retain the requester's current authority and target-session checks before queueing. They remain in memory and do not start a run.
+
 Child coordination stays in agent context and raw transcripts. The receiving chat hides child reports and automatic coordination replies, while normal task-completion summaries and direct human answers remain visible. Historical messages without source provenance cannot be classified as child traffic.
 
 Pass `watch: true` to also register the sender as a state-change watcher of the target: when another actor later sends the target a direct human message or changes its goal, the sender receives a system notice pointing at `session_status` `changesSince`. Registration happens after successful dispatch, targets the session that actually received the message, and starts at its current state version, so only later changes produce notices. The result reports `watched: true` when registration succeeded. See [Session state awareness](/concepts/session-state).

src/agents/tools/sessions-send-tool.delivery.ts

@@ -68,6 +68,7 @@ export async function notifySessionsSendSession(params: {
   };
   const event = selection.operatorAuthority
     ? await runWithInProcessGatewaySessionMutation(
+        "sessions.send",
         { sessionKey: params.sessionKey, agentId: params.targetAgentId },
         (assertCurrent) => {
           assertCurrent();

src/agents/tools/sessions-send-tool.steering.ts

@@ -29,7 +29,7 @@ export async function queueSessionsSendSteeringWithCustody(
   void (async () => {
     try {
       await custody.run(() =>
-        runWithInProcessGatewaySessionMutation(target, async (assertMutationCurrent) => {
+        runWithInProcessGatewaySessionMutation("agent", target, async (assertMutationCurrent) => {
           const assertCurrent = () => {
             if (!accepted) {
               assertCallerCurrent?.("agent");

src/gateway/local-request-context.session-notify.test.ts

@@ -0,0 +1,217 @@
+import { expectDefined } from "@openclaw/normalization-core/expect";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { awaitGateBeforeSettlement } from "../../test/helpers/promise.js";
+import * as delivery from "../agents/tools/sessions-send-tool.delivery.js";
+import { createSessionsSendTool } from "../agents/tools/sessions-send-tool.js";
+import { withPersonalToolTurn } from "../auto-reply/reply/personal-tool-turn.test-support.js";
+import { resolveSessionStorePathCore } from "../config/sessions/paths.js";
+import {
+  deleteSessionEntryLifecycle,
+  upsertSessionEntryCore,
+} from "../config/sessions/session-accessor.js";
+import { drainSystemEvents } from "../infra/system-events.js";
+import { getPluginRuntimeGatewayRequestScope } from "../plugins/runtime/gateway-request-scope.js";
+import { createDeferredCore } from "../shared/deferred.js";
+import {
+  drainSessionToolsFixture,
+  withSessionToolsFixture,
+} from "./local-request-context.session-tools.test-support.js";
+import { captureGatewayOperatorRunAuthority } from "./operator-run-authority.js";
+import { roleClient } from "./session-sharing.test-utils.js";
+
+const REQUESTER = "agent:main:dashboard:notify-parent";
+const TARGET = "agent:main:dashboard:notify-child";
+const FOREIGN = "agent:main:dashboard:notify-foreign";
+
+async function withNotification(
+  scope: "operator.sessions.write" | "operator.write",
+  run: (fixture: {
+    notify: (
+      sessionKey?: string,
+    ) => ReturnType<ReturnType<typeof createSessionsSendTool>["execute"]>;
+    revoke: () => void;
+    changeRole: () => void;
+    deleteTarget: () => Promise<void>;
+  }) => Promise<void>,
+) {
+  await withSessionToolsFixture(async (cfg) => {
+    const role = expectDefined(cfg.gateway?.roles?.definitions?.view, "guest role");
+    role.scopes = [scope];
+    role.sandbox = "required";
+    const client = roleClient("view", "notification-owner");
+    client.connect.scopes = [scope];
+    const profileId = expectDefined(
+      client.authenticatedUserProfile,
+      "notification person",
+    ).profileId;
+    for (const [sessionKey, sessionId, creatorId] of [
+      [REQUESTER, "notify-parent-session", profileId],
+      [TARGET, "notify-child-session", profileId],
+      [FOREIGN, "notify-foreign-session", "another-person"],
+    ] as const) {
+      await upsertSessionEntryCore(
+        { agentId: "main", sessionKey },
+        {
+          sessionId,
+          updatedAt: 1,
+          visibility: "shared",
+          createdVia: "operator",
+          createdActor: { type: "human", source: "profile", id: creatorId },
+          sandbox: "required",
+          ...(sessionKey !== REQUESTER ? { spawnedBy: REQUESTER } : {}),
+        },
+      );
+    }
+    const context = expectDefined(getPluginRuntimeGatewayRequestScope()?.context, "Gateway");
+    const source = new AbortController();
+    const captured = expectDefined(
+      await captureGatewayOperatorRunAuthority({
+        client,
+        context,
+        sourceAuthority: {
+          signal: source.signal,
+          assertCurrent: () => source.signal.throwIfAborted(),
+        },
+      }),
+      "original operator source",
+    );
+    try {
+      await withPersonalToolTurn(
+        {
+          owner: {
+            profileId,
+            senderId: "notification-owner",
+            name: "Notification owner",
+            operatorAuthority: captured.authority,
+          },
+          sessionKey: REQUESTER,
+          sessionId: "notify-parent-session",
+        },
+        async () => {
+          const tool = createSessionsSendTool({ config: cfg, agentSessionKey: REQUESTER });
+          await run({
+            notify: (sessionKey = TARGET) =>
+              tool.execute("notify-child", {
+                sessionKey,
+                mode: "notify",
+                message: "Inspect the prepared change.",
+              }),
+            revoke: () => source.abort(new Error("notification source revoked")),
+            changeRole: () => {
+              role.scopes = ["operator.sessions.read"];
+            },
+            deleteTarget: async () => {
+              const removed = await deleteSessionEntryLifecycle({
+                agentId: "main",
+                storePath: resolveSessionStorePathCore(cfg.session?.store, { agentId: "main" }),
+                target: { canonicalKey: TARGET, storeKeys: [TARGET] },
+                archiveTranscript: false,
+                deleteTranscriptWithoutArchive: true,
+              });
+              expect(removed).toMatchObject({
+                deleted: true,
+                deletedSessionId: "notify-child-session",
+              });
+            },
+          });
+        },
+      );
+    } finally {
+      captured.release();
+      drainSystemEvents(TARGET);
+      drainSystemEvents(FOREIGN);
+      drainSystemEvents(REQUESTER);
+    }
+  });
+}
+
+describe("session notification authority", () => {
+  afterEach(drainSessionToolsFixture);
+
+  it.each(["operator.sessions.write", "operator.write"] as const)(
+    "queues an owned-child notification for %s through the real tool and router",
+    async (scope) => {
+      await withNotification(scope, async ({ notify, revoke }) => {
+        await expect(notify()).resolves.toMatchObject({
+          details: {
+            status: "queued",
+            sessionKey: TARGET,
+            notificationId: expect.any(String),
+            durability: "process",
+            runStarted: false,
+          },
+        });
+        revoke();
+        expect(drainSystemEvents(TARGET)).toEqual([
+          expect.stringContaining("Inspect the prepared change."),
+        ]);
+        expect(drainSystemEvents(REQUESTER)).toEqual([]);
+      });
+    },
+  );
+
+  it("denies a visible child now owned by another person", async () => {
+    await withNotification("operator.sessions.write", async ({ notify }) => {
+      await expect(notify(FOREIGN)).rejects.toThrow(/own session|not allowed/);
+      expect(drainSystemEvents(FOREIGN)).toEqual([]);
+    });
+  });
+
+  it.each(["source revoked", "role changed", "target deleted"] as const)(
+    "cannot queue when %s during notification authorization",
+    async (change) => {
+      await withNotification(
+        "operator.sessions.write",
+        async ({ notify, revoke, changeRole, deleteTarget }) => {
+          const context = expectDefined(getPluginRuntimeGatewayRequestScope()?.context, "Gateway");
+          const entered = createDeferredCore();
+          const resume = createDeferredCore();
+          const prepare = context.ensureSessionRowProjection;
+          const originalNotify = delivery.notifySessionsSendSession;
+          const spy = vi
+            .spyOn(delivery, "notifySessionsSendSession")
+            .mockImplementationOnce((params) => {
+              // Pause the real router after tool visibility checks, before mutation admission.
+              context.ensureSessionRowProjection = async () => {
+                await prepare?.();
+                entered.resolve();
+                await resume.promise;
+              };
+              return originalNotify(params);
+            });
+          const pending = notify();
+          void pending.catch(() => {});
+          try {
+            await awaitGateBeforeSettlement(
+              entered.promise,
+              pending,
+              "Notification settled before mutation admission",
+            );
+            expect(spy).toHaveBeenCalledOnce();
+            if (change === "source revoked") {
+              revoke();
+            } else if (change === "role changed") {
+              changeRole();
+            } else {
+              await deleteTarget();
+            }
+            resume.resolve();
+            await expect(pending).rejects.toThrow(
+              change === "source revoked"
+                ? /notification source revoked/
+                : change === "role changed"
+                  ? /operator role changed/
+                  : /session target is unavailable/i,
+            );
+          } finally {
+            resume.resolve();
+            await pending.catch(() => {});
+            context.ensureSessionRowProjection = prepare;
+            spy.mockRestore();
+          }
+          expect(drainSystemEvents(TARGET)).toEqual([]);
+        },
+      );
+    },
+  );
+});

src/gateway/server-plugin-in-process-dispatch.ts

@@ -227,28 +227,31 @@ export function withInProcessGatewayRead<T>(
   );
 }
 
-/** Local session input uses the same authorization and commit fences as an agent RPC. */
+/** Local session input retains its operation's authorization and commit fences. */
 export async function runWithInProcessGatewaySessionMutation<T>(
+  method: "agent" | "sessions.send",
   params: { sessionKey: string; agentId?: string },
   run: (assertCurrent: () => void) => Promise<T> | T,
 ): Promise<T> {
   const assertCallerCurrent = captureGatewayToolCallerAssertion();
+  const requestParams =
+    method === "sessions.send" ? { key: params.sessionKey, agentId: params.agentId } : params;
   return await withInProcessGatewayDispatch(
-    "agent",
-    params,
+    method,
+    requestParams,
     { forceSyntheticClient: true, syntheticScopeMode: "minimum" },
     async (resolved) => {
       const { authorizeGatewayRequestPreDispatch, createRequestGatewayMethodRegistry } =
         await import("./server-methods.js");
       const assertInvocationCurrent = () => {
-        assertCallerCurrent?.("agent");
+        assertCallerCurrent?.(method);
         resolved.assertContextCurrent();
         resolved.assertInvocationCurrent();
       };
       assertInvocationCurrent();
       const authorization = await authorizeGatewayRequestPreDispatch({
-        method: "agent",
-        requestParams: params,
+        method,
+        requestParams,
         client: resolved.client,
         context: resolved.context,
         methodRegistry:
@@ -260,11 +263,18 @@ export async function runWithInProcessGatewaySessionMutation<T>(
         const assertCurrent = () => {
           assertInvocationCurrent();
           if (authorization.error) {
-            unwrapGatewayMethodDispatchResponse("agent", {
+            unwrapGatewayMethodDispatchResponse(method, {
               ok: false,
               error: authorization.error,
             });
           }
+          // Unlike the public send RPC, local notifications cannot create a session.
+          if (
+            method === "sessions.send" &&
+            !authorization.sessionMutationAuthorization?.admittedTarget
+          ) {
+            throw new Error("Session target is unavailable for notification.");
+          }
           authorization.sessionMutationAuthorization?.assertCurrent();
           authorization.sessionAccessAuthority?.assertCurrent();
         };

test/vitest/vitest.gateway-server-paths.mjs

@@ -78,6 +78,7 @@ export const gatewayDatabaseWorkerTestFiles = [
   "src/gateway/http-auth-utils.test.ts",
   "src/gateway/internal-source-reply-persistence.test.ts",
   "src/gateway/link-understanding.product.test.ts",
+  "src/gateway/local-request-context.session-notify.test.ts",
   "src/gateway/local-request-context.session-tools.test.ts",
   "src/gateway/local-request-context.test.ts",
   "src/gateway/managed-image-attachments.authority.test.ts",