Skip to content

Commit 67dece7

Browse files
youngting520youngting520
committed
fix(outbound): reject split agent session ownership
1 parent 0ac7b24 commit 67dece7

17 files changed

Lines changed: 1035 additions & 26 deletions

‎docs/gateway/protocol.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -488,6 +488,10 @@ methods. Treat this as feature discovery, not a full enumeration of
488488

489489
<Accordion title="Messaging and logs">
490490
- `send` is the direct outbound-delivery RPC for channel/account/thread-targeted sends outside the chat runner.
491+
- `send` and `message.action` reject a request when an explicit `agentId`
492+
does not own its agent-scoped `sessionKey`. Agent-scoped keys must use the
493+
complete `agent:<agentId>:<session>` shape; malformed `agent:` keys are
494+
rejected, while legacy unscoped session keys remain valid.
491495
- `logs.tail` returns the configured gateway file-log tail with cursor/limit and max-byte controls.
492496

493497
</Accordion>

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

Lines changed: 137 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1510,6 +1510,26 @@ describe("gateway send mirroring", () => {
15101510
});
15111511
});
15121512

1513+
it("keeps explicit agent ownership for legacy unscoped session keys", async () => {
1514+
mockDeliverySuccess("m-legacy-unscoped-agent");
1515+
1516+
await runSend({
1517+
to: "channel:C1",
1518+
message: "hello",
1519+
channel: "slack",
1520+
agentId: "work",
1521+
sessionKey: "legacy:slack:channel:c1",
1522+
idempotencyKey: "idem-legacy-unscoped-agent",
1523+
});
1524+
1525+
expect(outboundRouteCall()?.agentId).toBe("work");
1526+
expect(outboundRouteCall()?.currentSessionKey).toBe("legacy:slack:channel:c1");
1527+
expectDeliverySessionMirror({
1528+
agentId: "work",
1529+
sessionKey: "legacy:slack:channel:c1",
1530+
});
1531+
});
1532+
15131533
it("rejects a missing reserved agent-harness session before persistence or delivery", async () => {
15141534
const sessionKey = "agent:main:harness:codex:supervision:missing";
15151535

@@ -1606,22 +1626,84 @@ describe("gateway send mirroring", () => {
16061626
expect(deliveryCall()?.mirror?.agentId).toBe("work");
16071627
});
16081628

1609-
it("prefers explicit agentId over sessionKey agent for delivery and mirror", async () => {
1610-
mockDeliverySuccess("m-agent-precedence");
1611-
1612-
await runSend({
1629+
it("rejects explicit agentId when it conflicts with the sessionKey agent", async () => {
1630+
const { respond } = await runSend({
16131631
to: "channel:C1",
16141632
message: "hello",
16151633
channel: "slack",
16161634
agentId: "work",
16171635
sessionKey: "agent:main:slack:channel:c1",
1618-
idempotencyKey: "idem-agent-precedence",
1636+
idempotencyKey: "idem-agent-mismatch",
16191637
});
16201638

1621-
expect(deliveryCall()?.session?.agentId).toBe("work");
1622-
expect(deliveryCall()?.session?.key).toBe("agent:main:slack:channel:c1");
1623-
expect(deliveryCall()?.mirror?.sessionKey).toBe("agent:main:slack:channel:c1");
1624-
expect(deliveryCall()?.mirror?.agentId).toBe("work");
1639+
expect(firstRespondCall(respond)).toMatchObject([
1640+
false,
1641+
undefined,
1642+
{
1643+
code: "INVALID_REQUEST",
1644+
message: 'send agentId "work" does not match session key agent "main"',
1645+
},
1646+
undefined,
1647+
]);
1648+
expect(mocks.getChannelPlugin).not.toHaveBeenCalled();
1649+
expect(mocks.resolveOutboundTarget).not.toHaveBeenCalled();
1650+
expect(mocks.resolveOutboundSessionRoute).not.toHaveBeenCalled();
1651+
expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled();
1652+
});
1653+
1654+
it.each(["agent::broken", "agent:main"])(
1655+
"rejects malformed agent-scoped send session key %s before channel work",
1656+
async (sessionKey) => {
1657+
const { respond } = await runSend({
1658+
to: "channel:C1",
1659+
message: "hello",
1660+
channel: "slack",
1661+
sessionKey,
1662+
idempotencyKey: `idem-malformed-${sessionKey}`,
1663+
});
1664+
1665+
expect(firstRespondCall(respond)?.[2]).toMatchObject({
1666+
code: "INVALID_REQUEST",
1667+
message: expect.stringContaining("is malformed"),
1668+
});
1669+
expect(mocks.getChannelPlugin).not.toHaveBeenCalled();
1670+
expect(mocks.resolveOutboundTarget).not.toHaveBeenCalled();
1671+
expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled();
1672+
},
1673+
);
1674+
1675+
it("caches ownership validation failures under the send idempotency key", async () => {
1676+
const context = makeContext();
1677+
const invoke = async () => {
1678+
const respond = vi.fn();
1679+
await expectDefined(sendHandlers.send, "sendHandlers.send test invariant").call(
1680+
sendHandlers,
1681+
{
1682+
params: {
1683+
to: "channel:C1",
1684+
message: "hello",
1685+
channel: "slack",
1686+
agentId: "work",
1687+
sessionKey: "agent:main:slack:channel:c1",
1688+
idempotencyKey: "idem-agent-mismatch-cached",
1689+
} as never,
1690+
respond,
1691+
context,
1692+
req: { type: "req", id: "1", method: "send" },
1693+
client: null as never,
1694+
isWebchatConnect: () => false,
1695+
},
1696+
);
1697+
return respond;
1698+
};
1699+
1700+
const firstRespond = await invoke();
1701+
const secondRespond = await invoke();
1702+
1703+
expect(firstRespondCall(firstRespond)?.[2]?.code).toBe("INVALID_REQUEST");
1704+
expect(firstRespondCall(secondRespond)?.[3]).toEqual({ cached: true });
1705+
expect(mocks.getChannelPlugin).not.toHaveBeenCalled();
1706+
expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled();
16251707
});
16261708

16271709
it("ignores blank explicit agentId and falls back to sessionKey agent", async () => {
@@ -2005,6 +2087,52 @@ describe("gateway send mirroring", () => {
20052087
);
20062088
});
20072089

2090+
it("rejects message.action agentId when it conflicts with the sessionKey agent", async () => {
2091+
const { respond } = await runMessageActionRequest(
2092+
{
2093+
channel: "whatsapp",
2094+
action: "react",
2095+
params: { messageId: "wamid.1", emoji: "ok" },
2096+
sessionKey: "agent:main:whatsapp:direct:15551234567",
2097+
agentId: "work",
2098+
idempotencyKey: "idem-message-action-agent-mismatch",
2099+
},
2100+
null,
2101+
);
2102+
2103+
expect(firstRespondCall(respond)).toMatchObject([
2104+
false,
2105+
undefined,
2106+
{
2107+
code: "INVALID_REQUEST",
2108+
message: 'message.action agentId "work" does not match session key agent "main"',
2109+
},
2110+
undefined,
2111+
]);
2112+
expect(mocks.getChannelPlugin).not.toHaveBeenCalled();
2113+
expect(mocks.dispatchChannelMessageAction).not.toHaveBeenCalled();
2114+
});
2115+
2116+
it("rejects malformed message.action session keys before plugin discovery", async () => {
2117+
const { respond } = await runMessageActionRequest(
2118+
{
2119+
channel: "whatsapp",
2120+
action: "react",
2121+
params: { messageId: "wamid.1", emoji: "ok" },
2122+
sessionKey: "agent::broken",
2123+
idempotencyKey: "idem-message-action-malformed-session",
2124+
},
2125+
null,
2126+
);
2127+
2128+
expect(firstRespondCall(respond)?.[2]).toMatchObject({
2129+
code: "INVALID_REQUEST",
2130+
message: expect.stringContaining("is malformed"),
2131+
});
2132+
expect(mocks.getChannelPlugin).not.toHaveBeenCalled();
2133+
expect(mocks.dispatchChannelMessageAction).not.toHaveBeenCalled();
2134+
});
2135+
20082136
it("strips current-turn context from unauthenticated message action callers", async () => {
20092137
mocks.getChannelPlugin.mockReturnValue({
20102138
actions: {

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

Lines changed: 45 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ import {
4242
projectOutboundPayloadPlanForMirror,
4343
} from "../../infra/outbound/payloads.js";
4444
import { buildOutboundSessionContext } from "../../infra/outbound/session-context.js";
45+
import { validateAgentSessionOwnerPair } from "../../infra/outbound/session-owner.js";
4546
import {
4647
beginTerminalSourceReplyDelivery,
4748
cancelTerminalSourceReplyDelivery,
@@ -416,6 +417,20 @@ function createGatewayInflightUnavailableFailure(params: {
416417
};
417418
}
418419

420+
function createGatewayInflightInvalidRequestFailure(params: {
421+
context: GatewayRequestContext;
422+
dedupeKey: string;
423+
message: string;
424+
}): InflightResult {
425+
const error = errorShape(ErrorCodes.INVALID_REQUEST, params.message);
426+
cacheGatewayDedupeFailure({
427+
context: params.context,
428+
dedupeKey: params.dedupeKey,
429+
error,
430+
});
431+
return { ok: false, error };
432+
}
433+
419434
async function mirrorDeliveredSourceReplyToTranscriptBestEffort(params: {
420435
context: GatewayRequestContext;
421436
mirror: Parameters<typeof mirrorDeliveredSourceReplyToTranscript>[0];
@@ -513,6 +528,20 @@ export const sendHandlers: GatewayRequestHandlers = {
513528
}
514529
const { dedupeKey, inflightMap } = inflight;
515530
const work = (async (): Promise<InflightResult> => {
531+
const sessionKey = normalizeOptionalString(request.sessionKey) ?? undefined;
532+
const explicitAgentId = normalizeOptionalString(request.agentId);
533+
const ownership = validateAgentSessionOwnerPair({
534+
ownerLabel: "message.action",
535+
agentId: explicitAgentId,
536+
sessionKey,
537+
});
538+
if (!ownership.ok) {
539+
return createGatewayInflightInvalidRequestFailure({
540+
context,
541+
dedupeKey,
542+
message: ownership.error.message,
543+
});
544+
}
516545
const resolvedChannel = await resolveRequestedChannel({
517546
requestChannel: request.channel,
518547
unsupportedMessage: (input) => `unsupported channel: ${input}`,
@@ -536,9 +565,8 @@ export const sendHandlers: GatewayRequestHandlers = {
536565
}
537566

538567
try {
539-
const sessionKey = normalizeOptionalString(request.sessionKey) ?? undefined;
540568
const agentId =
541-
normalizeOptionalString(request.agentId) ??
569+
explicitAgentId ??
542570
(sessionKey ? resolveSessionAgentId({ sessionKey, config: cfg }) : undefined);
543571
const accountId = normalizeOptionalString(request.accountId) ?? undefined;
544572
if (request.action === "send") {
@@ -703,6 +731,21 @@ export const sendHandlers: GatewayRequestHandlers = {
703731
const threadId = normalizeOptionalString(request.threadId);
704732

705733
const work = (async (): Promise<InflightResult> => {
734+
const providedSessionKey =
735+
normalizeSessionKeyPreservingOpaquePeerIds(request.sessionKey) || undefined;
736+
const explicitAgentId = normalizeOptionalString(request.agentId);
737+
const ownership = validateAgentSessionOwnerPair({
738+
ownerLabel: "send",
739+
agentId: explicitAgentId,
740+
sessionKey: providedSessionKey,
741+
});
742+
if (!ownership.ok) {
743+
return createGatewayInflightInvalidRequestFailure({
744+
context,
745+
dedupeKey,
746+
message: ownership.error.message,
747+
});
748+
}
706749
const resolvedChannel = await resolveInternalDeliveryChannel(request.channel, context);
707750
if (resolvedChannel.kind !== "ready") {
708751
return resolvedChannel.result;
@@ -741,9 +784,6 @@ export const sendHandlers: GatewayRequestHandlers = {
741784
// Preserve opaque, case-sensitive peer IDs (e.g. Matrix room ids) on an
742785
// explicit session key instead of raw-lowercasing it (openclaw#75670).
743786
// Non-enrolled channels still canonicalize to lowercase via the registry.
744-
const providedSessionKey =
745-
normalizeSessionKeyPreservingOpaquePeerIds(request.sessionKey) || undefined;
746-
const explicitAgentId = normalizeOptionalString(request.agentId);
747787
const sessionAgentId = providedSessionKey
748788
? resolveSessionAgentId({ sessionKey: providedSessionKey, config: cfg })
749789
: undefined;

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

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -417,6 +417,89 @@ describe("deliverOutboundPayloads", () => {
417417
expect(results).toEqual([{ channel: "matrix", messageId: "m1", roomId: "!room:example" }]);
418418
});
419419

420+
it("rejects mismatched session ownership before queue or platform I/O", async () => {
421+
const events: TrustedMessageAuditEvent[] = [];
422+
const unsubscribe = onTrustedMessageAuditEvent((event) => events.push(event));
423+
const sendMatrix = vi.fn().mockResolvedValue({ messageId: "m1", roomId: "!room:example" });
424+
425+
try {
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+
agentId: "work",
435+
key: "agent:main:matrix:room:ops",
436+
},
437+
}),
438+
).rejects.toThrow(
439+
'deliverOutboundPayloads agentId "work" does not match session key agent "main"',
440+
);
441+
} finally {
442+
unsubscribe();
443+
}
444+
445+
expect(queueMocks.enqueueDelivery).not.toHaveBeenCalled();
446+
expect(queueMocks.enqueueDeliveryOnce).not.toHaveBeenCalled();
447+
expect(sendMatrix).not.toHaveBeenCalled();
448+
expect(events).toHaveLength(1);
449+
expect(events[0]).toMatchObject({ outcome: "failed", failureStage: "queue" });
450+
});
451+
452+
it("accepts same-owner control and transcript keys with a cross-agent policy key", async () => {
453+
const sendMatrix = vi.fn().mockResolvedValue({ messageId: "m1", roomId: "!room:example" });
454+
455+
await deliverOutboundPayloads({
456+
cfg: matrixChunkConfig,
457+
channel: "matrix",
458+
to: "!room:example",
459+
payloads: [{ text: "hello" }],
460+
deps: { matrix: sendMatrix },
461+
session: {
462+
agentId: "controller",
463+
key: "agent:controller:acp:spawned",
464+
policyKey: "agent:policy-owner:matrix:room:ops",
465+
},
466+
mirror: {
467+
agentId: "controller",
468+
sessionKey: "agent:controller:matrix:room:ops",
469+
},
470+
});
471+
472+
expect(sendMatrix).toHaveBeenCalledOnce();
473+
});
474+
475+
it("rejects different control and transcript owners before queueing or sending", async () => {
476+
const sendMatrix = vi.fn().mockResolvedValue({ messageId: "m1", roomId: "!room:example" });
477+
478+
await expect(
479+
deliverOutboundPayloads({
480+
cfg: matrixChunkConfig,
481+
channel: "matrix",
482+
to: "!room:example",
483+
payloads: [{ text: "hello" }],
484+
deps: { matrix: sendMatrix },
485+
session: {
486+
agentId: "controller",
487+
key: "agent:controller:acp:spawned",
488+
policyKey: "agent:policy-owner:matrix:room:ops",
489+
},
490+
mirror: {
491+
sessionKey: "agent:transcript:matrix:room:ops",
492+
},
493+
}),
494+
).rejects.toThrow(
495+
'deliverOutboundPayloads mirror session key agent "transcript" does not match operation agent "controller"',
496+
);
497+
498+
expect(queueMocks.enqueueDelivery).not.toHaveBeenCalled();
499+
expect(queueMocks.enqueueDeliveryOnce).not.toHaveBeenCalled();
500+
expect(sendMatrix).not.toHaveBeenCalled();
501+
});
502+
420503
it("reports unsupported durable final delivery when required capabilities are missing", async () => {
421504
setActivePluginRegistry(
422505
createTestRegistry([

‎src/infra/outbound/deliver.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ import { stripInternalRuntimeScaffolding } from "./protocol-scaffolding.js";
119119
import { createReplyToDeliveryPolicy } from "./reply-policy.js";
120120
import type { OutboundSendDeps } from "./send-deps.js";
121121
import type { OutboundSessionContext } from "./session-context.js";
122+
import { assertOutboundDeliverySessionOwnership } from "./session-owner.js";
122123
import type { OutboundChannel } from "./targets.js";
123124

124125
export type { OutboundDeliveryResult } from "./deliver-types.js";
@@ -1476,6 +1477,16 @@ export async function deliverOutboundPayloadsInternal(
14761477
startedAt: auditStartedAt,
14771478
});
14781479
};
1480+
try {
1481+
assertOutboundDeliverySessionOwnership({
1482+
ownerLabel: "deliverOutboundPayloads",
1483+
session: params.session,
1484+
mirror: params.mirror,
1485+
});
1486+
} catch (error) {
1487+
emitPreQueueFailure();
1488+
throw error;
1489+
}
14791490
if (params.requireUnknownSendReconciliation === true && payloads.length !== 1) {
14801491
emitPreQueueFailure();
14811492
throw new Error(

0 commit comments

Comments
 (0)