Skip to content

Commit 0b58b05

Browse files
youngting520youngting520
committed
fix(outbound): reject split agent session ownership
1 parent 300b09b commit 0b58b05

12 files changed

Lines changed: 608 additions & 11 deletions

‎src/gateway/server-methods/send.test.ts‎

Lines changed: 81 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1266,22 +1266,43 @@ describe("gateway send mirroring", () => {
12661266
expect(deliveryCall()?.mirror?.agentId).toBe("work");
12671267
});
12681268

1269-
it("prefers explicit agentId over sessionKey agent for delivery and mirror", async () => {
1270-
mockDeliverySuccess("m-agent-precedence");
1269+
it("rejects explicit agentId when it conflicts with the sessionKey agent", async () => {
1270+
const { respond } = await runSend({
1271+
to: "channel:C1",
1272+
message: "hello",
1273+
channel: "slack",
1274+
agentId: "work",
1275+
sessionKey: "agent:main:slack:channel:c1",
1276+
idempotencyKey: "idem-agent-mismatch",
1277+
});
1278+
1279+
const response = firstRespondCall(respond);
1280+
expect(response?.[0]).toBe(false);
1281+
expect(response?.[1]).toBeUndefined();
1282+
expect(response?.[2]).toMatchObject({
1283+
code: "INVALID_REQUEST",
1284+
message: 'send agentId "work" does not match session key agent "main"',
1285+
});
1286+
expect(mocks.resolveOutboundSessionRoute).not.toHaveBeenCalled();
1287+
expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled();
1288+
});
1289+
1290+
it("allows explicit agentId when it matches a different sessionKey conversation", async () => {
1291+
mockDeliverySuccess("m-agent-match");
12711292

12721293
await runSend({
12731294
to: "channel:C1",
12741295
message: "hello",
12751296
channel: "slack",
12761297
agentId: "work",
1277-
sessionKey: "agent:main:slack:channel:c1",
1278-
idempotencyKey: "idem-agent-precedence",
1298+
sessionKey: "agent:work:slack:channel:c1",
1299+
idempotencyKey: "idem-agent-match",
12791300
});
12801301

1281-
expect(deliveryCall()?.session?.agentId).toBe("work");
1282-
expect(deliveryCall()?.session?.key).toBe("agent:main:slack:channel:c1");
1283-
expect(deliveryCall()?.mirror?.sessionKey).toBe("agent:main:slack:channel:c1");
1284-
expect(deliveryCall()?.mirror?.agentId).toBe("work");
1302+
expectDeliverySessionMirror({
1303+
agentId: "work",
1304+
sessionKey: "agent:work:slack:channel:c1",
1305+
});
12851306
});
12861307

12871308
it("ignores blank explicit agentId and falls back to sessionKey agent", async () => {
@@ -1627,6 +1648,58 @@ describe("gateway send mirroring", () => {
16271648
);
16281649
});
16291650

1651+
it("rejects message.action agentId when it conflicts with the sessionKey agent", async () => {
1652+
const reactPlugin: ChannelPlugin = {
1653+
id: "whatsapp",
1654+
meta: {
1655+
id: "whatsapp",
1656+
label: "WhatsApp",
1657+
selectionLabel: "WhatsApp",
1658+
docsPath: "/channels/whatsapp",
1659+
blurb: "WhatsApp action mismatch test plugin.",
1660+
},
1661+
capabilities: { chatTypes: ["direct"], reactions: true },
1662+
config: {
1663+
listAccountIds: () => ["default"],
1664+
resolveAccount: () => ({ enabled: true }),
1665+
isConfigured: () => true,
1666+
},
1667+
actions: {
1668+
describeMessageTool: () => ({ actions: ["react"] }),
1669+
supportsAction: ({ action }) => action === "react",
1670+
handleAction: async () => jsonResult({ ok: true }),
1671+
},
1672+
};
1673+
mocks.getChannelPlugin.mockReturnValue(reactPlugin);
1674+
setActivePluginRegistry(
1675+
createTestRegistry([{ pluginId: "whatsapp", source: "test", plugin: reactPlugin }]),
1676+
"send-test-message-action-agent-mismatch",
1677+
);
1678+
1679+
const { respond } = await runMessageActionRequest({
1680+
channel: "whatsapp",
1681+
action: "react",
1682+
params: {
1683+
chatJid: "+15551234567",
1684+
messageId: "wamid.1",
1685+
emoji: "✅",
1686+
},
1687+
sessionKey: "agent:main:whatsapp:direct:15551234567",
1688+
agentId: "work",
1689+
idempotencyKey: "idem-message-action-agent-mismatch",
1690+
});
1691+
1692+
const response = firstRespondCall(respond);
1693+
expect(response?.[0]).toBe(false);
1694+
expect(response?.[1]).toBeUndefined();
1695+
expect(response?.[2]).toMatchObject({
1696+
code: "INVALID_REQUEST",
1697+
message: 'message.action agentId "work" does not match session key agent "main"',
1698+
});
1699+
expect(mocks.dispatchChannelMessageAction).not.toHaveBeenCalled();
1700+
expect(mocks.appendAssistantMessageToSessionTranscript).not.toHaveBeenCalled();
1701+
});
1702+
16301703
it("mirrors successful source-conversation message.action sends into the assistant transcript", async () => {
16311704
const telegramPlugin: ChannelPlugin = {
16321705
id: "telegram",

‎src/gateway/server-methods/send.ts‎

Lines changed: 48 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ import {
3939
projectOutboundPayloadPlanForMirror,
4040
} from "../../infra/outbound/payloads.js";
4141
import { buildOutboundSessionContext } from "../../infra/outbound/session-context.js";
42+
import { resolveAgentSessionOwnerMismatch } from "../../infra/outbound/session-owner.js";
4243
import { mirrorDeliveredSourceReplyToTranscript } from "../../infra/outbound/source-reply-mirror.js";
4344
import { maybeResolveIdLikeTarget } from "../../infra/outbound/target-resolver.js";
4445
import { resolveOutboundTarget } from "../../infra/outbound/targets.js";
@@ -396,6 +397,25 @@ function createGatewayInflightUnavailableFailure(params: {
396397
};
397398
}
398399

400+
function createGatewayInflightInvalidRequestFailure(params: {
401+
context: GatewayRequestContext;
402+
dedupeKey: string;
403+
channel: string;
404+
message: string;
405+
}): InflightResult {
406+
const error = errorShape(ErrorCodes.INVALID_REQUEST, params.message);
407+
cacheGatewayDedupeFailure({
408+
context: params.context,
409+
dedupeKey: params.dedupeKey,
410+
error,
411+
});
412+
return {
413+
ok: false,
414+
error,
415+
meta: { channel: params.channel },
416+
};
417+
}
418+
399419
async function mirrorDeliveredSourceReplyToTranscriptBestEffort(params: {
400420
context: GatewayRequestContext;
401421
mirror: Parameters<typeof mirrorDeliveredSourceReplyToTranscript>[0];
@@ -519,8 +539,22 @@ export const sendHandlers: GatewayRequestHandlers = {
519539

520540
try {
521541
const sessionKey = normalizeOptionalString(request.sessionKey) ?? undefined;
542+
const explicitAgentId = normalizeOptionalString(request.agentId);
543+
const ownerMismatch = resolveAgentSessionOwnerMismatch({
544+
ownerLabel: "message.action",
545+
agentId: explicitAgentId,
546+
sessionKey,
547+
});
548+
if (ownerMismatch) {
549+
return createGatewayInflightInvalidRequestFailure({
550+
context,
551+
dedupeKey,
552+
channel,
553+
message: ownerMismatch.message,
554+
});
555+
}
522556
const agentId =
523-
normalizeOptionalString(request.agentId) ??
557+
explicitAgentId ??
524558
(sessionKey ? resolveSessionAgentId({ sessionKey, config: cfg }) : undefined);
525559
const accountId = normalizeOptionalString(request.accountId) ?? undefined;
526560
if (request.action === "send") {
@@ -695,6 +729,19 @@ export const sendHandlers: GatewayRequestHandlers = {
695729
const providedSessionKey =
696730
normalizeSessionKeyPreservingOpaquePeerIds(request.sessionKey) || undefined;
697731
const explicitAgentId = normalizeOptionalString(request.agentId);
732+
const ownerMismatch = resolveAgentSessionOwnerMismatch({
733+
ownerLabel: "send",
734+
agentId: explicitAgentId,
735+
sessionKey: providedSessionKey,
736+
});
737+
if (ownerMismatch) {
738+
return createGatewayInflightInvalidRequestFailure({
739+
context,
740+
dedupeKey,
741+
channel,
742+
message: ownerMismatch.message,
743+
});
744+
}
698745
const sessionAgentId = providedSessionKey
699746
? resolveSessionAgentId({ sessionKey: providedSessionKey, config: cfg })
700747
: undefined;

‎src/infra/outbound/deliver.test.ts‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -374,6 +374,100 @@ describe("deliverOutboundPayloads", () => {
374374
expect(results).toEqual([{ channel: "matrix", messageId: "m1", roomId: "!room:example" }]);
375375
});
376376

377+
it("rejects session agentId that contradicts session key before queueing", async () => {
378+
const sendMatrix = vi.fn().mockResolvedValue({ messageId: "m1", roomId: "!room:example" });
379+
380+
await expect(
381+
deliverOutboundPayloads({
382+
cfg: matrixChunkConfig,
383+
channel: "matrix",
384+
to: "!room:example",
385+
payloads: [{ text: "hello" }],
386+
deps: { matrix: sendMatrix },
387+
session: {
388+
agentId: "work",
389+
key: "agent:main:matrix:room:ops",
390+
},
391+
}),
392+
).rejects.toThrow(
393+
'deliverOutboundPayloads agentId "work" does not match session key agent "main"',
394+
);
395+
396+
expect(queueMocks.enqueueDelivery).not.toHaveBeenCalled();
397+
expect(sendMatrix).not.toHaveBeenCalled();
398+
});
399+
400+
it("rejects mirror agentId that contradicts mirror sessionKey before queueing", async () => {
401+
const sendMatrix = vi.fn().mockResolvedValue({ messageId: "m1", roomId: "!room:example" });
402+
403+
await expect(
404+
deliverOutboundPayloads({
405+
cfg: matrixChunkConfig,
406+
channel: "matrix",
407+
to: "!room:example",
408+
payloads: [{ text: "hello" }],
409+
deps: { matrix: sendMatrix },
410+
mirror: {
411+
agentId: "work",
412+
sessionKey: "agent:main:matrix:room:ops",
413+
},
414+
}),
415+
).rejects.toThrow(
416+
'deliverOutboundPayloads agentId "work" does not match mirror session key agent "main"',
417+
);
418+
419+
expect(queueMocks.enqueueDelivery).not.toHaveBeenCalled();
420+
expect(sendMatrix).not.toHaveBeenCalled();
421+
});
422+
423+
it("rejects session and mirror keys owned by different agents before queueing", async () => {
424+
const sendMatrix = vi.fn().mockResolvedValue({ messageId: "m1", roomId: "!room:example" });
425+
426+
await expect(
427+
deliverOutboundPayloads({
428+
cfg: matrixChunkConfig,
429+
channel: "matrix",
430+
to: "!room:example",
431+
payloads: [{ text: "hello" }],
432+
deps: { matrix: sendMatrix },
433+
session: {
434+
key: "agent:work:matrix:room:ops",
435+
},
436+
mirror: {
437+
sessionKey: "agent:main:matrix:room:ops",
438+
},
439+
}),
440+
).rejects.toThrow(
441+
'deliverOutboundPayloads session key agent "work" does not match mirror session key agent "main"',
442+
);
443+
444+
expect(queueMocks.enqueueDelivery).not.toHaveBeenCalled();
445+
expect(sendMatrix).not.toHaveBeenCalled();
446+
});
447+
448+
it("allows policy session keys owned by another agent", async () => {
449+
const sendMatrix = vi.fn().mockResolvedValue({ messageId: "m1", roomId: "!room:example" });
450+
451+
await deliverOutboundPayloads({
452+
cfg: matrixChunkConfig,
453+
channel: "matrix",
454+
to: "!room:example",
455+
payloads: [{ text: "hello" }],
456+
deps: { matrix: sendMatrix },
457+
session: {
458+
agentId: "claude",
459+
key: "agent:claude:acp:spawned",
460+
policyKey: "agent:main:main",
461+
},
462+
});
463+
464+
expect(sendMatrix).toHaveBeenCalledWith("!room:example", "hello", {
465+
cfg: matrixChunkConfig,
466+
accountId: undefined,
467+
gifPlayback: undefined,
468+
});
469+
});
470+
377471
it("reports unsupported durable final delivery when required capabilities are missing", async () => {
378472
setActivePluginRegistry(
379473
createTestRegistry([

‎src/infra/outbound/deliver.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,7 @@ import { createReplyToDeliveryPolicy } from "./reply-policy.js";
8989
import { stripInternalRuntimeScaffolding } from "./sanitize-text.js";
9090
import type { OutboundSendDeps } from "./send-deps.js";
9191
import type { OutboundSessionContext } from "./session-context.js";
92+
import { assertAgentSessionOwnership } from "./session-owner.js";
9293
import type { OutboundChannel } from "./targets.js";
9394

9495
export type { OutboundDeliveryResult } from "./deliver-types.js";
@@ -704,6 +705,20 @@ function sessionKeyForDeliveryDiagnostics(params: {
704705
return params.mirror?.sessionKey ?? params.session?.key ?? params.session?.policyKey;
705706
}
706707

708+
function assertDeliveryOwnership(params: {
709+
session?: OutboundSessionContext;
710+
mirror?: DeliveryMirror;
711+
}): void {
712+
assertAgentSessionOwnership({
713+
ownerLabel: "deliverOutboundPayloads",
714+
agentIds: [params.session?.agentId, params.mirror?.agentId],
715+
sessionKeys: [
716+
{ sessionKey: params.session?.key, label: "session key" },
717+
{ sessionKey: params.mirror?.sessionKey, label: "mirror session key" },
718+
],
719+
});
720+
}
721+
707722
function deliveryKindForPayload(
708723
payload: ReplyPayload,
709724
payloadSummary: NormalizedOutboundPayload,
@@ -1245,6 +1260,7 @@ export async function deliverOutboundPayloadsInternal(
12451260
params: DeliverOutboundPayloadsParams,
12461261
): Promise<OutboundDeliveryResult[]> {
12471262
const { channel, to, payloads } = params;
1263+
assertDeliveryOwnership(params);
12481264
const queuePolicy = params.queuePolicy ?? "best_effort";
12491265
const queuePayloads = payloads.map(stripInternalRuntimeScaffoldingFromPayload);
12501266
const queuePayloadsChanged = queuePayloads.some((payload, index) => payload !== payloads[index]);

‎src/infra/outbound/message-action-runner.plugin-dispatch.test.ts‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -333,6 +333,30 @@ describe("runMessageAction plugin dispatch", () => {
333333
vi.unstubAllEnvs();
334334
});
335335

336+
it("rejects explicit agentId that contradicts sessionKey owner before plugin dispatch", async () => {
337+
await expect(
338+
runMessageAction({
339+
cfg: {
340+
channels: {
341+
actionhub: {
342+
enabled: true,
343+
},
344+
},
345+
} as OpenClawConfig,
346+
action: "pin",
347+
params: {
348+
channel: "actionhub",
349+
messageId: "om_123",
350+
},
351+
sessionKey: "agent:main:actionhub:direct:oc_123",
352+
agentId: "work",
353+
dryRun: false,
354+
}),
355+
).rejects.toThrow('message action agentId "work" does not match session key agent "main"');
356+
357+
expect(handleAction).not.toHaveBeenCalled();
358+
});
359+
336360
it("dispatches messageId/chatId-based plugin actions through the shared runner", async () => {
337361
await runMessageAction({
338362
cfg: {

‎src/infra/outbound/message-action-runner.send-validation.test.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,24 @@ describe("runMessageAction send validation", () => {
3636
setActivePluginRegistry(createTestRegistry([]));
3737
});
3838

39+
it("rejects explicit agentId that contradicts sessionKey owner", async () => {
40+
await expect(
41+
runMessageAction({
42+
cfg: emptyConfig,
43+
action: "send",
44+
params: {
45+
message: "hello from codex",
46+
},
47+
toolContext: {
48+
currentChannelProvider: "webchat",
49+
},
50+
sessionKey: "agent:main:webchat:direct:current-run",
51+
agentId: "work",
52+
sourceReplyDeliveryMode: "message_tool_only",
53+
}),
54+
).rejects.toThrow('message action agentId "work" does not match session key agent "main"');
55+
});
56+
3957
it("requires message when no media hint is provided", async () => {
4058
await expect(
4159
runDrySend({

0 commit comments

Comments
 (0)