Skip to content

Commit 4c23d1d

Browse files
zhangguiping-xydtzhangguiping-xydt
authored
fix(agents): narrow sessions_send active-run fallback
Limit sessions_send active-run queue delivery to run-scoped targets, keep stranded cron-run fallback for valid cron run keys, and report unsafe queue rejections without rerouting through durable sessions. Fixes #91420.
1 parent 8eb1fa0 commit 4c23d1d

2 files changed

Lines changed: 285 additions & 69 deletions

File tree

‎src/agents/openclaw-tools.sessions.test.ts‎

Lines changed: 233 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -1513,11 +1513,10 @@ describe("sessions tools", () => {
15131513
expect(calls.find((call) => call.method === "send")).toBeUndefined();
15141514
});
15151515

1516-
it("sessions_send reroutes run-scoped active deliveries when transcript steering is rejected", async () => {
1516+
it("sessions_send reports active-run queue rejection without durable-session fallback", async () => {
15171517
const calls: Array<{ method?: string; params?: unknown }> = [];
15181518
const requesterKey = "agent:re-portal:main";
15191519
const runScopedCallerKey = "agent:leasing-ops:cron:monthly-utility:run:run-fast";
1520-
const durableCallerKey = "agent:leasing-ops:cron:monthly-utility";
15211520
const queueMessage = vi.fn(async (_text: string, _options?: unknown) => {
15221521
throw new Error("active session ended before queued steering message was committed");
15231522
});
@@ -1539,13 +1538,6 @@ describe("sessions tools", () => {
15391538
if (request.method === "agent") {
15401539
return { runId: "fallback-run", status: "accepted", acceptedAt: 2000 };
15411540
}
1542-
if (request.method === "agent.wait") {
1543-
const params = request.params as { runId?: string } | undefined;
1544-
return { runId: params?.runId ?? "fallback-run", status: "ok" };
1545-
}
1546-
if (request.method === "chat.history") {
1547-
return { messages: [] };
1548-
}
15491541
return {};
15501542
});
15511543

@@ -1570,9 +1562,11 @@ describe("sessions tools", () => {
15701562
timeoutSeconds: 0,
15711563
});
15721564
const details = sessionsSendDetails(result.details);
1573-
expect(details.status).toBe("accepted");
1565+
expect(details.status).toBe("error");
15741566
expect(details.sessionKey).toBe(runScopedCallerKey);
1575-
expect(details.delivery?.status).toBe("pending");
1567+
expect(details.error).toContain("queue_message_failed reason=runtime_rejected");
1568+
expect(details.error).toContain("caller-active-session");
1569+
expect(details.error).not.toContain("fallback_failed");
15761570
const queuedText = queueMessage.mock.calls[0]?.[0];
15771571
expect(queuedText).toContain("[Inter-session message]");
15781572
expect(queuedText).toContain("[TASK-COMPLETE] re-portal occupancy ready");
@@ -1583,47 +1577,233 @@ describe("sessions tools", () => {
15831577
waitForTranscriptCommit: true,
15841578
sourceReplyDeliveryMode: "message_tool_only",
15851579
});
1580+
expect(calls.some((call) => call.method === "agent")).toBe(false);
1581+
});
15861582

1587-
await vi.waitFor(() => {
1588-
const fallbackCall = calls.find(
1589-
(call) =>
1590-
call.method === "agent" &&
1591-
(call.params as { sessionKey?: string } | undefined)?.sessionKey === durableCallerKey,
1592-
);
1593-
expect(fallbackCall).toBeDefined();
1583+
it("sessions_send reports source reply delivery mode mismatch without durable-session fallback", async () => {
1584+
const calls: Array<{ method?: string; params?: unknown }> = [];
1585+
const runScopedCallerKey = "agent:leasing-ops:cron:monthly-utility:run:run-fast";
1586+
const queueMessage = vi.fn(async () => {});
1587+
setActiveEmbeddedRun(
1588+
"caller-active-session",
1589+
{
1590+
queueMessage,
1591+
isStreaming: () => true,
1592+
isCompacting: () => false,
1593+
supportsTranscriptCommitWait: true,
1594+
sourceReplyDeliveryMode: "automatic",
1595+
abort: () => {},
1596+
},
1597+
runScopedCallerKey,
1598+
);
1599+
callGatewayMock.mockImplementation(async (opts: unknown) => {
1600+
const request = opts as { method?: string; params?: unknown };
1601+
calls.push(request);
1602+
if (request.method === "agent") {
1603+
return { runId: "fallback-run", status: "accepted", acceptedAt: 2000 };
1604+
}
1605+
return {};
15941606
});
15951607

1608+
const tool = createOpenClawTools({
1609+
agentSessionKey: "agent:re-portal:main",
1610+
agentChannel: "telegram",
1611+
config: {
1612+
...TEST_CONFIG,
1613+
session: {
1614+
...TEST_CONFIG.session,
1615+
agentToAgent: { maxPingPongTurns: 0 },
1616+
},
1617+
},
1618+
}).find((candidate) => candidate.name === "sessions_send");
1619+
if (!tool) {
1620+
throw new Error("missing sessions_send tool");
1621+
}
1622+
1623+
const result = await tool.execute("call-run-scoped-caller", {
1624+
sessionKey: runScopedCallerKey,
1625+
message: "[TASK-COMPLETE] re-portal occupancy ready",
1626+
timeoutSeconds: 0,
1627+
});
1628+
1629+
const details = sessionsSendDetails(result.details);
1630+
expect(details.status).toBe("error");
1631+
expect(details.sessionKey).toBe(runScopedCallerKey);
1632+
expect(details.error).toContain(
1633+
"queue_message_failed reason=source_reply_delivery_mode_mismatch",
1634+
);
1635+
expect(queueMessage).not.toHaveBeenCalled();
1636+
expect(calls.some((call) => call.method === "agent")).toBe(false);
1637+
});
1638+
1639+
it("sessions_send keeps ordinary active session targets on the gateway agent path", async () => {
1640+
const calls: Array<{ method?: string; params?: unknown }> = [];
1641+
const ordinaryActiveKey = "agent:main:main";
1642+
const queueMessage = vi.fn(async () => {});
1643+
setActiveEmbeddedRun(
1644+
"ordinary-active-session",
1645+
{
1646+
queueMessage,
1647+
isStreaming: () => true,
1648+
isCompacting: () => false,
1649+
supportsTranscriptCommitWait: true,
1650+
sourceReplyDeliveryMode: "automatic",
1651+
abort: () => {},
1652+
},
1653+
ordinaryActiveKey,
1654+
);
1655+
callGatewayMock.mockImplementation(async (opts: unknown) => {
1656+
const request = opts as { method?: string; params?: unknown };
1657+
calls.push(request);
1658+
if (request.method === "agent") {
1659+
return { runId: "ordinary-agent-run", status: "accepted", acceptedAt: 2000 };
1660+
}
1661+
return {};
1662+
});
1663+
1664+
const tool = createOpenClawTools({
1665+
agentSessionKey: "agent:re-portal:main",
1666+
agentChannel: "telegram",
1667+
config: {
1668+
...TEST_CONFIG,
1669+
session: {
1670+
...TEST_CONFIG.session,
1671+
agentToAgent: { maxPingPongTurns: 0 },
1672+
},
1673+
},
1674+
}).find((candidate) => candidate.name === "sessions_send");
1675+
if (!tool) {
1676+
throw new Error("missing sessions_send tool");
1677+
}
1678+
1679+
const result = await tool.execute("call-ordinary-active", {
1680+
sessionKey: ordinaryActiveKey,
1681+
message: "ordinary active target should stay gateway routed",
1682+
timeoutSeconds: 0,
1683+
});
1684+
1685+
const details = sessionsSendDetails(result.details);
1686+
expect(details.status).toBe("accepted");
1687+
expect(details.runId).toBe("ordinary-agent-run");
1688+
expect(details.sessionKey).toBe(ordinaryActiveKey);
1689+
expect(queueMessage).not.toHaveBeenCalled();
15961690
const agentCalls = calls.filter((call) => call.method === "agent");
1597-
expect(
1598-
agentCalls.some(
1599-
(call) =>
1600-
(call.params as { sessionKey?: string } | undefined)?.sessionKey === runScopedCallerKey,
1601-
),
1602-
).toBe(false);
1603-
const fallbackParams = agentCalls.find(
1604-
(call) =>
1605-
(call.params as { sessionKey?: string } | undefined)?.sessionKey === durableCallerKey,
1606-
)?.params as { inputProvenance?: { sourceSessionKey?: string }; message?: string } | undefined;
1607-
expect(fallbackParams?.message).toContain("[Inter-session message]");
1608-
expect(fallbackParams?.message).toContain("[TASK-COMPLETE] re-portal occupancy ready");
1609-
expect(fallbackParams?.inputProvenance?.sourceSessionKey).toBe(requesterKey);
1610-
1611-
await vi.waitFor(() => {
1612-
const waitCall = calls.find(
1613-
(call) =>
1614-
call.method === "agent.wait" &&
1615-
(call.params as { runId?: string } | undefined)?.runId === "fallback-run",
1616-
);
1617-
expect(waitCall).toBeDefined();
1691+
expect(agentCalls).toHaveLength(1);
1692+
expect(agentParams(agentCalls[0] ?? {}).sessionKey).toBe(ordinaryActiveKey);
1693+
});
1694+
1695+
it("sessions_send falls back from stranded cron run key to durable cron parent", async () => {
1696+
const calls: Array<{ method?: string; params?: unknown }> = [];
1697+
const runScopedCallerKey = "agent:leasing-ops:cron:monthly-utility:run:run-fast";
1698+
const durableCronCallerKey = "agent:leasing-ops:cron:monthly-utility";
1699+
const queueMessage = vi.fn(async () => {});
1700+
setActiveEmbeddedRun(
1701+
"caller-active-session",
1702+
{
1703+
queueMessage,
1704+
isStreaming: () => false,
1705+
isCompacting: () => false,
1706+
supportsTranscriptCommitWait: true,
1707+
sourceReplyDeliveryMode: "message_tool_only",
1708+
abort: () => {},
1709+
},
1710+
runScopedCallerKey,
1711+
);
1712+
callGatewayMock.mockImplementation(async (opts: unknown) => {
1713+
const request = opts as { method?: string; params?: unknown };
1714+
calls.push(request);
1715+
if (request.method === "agent") {
1716+
return { runId: "durable-fallback-run", status: "accepted", acceptedAt: 2000 };
1717+
}
1718+
return {};
16181719
});
1619-
await vi.waitFor(() => {
1620-
const historyCall = calls.find(
1621-
(call) =>
1622-
call.method === "chat.history" &&
1623-
(call.params as { sessionKey?: string } | undefined)?.sessionKey === durableCallerKey,
1624-
);
1625-
expect(historyCall).toBeDefined();
1720+
1721+
const tool = createOpenClawTools({
1722+
agentSessionKey: "agent:re-portal:main",
1723+
agentChannel: "telegram",
1724+
config: {
1725+
...TEST_CONFIG,
1726+
session: {
1727+
...TEST_CONFIG.session,
1728+
agentToAgent: { maxPingPongTurns: 0 },
1729+
},
1730+
},
1731+
}).find((candidate) => candidate.name === "sessions_send");
1732+
if (!tool) {
1733+
throw new Error("missing sessions_send tool");
1734+
}
1735+
1736+
const result = await tool.execute("call-run-scoped-caller", {
1737+
sessionKey: runScopedCallerKey,
1738+
message: "[TASK-COMPLETE] re-portal occupancy ready",
1739+
timeoutSeconds: 0,
16261740
});
1741+
1742+
const details = sessionsSendDetails(result.details);
1743+
expect(details.status).toBe("accepted");
1744+
expect(details.runId).toBe("durable-fallback-run");
1745+
expect(details.sessionKey).toBe(runScopedCallerKey);
1746+
expect(queueMessage).not.toHaveBeenCalled();
1747+
const agentCalls = calls.filter((call) => call.method === "agent");
1748+
expect(agentCalls).toHaveLength(1);
1749+
const params = agentParams(agentCalls[0] ?? {});
1750+
expect(params.sessionKey).toBe(durableCronCallerKey);
1751+
expect(params.message).toContain("[Inter-session message]");
1752+
expect(params.message).toContain("[TASK-COMPLETE] re-portal occupancy ready");
1753+
});
1754+
1755+
it("sessions_send rejects non-cron run-looking keys without durable-session fallback", async () => {
1756+
const calls: Array<{ method?: string; params?: unknown }> = [];
1757+
const runScopedCallerKey = "agent:leasing-ops:slack:channel:c-room:run:run-fast";
1758+
const queueMessage = vi.fn(async () => {});
1759+
setActiveEmbeddedRun(
1760+
"caller-active-session",
1761+
{
1762+
queueMessage,
1763+
isStreaming: () => false,
1764+
isCompacting: () => false,
1765+
supportsTranscriptCommitWait: true,
1766+
sourceReplyDeliveryMode: "message_tool_only",
1767+
abort: () => {},
1768+
},
1769+
runScopedCallerKey,
1770+
);
1771+
callGatewayMock.mockImplementation(async (opts: unknown) => {
1772+
const request = opts as { method?: string; params?: unknown };
1773+
calls.push(request);
1774+
if (request.method === "agent") {
1775+
return { runId: "durable-fallback-run", status: "accepted", acceptedAt: 2000 };
1776+
}
1777+
return {};
1778+
});
1779+
1780+
const tool = createOpenClawTools({
1781+
agentSessionKey: "agent:re-portal:main",
1782+
agentChannel: "telegram",
1783+
config: {
1784+
...TEST_CONFIG,
1785+
session: {
1786+
...TEST_CONFIG.session,
1787+
agentToAgent: { maxPingPongTurns: 0 },
1788+
},
1789+
},
1790+
}).find((candidate) => candidate.name === "sessions_send");
1791+
if (!tool) {
1792+
throw new Error("missing sessions_send tool");
1793+
}
1794+
1795+
const result = await tool.execute("call-run-scoped-caller", {
1796+
sessionKey: runScopedCallerKey,
1797+
message: "[TASK-COMPLETE] re-portal occupancy ready",
1798+
timeoutSeconds: 0,
1799+
});
1800+
1801+
const details = sessionsSendDetails(result.details);
1802+
expect(details.status).toBe("error");
1803+
expect(details.sessionKey).toBe(runScopedCallerKey);
1804+
expect(details.error).toContain("queue_message_failed reason=not_streaming");
1805+
expect(queueMessage).not.toHaveBeenCalled();
1806+
expect(calls.some((call) => call.method === "agent")).toBe(false);
16271807
});
16281808

16291809
it("sessions_send preserves active delivery when transcript commit wait is unsupported", async () => {
@@ -1677,7 +1857,7 @@ describe("sessions tools", () => {
16771857
expect(calls.some((call) => call.method === "agent")).toBe(false);
16781858
});
16791859

1680-
it("sessions_send reports run-scoped fallback admission failures", async () => {
1860+
it("sessions_send reports run-scoped queue admission failures without gateway fallback", async () => {
16811861
const runScopedCallerKey = "agent:leasing-ops:cron:monthly-utility:run:run-fast";
16821862
const queueMessage = vi.fn(async () => {
16831863
throw new Error("active session ended before queued steering message was committed");
@@ -1720,7 +1900,12 @@ describe("sessions tools", () => {
17201900
expect(details.status).toBe("error");
17211901
expect(details.sessionKey).toBe(runScopedCallerKey);
17221902
expect(details.error).toContain("queue_message_failed reason=runtime_rejected");
1723-
expect(details.error).toContain("fallback_failed error=gateway request timeout for agent");
1903+
expect(details.error).not.toContain("fallback_failed");
1904+
expect(
1905+
callGatewayMock.mock.calls.some(
1906+
(call) => (call[0] as { method?: string } | undefined)?.method === "agent",
1907+
),
1908+
).toBe(false);
17241909
});
17251910

17261911
it("sessions_send preserves terminal timeouts without starting A2A", async () => {

0 commit comments

Comments
 (0)